diff --git a/app/integrations/meter_sources.py b/app/integrations/meter_sources.py new file mode 100644 index 0000000..50bcb71 --- /dev/null +++ b/app/integrations/meter_sources.py @@ -0,0 +1,171 @@ +"""Registry and configuration helpers for meter-source integrations. + +The registry is deliberately I/O-free. Workers and HTTP handlers use these +helpers to share one config contract without opening a broker or serial port. +""" + +from __future__ import annotations + +from dataclasses import dataclass +from typing import Any + + +SECRET_MASK = "" + + +class SourceProfileError(ValueError): + """Raised when a source kind or its configuration is invalid.""" + + +@dataclass(frozen=True) +class SourceConfigField: + """One source configuration field and its public metadata.""" + + name: str + value_type: type + default: Any = None + required: bool = False + secret: bool = False + + +@dataclass(frozen=True) +class MeterSourceProfile: + """A supported source kind's config, capabilities, and channel units.""" + + kind: str + fields: tuple[SourceConfigField, ...] + capabilities: frozenset[str] + allowed_units: frozenset[str] + + +DSMR_MQTT_PROFILE = MeterSourceProfile( + kind="dsmr_mqtt", + fields=( + SourceConfigField("broker_host", str, default=""), + SourceConfigField("broker_port", int, default=1883), + SourceConfigField("username", str, default="", secret=True), + SourceConfigField("password", str, default="", secret=True), + SourceConfigField("tls_enabled", bool, default=False), + SourceConfigField("topic", str, default="dsmr/json"), + SourceConfigField("tariff_topic", str, default="dsmr/meter-stats/electricity_tariff"), + SourceConfigField("sample_interval_s", int, default=10), + ), + capabilities=frozenset({"discover", "mqtt_subscribe", "tariff"}), + allowed_units=frozenset({"kWh"}), +) + +WARMTELINK_SERIAL_PROFILE = MeterSourceProfile( + kind="warmtelink_serial", + fields=( + SourceConfigField("path", str, required=True), + SourceConfigField("baudrate", int, default=115200), + SourceConfigField("data_bits", int, default=7), + SourceConfigField("parity", str, default="N"), + SourceConfigField("stop_bits", int, default=1), + ), + capabilities=frozenset({"discover", "read_only_serial"}), + allowed_units=frozenset({"GJ", "m³"}), +) + +SOURCE_PROFILES: dict[str, MeterSourceProfile] = { + DSMR_MQTT_PROFILE.kind: DSMR_MQTT_PROFILE, + WARMTELINK_SERIAL_PROFILE.kind: WARMTELINK_SERIAL_PROFILE, +} + + +def get_source_profile(kind: str) -> MeterSourceProfile: + """Return the profile for *kind*, or raise a stable validation error.""" + try: + return SOURCE_PROFILES[kind] + except KeyError as exc: + raise SourceProfileError(f"Unsupported meter source kind: {kind!r}") from exc + + +def list_source_profiles() -> list[MeterSourceProfile]: + """Return profiles in deterministic kind order for a future API/UI.""" + return [SOURCE_PROFILES[kind] for kind in sorted(SOURCE_PROFILES)] + + +def _check_type(field: SourceConfigField, value: Any) -> None: + # bool is a subclass of int; accept it only for explicitly boolean fields. + if type(value) is not field.value_type: + raise SourceProfileError( + f"Config field {field.name!r} must be a {field.value_type.__name__}." + ) + + +def _validate_field_value(kind: str, field: SourceConfigField, value: Any) -> None: + _check_type(field, value) + if field.name == "path" and not value.startswith("/dev/"): + raise SourceProfileError("warmtelink_serial config path must start with '/dev/'.") + if field.name in {"broker_port", "sample_interval_s", "baudrate"} and value <= 0: + raise SourceProfileError(f"Config field {field.name!r} must be greater than zero.") + if field.name == "data_bits" and value != 7: + raise SourceProfileError("warmtelink_serial data_bits must be 7.") + if field.name == "parity" and value != "N": + raise SourceProfileError("warmtelink_serial parity must be 'N'.") + if field.name == "stop_bits" and value != 1: + raise SourceProfileError("warmtelink_serial stop_bits must be 1.") + + +def validate_source_config(kind: str, config: dict[str, Any]) -> dict[str, Any]: + """Validate a complete config and return it with profile defaults filled. + + Unknown keys are rejected to make configuration additions explicit. Secret + masking is intentionally not interpreted here: callers must merge a PATCH + with its stored config first. + """ + profile = get_source_profile(kind) + if not isinstance(config, dict): + raise SourceProfileError("Source config must be an object.") + fields = {field.name: field for field in profile.fields} + unknown = set(config) - set(fields) + if unknown: + raise SourceProfileError(f"Unknown {kind} config field(s): {sorted(unknown)!r}") + + validated: dict[str, Any] = {} + for field in profile.fields: + if field.name in config: + value = config[field.name] + elif field.required: + raise SourceProfileError(f"Missing required {kind} config field: {field.name!r}") + else: + value = field.default + _validate_field_value(kind, field, value) + validated[field.name] = value + return validated + + +def sanitize_source_config(kind: str, config: dict[str, Any]) -> dict[str, Any]: + """Validate and return a response-safe config with secrets masked.""" + profile = get_source_profile(kind) + sanitized = validate_source_config(kind, config) + for field in profile.fields: + if field.secret: + sanitized[field.name] = SECRET_MASK + return sanitized + + +def merge_source_config(kind: str, current: dict[str, Any], patch: dict[str, Any]) -> dict[str, Any]: + """Merge a partial PATCH into stored config, retaining masked secrets. + + An empty secret value is the public response mask and therefore means + "keep the old value". New sources use :func:`validate_source_config` + instead, so an explicitly empty secret can still be initially configured. + """ + profile = get_source_profile(kind) + current_validated = validate_source_config(kind, current) + if not isinstance(patch, dict): + raise SourceProfileError("Source config patch must be an object.") + fields = {field.name: field for field in profile.fields} + unknown = set(patch) - set(fields) + if unknown: + raise SourceProfileError(f"Unknown {kind} config field(s): {sorted(unknown)!r}") + + merged = dict(current_validated) + for name, value in patch.items(): + field = fields[name] + if field.secret and value == SECRET_MASK: + continue + merged[name] = value + return validate_source_config(kind, merged) diff --git a/app/services/meter_sources.py b/app/services/meter_sources.py new file mode 100644 index 0000000..99a8b83 --- /dev/null +++ b/app/services/meter_sources.py @@ -0,0 +1,353 @@ +"""Service layer for source/channel discovery and meter-source bindings. + +All mutating functions receive a caller-owned :class:`~sqlalchemy.orm.Session` +and never commit. This lets HTTP handlers compose source and meter changes in +one transaction later without exposing any connection I/O here. +""" + +from __future__ import annotations + +from datetime import UTC, datetime +from typing import Any + +from sqlalchemy import or_, select +from sqlalchemy.orm import Session + +from app.integrations.meter_sources import ( + SourceProfileError, + get_source_profile, + merge_source_config, + validate_source_config, +) +from app.models.energy import Meter +from app.models.meter_source import ( + MeterSource, + MeterSourceBinding, + MeterSourceChannel, + half_open_intervals_overlap, +) + + +class MeterSourceError(ValueError): + """Base class for source-domain validation errors.""" + + +class SourceNotFoundError(MeterSourceError): + """Raised when the requested source does not exist.""" + + +class ChannelNotFoundError(MeterSourceError): + """Raised when the requested source channel does not exist.""" + + +class MeterNotFoundError(MeterSourceError): + """Raised when the requested meter does not exist.""" + + +class BindingNotFoundError(MeterSourceError): + """Raised when the requested binding does not exist.""" + + +class BindingValidationError(MeterSourceError): + """Raised for an incompatible unit, commodity, or invalid interval.""" + + +class BindingOverlapError(BindingValidationError): + """Raised when a meter or channel already has an overlapping binding.""" + + +class SourceDeleteRestrictedError(MeterSourceError): + """Raised when a source has retained channel, binding, or reading history.""" + + +COMMODITY_UNITS = {"electricity": "kWh", "heating": "GJ", "hot_water": "m³"} +_UNSET = object() + + +def _utc_now() -> datetime: + return datetime.now(UTC) + + +def _as_utc(value: datetime) -> datetime: + return value.replace(tzinfo=UTC) if value.tzinfo is None else value.astimezone(UTC) + + +def get_source(session: Session, source_id: int) -> MeterSource: + source = session.get(MeterSource, source_id) + if source is None: + raise SourceNotFoundError(f"Meter source {source_id} was not found.") + return source + + +def list_sources(session: Session, *, kind: str | None = None) -> list[MeterSource]: + statement = select(MeterSource).order_by(MeterSource.id) + if kind is not None: + get_source_profile(kind) + statement = statement.where(MeterSource.kind == kind) + return list(session.execute(statement).scalars()) + + +def create_source( + session: Session, + *, + name: str, + kind: str, + config: dict[str, Any], + enabled: bool = True, +) -> MeterSource: + """Add a source after validating its complete kind-specific config.""" + now = _utc_now() + source = MeterSource( + name=name, + kind=kind, + enabled=enabled, + config=validate_source_config(kind, config), + created_at=now, + updated_at=now, + ) + session.add(source) + return source + + +def update_source( + session: Session, + source_id: int, + *, + name: str | None = None, + enabled: bool | None = None, + config_patch: dict[str, Any] | None = None, +) -> MeterSource: + """Update source metadata and/or merge a partial source config without commit.""" + source = get_source(session, source_id) + if name is not None: + source.name = name + if enabled is not None: + source.enabled = enabled + if config_patch is not None: + source.config = merge_source_config(source.kind, source.config, config_patch) + source.updated_at = _utc_now() + return source + + +def delete_source(session: Session, source_id: int) -> None: + """Delete an entirely unused source; history is always retained instead.""" + source = get_source(session, source_id) + has_channel = session.execute( + select(MeterSourceChannel.id).where(MeterSourceChannel.source_id == source.id).limit(1) + ).scalar_one_or_none() + has_binding = session.execute( + select(MeterSourceBinding.id) + .join(MeterSourceChannel) + .where(MeterSourceChannel.source_id == source.id) + .limit(1) + ).scalar_one_or_none() + if has_channel is not None or has_binding is not None: + raise SourceDeleteRestrictedError( + f"Meter source {source_id} has dependent channels, bindings, or readings." + ) + session.delete(source) + + +def get_channel(session: Session, channel_id: int) -> MeterSourceChannel: + channel = session.get(MeterSourceChannel, channel_id) + if channel is None: + raise ChannelNotFoundError(f"Meter source channel {channel_id} was not found.") + return channel + + +def upsert_discovered_channel( + session: Session, + *, + source_id: int, + channel_key: str, + label: str, + unit: str, + suggested_commodity: str | None = None, + device_type: str | None = None, + fingerprint: str | None = None, + latest_value: Any = None, + latest_at: datetime | None = None, + latest_quality: str | None = None, +) -> MeterSourceChannel: + """Idempotently create or refresh a discovered channel's metadata. + + ``suggested_commodity`` remains metadata only; this function never creates + a meter or a binding. + """ + source = get_source(session, source_id) + if unit not in get_source_profile(source.kind).allowed_units: + raise SourceProfileError(f"Unit {unit!r} is not allowed for source kind {source.kind!r}.") + channel = session.execute( + select(MeterSourceChannel).where( + MeterSourceChannel.source_id == source.id, + MeterSourceChannel.channel_key == channel_key, + ) + ).scalar_one_or_none() + now = _utc_now() + if channel is None: + channel = MeterSourceChannel( + source_id=source.id, + channel_key=channel_key, + label=label, + unit=unit, + suggested_commodity=suggested_commodity, + device_type=device_type, + fingerprint=fingerprint, + latest_value=latest_value, + latest_at=latest_at, + latest_quality=latest_quality, + created_at=now, + updated_at=now, + ) + session.add(channel) + return channel + + if channel.unit != unit: + binding_id = session.execute( + select(MeterSourceBinding.id) + .where(MeterSourceBinding.channel_id == channel.id) + .limit(1) + ).scalar_one_or_none() + if binding_id is not None: + raise BindingValidationError( + f"Cannot change unit of bound channel {channel.id} from {channel.unit!r} to {unit!r}." + ) + + channel.label = label + channel.unit = unit + channel.suggested_commodity = suggested_commodity + channel.device_type = device_type + channel.fingerprint = fingerprint + channel.latest_value = latest_value + channel.latest_at = latest_at + channel.latest_quality = latest_quality + channel.updated_at = now + return channel + + +def list_bindings( + session: Session, *, meter_id: int | None = None, channel_id: int | None = None +) -> list[MeterSourceBinding]: + statement = select(MeterSourceBinding).order_by(MeterSourceBinding.started_at, MeterSourceBinding.id) + if meter_id is not None: + statement = statement.where(MeterSourceBinding.meter_id == meter_id) + if channel_id is not None: + statement = statement.where(MeterSourceBinding.channel_id == channel_id) + return list(session.execute(statement).scalars()) + + +def _get_meter(session: Session, meter_id: int) -> Meter: + meter = session.get(Meter, meter_id) + if meter is None: + raise MeterNotFoundError(f"Meter {meter_id} was not found.") + return meter + + +def _validate_binding( + session: Session, + *, + meter_id: int, + channel_id: int, + started_at: datetime, + ended_at: datetime | None, + excluding_id: int | None = None, +) -> None: + meter = _get_meter(session, meter_id) + channel = get_channel(session, channel_id) + expected_unit = COMMODITY_UNITS.get(meter.commodity) + if expected_unit is None: + raise BindingValidationError(f"Commodity {meter.commodity!r} cannot be bound to a source channel.") + if channel.unit != expected_unit: + raise BindingValidationError( + f"Meter commodity {meter.commodity!r} requires unit {expected_unit!r}, " + f"but channel has {channel.unit!r}." + ) + if ended_at is not None and _as_utc(ended_at) <= _as_utc(started_at): + raise BindingValidationError("Binding ended_at must be strictly after started_at.") + + candidates = session.execute( + select(MeterSourceBinding).where( + or_( + MeterSourceBinding.meter_id == meter_id, + MeterSourceBinding.channel_id == channel_id, + ) + ) + ).scalars() + for existing in candidates: + if existing.id == excluding_id: + continue + if half_open_intervals_overlap( + _as_utc(started_at), + _as_utc(ended_at) if ended_at is not None else None, + _as_utc(existing.started_at), + _as_utc(existing.ended_at) if existing.ended_at is not None else None, + ): + side = "meter" if existing.meter_id == meter_id else "channel" + raise BindingOverlapError(f"Binding overlaps existing {side} binding {existing.id}.") + + +def create_binding( + session: Session, + *, + meter_id: int, + channel_id: int, + started_at: datetime, + ended_at: datetime | None = None, +) -> MeterSourceBinding: + """Create a compatible non-overlapping half-open source binding.""" + _validate_binding( + session, + meter_id=meter_id, + channel_id=channel_id, + started_at=started_at, + ended_at=ended_at, + ) + now = _utc_now() + binding = MeterSourceBinding( + meter_id=meter_id, + channel_id=channel_id, + started_at=started_at, + ended_at=ended_at, + created_at=now, + updated_at=now, + ) + session.add(binding) + return binding + + +def update_binding( + session: Session, + binding_id: int, + *, + meter_id: int | None = None, + channel_id: int | None = None, + started_at: datetime | None = None, + ended_at: datetime | None | object = _UNSET, +) -> MeterSourceBinding: + """Correct a binding while preserving half-open timeline constraints.""" + binding = session.get(MeterSourceBinding, binding_id) + if binding is None: + raise BindingNotFoundError(f"Meter source binding {binding_id} was not found.") + new_meter_id = binding.meter_id if meter_id is None else meter_id + new_channel_id = binding.channel_id if channel_id is None else channel_id + new_started_at = binding.started_at if started_at is None else started_at + new_ended_at = binding.ended_at if ended_at is _UNSET else ended_at + _validate_binding( + session, + meter_id=new_meter_id, + channel_id=new_channel_id, + started_at=new_started_at, + ended_at=new_ended_at, + excluding_id=binding.id, + ) + binding.meter_id = new_meter_id + binding.channel_id = new_channel_id + binding.started_at = new_started_at + binding.ended_at = new_ended_at + binding.updated_at = _utc_now() + return binding + + +def close_binding(session: Session, binding_id: int, *, ended_at: datetime) -> MeterSourceBinding: + """Close an existing binding at its exclusive end boundary.""" + return update_binding(session, binding_id, ended_at=ended_at) diff --git a/docs/design/m8-warmtelink-energy.md b/docs/design/m8-warmtelink-energy.md index 71cc67c..548a561 100644 --- a/docs/design/m8-warmtelink-energy.md +++ b/docs/design/m8-warmtelink-energy.md @@ -328,7 +328,7 @@ T01～T06 先把现有 DSMR 安全迁到统一 source/binding；T07～T11 再接 ### M8-T02 — Source profile registry 与 binding service -- **Status**: `todo` +- **Status**: `done` - **Depends**: M8-T01 - **Context**: 在开放 API 前集中 kind config、commodity/unit 兼容与时间线规则，避免各入口各写一套。 diff --git a/tests/test_meter_source_services.py b/tests/test_meter_source_services.py new file mode 100644 index 0000000..25be824 --- /dev/null +++ b/tests/test_meter_source_services.py @@ -0,0 +1,184 @@ +"""M8-T02 tests for source profiles and transaction-owned binding services.""" + +from __future__ import annotations + +from datetime import UTC, datetime, timedelta + +import pytest +from sqlalchemy import create_engine, event +from sqlalchemy.orm import Session + +from app.integrations.meter_sources import ( + SECRET_MASK, + SourceProfileError, + merge_source_config, + sanitize_source_config, + validate_source_config, +) +from app.models.energy import Meter +from app.models.meter_source import MeterSource, MeterSourceBinding, MeterSourceChannel +from app.services.meter_sources import ( + BindingOverlapError, + BindingValidationError, + SourceDeleteRestrictedError, + close_binding, + create_binding, + create_source, + delete_source, + upsert_discovered_channel, +) + + +@pytest.mark.parametrize( + ("kind", "config", "expected"), + [ + ("dsmr_mqtt", {}, {"broker_port": 1883, "topic": "dsmr/json", "sample_interval_s": 10}), + ( + "warmtelink_serial", + {"path": "/dev/warmtelink"}, + {"baudrate": 115200, "data_bits": 7, "parity": "N", "stop_bits": 1}, + ), + ], +) +def test_profiles_fill_defaults(kind, config, expected): + validated = validate_source_config(kind, config) + assert {key: validated[key] for key in expected} == expected + + +@pytest.mark.parametrize( + ("kind", "config"), + [ + ("warmtelink_serial", {"path": "/tmp/warmtelink"}), + ("warmtelink_serial", {"path": "/dev/warmtelink", "data_bits": 8}), + ("dsmr_mqtt", {"extra": True}), + ], +) +def test_profile_rejects_invalid_or_unknown_config(kind, config): + with pytest.raises(SourceProfileError): + validate_source_config(kind, config) + + +def test_secret_sanitize_and_mask_merge_keep_old_value(): + original = validate_source_config("dsmr_mqtt", {"password": "not-for-response"}) + assert sanitize_source_config("dsmr_mqtt", original)["password"] == SECRET_MASK + merged = merge_source_config("dsmr_mqtt", original, {"password": SECRET_MASK, "topic": "new/topic"}) + assert merged["password"] == "not-for-response" + assert merged["topic"] == "new/topic" + + +@pytest.fixture() +def session(tmp_path): + engine = create_engine(f"sqlite:///{tmp_path / 'source_services.db'}") + + @event.listens_for(engine, "connect") + def _enable_foreign_keys(dbapi_connection, _connection_record): + dbapi_connection.execute("PRAGMA foreign_keys=ON") + + from app.db import Base + + Base.metadata.create_all(engine) + with Session(engine) as db_session: + yield db_session + engine.dispose() + + +def _meter(session: Session, commodity: str, label: str = "Meter") -> Meter: + timestamp = datetime(2026, 8, 22, tzinfo=UTC) + meter = Meter( + label=label, + commodity=commodity, + started_at=timestamp, + reason="initial", + created_at=timestamp, + ) + session.add(meter) + session.flush() + return meter + + +def _source_and_channel(session: Session, kind: str, unit: str) -> tuple[MeterSource, MeterSourceChannel]: + config = {"path": "/dev/warmtelink"} if kind == "warmtelink_serial" else {} + source = create_source(session, name="Source", kind=kind, config=config) + session.flush() + channel = upsert_discovered_channel( + session, + source_id=source.id, + channel_key=f"{unit}-total", + label="Total", + unit=unit, + suggested_commodity="heating", + ) + session.flush() + return source, channel + + +def test_channel_upsert_is_idempotent_and_never_auto_binds(session): + _, channel = _source_and_channel(session, "warmtelink_serial", "GJ") + same = upsert_discovered_channel( + session, + source_id=channel.source_id, + channel_key=channel.channel_key, + label="Renamed total", + unit="GJ", + suggested_commodity="heating", + ) + session.flush() + assert same.id == channel.id + assert same.label == "Renamed total" + assert session.query(MeterSourceBinding).count() == 0 + + +def test_channel_upsert_rejects_unit_change_when_channel_is_bound(session): + meter = _meter(session, "heating") + _, channel = _source_and_channel(session, "warmtelink_serial", "GJ") + create_binding(session, meter_id=meter.id, channel_id=channel.id, started_at=meter.started_at) + session.flush() + + with pytest.raises(BindingValidationError, match="Cannot change unit of bound channel"): + upsert_discovered_channel( + session, + source_id=channel.source_id, + channel_key=channel.channel_key, + label="Total", + unit="m³", + suggested_commodity="hot_water", + ) + + assert channel.unit == "GJ" + + +def test_binding_checks_both_sides_and_allows_equal_boundaries(session): + meter_one = _meter(session, "heating", "one") + meter_two = _meter(session, "heating", "two") + _, channel_one = _source_and_channel(session, "warmtelink_serial", "GJ") + _, channel_two = _source_and_channel(session, "warmtelink_serial", "GJ") + start = datetime(2026, 8, 22, tzinfo=UTC) + boundary = start + timedelta(hours=1) + create_binding(session, meter_id=meter_one.id, channel_id=channel_one.id, started_at=start, ended_at=boundary) + create_binding(session, meter_id=meter_one.id, channel_id=channel_two.id, started_at=boundary) + create_binding(session, meter_id=meter_two.id, channel_id=channel_one.id, started_at=boundary) + with pytest.raises(BindingOverlapError): + create_binding(session, meter_id=meter_one.id, channel_id=channel_two.id, started_at=start + timedelta(minutes=30)) + with pytest.raises(BindingOverlapError): + create_binding(session, meter_id=meter_two.id, channel_id=channel_one.id, started_at=start + timedelta(minutes=30)) + + +def test_binding_rejects_incompatible_unit_and_close_keeps_transaction_open(session): + meter = _meter(session, "electricity") + _, channel = _source_and_channel(session, "warmtelink_serial", "GJ") + with pytest.raises(BindingValidationError): + create_binding(session, meter_id=meter.id, channel_id=channel.id, started_at=meter.started_at) + + thermal_meter = _meter(session, "heating", "thermal") + binding = create_binding(session, meter_id=thermal_meter.id, channel_id=channel.id, started_at=meter.started_at) + session.flush() + close_binding(session, binding.id, ended_at=meter.started_at + timedelta(minutes=1)) + assert session.in_transaction() + session.rollback() + assert session.get(MeterSourceBinding, binding.id) is None + + +def test_source_delete_is_restricted_by_discovered_channel(session): + source, _ = _source_and_channel(session, "dsmr_mqtt", "kWh") + with pytest.raises(SourceDeleteRestrictedError): + delete_source(session, source.id)