diff --git a/app/integrations/expose.py b/app/integrations/expose.py index d4f304d..16c5795 100644 --- a/app/integrations/expose.py +++ b/app/integrations/expose.py @@ -27,7 +27,7 @@ from __future__ import annotations import logging from dataclasses import dataclass, field -from datetime import timedelta +from datetime import UTC, datetime, timedelta from typing import Any, Callable, Optional, Protocol from sqlalchemy.orm import Session @@ -74,6 +74,17 @@ class DeviceInfo: their state is being published. """ + availability_id: Optional[str] = None + """Stable id used for the shared availability topic, when different from + this HA device's identity. A Meter is identified by its own UUID, while + its liveness comes from the source/channel feeding it. + """ + + availability_getter: Optional[Callable[["Session"], bool]] = field( + default=None, repr=False + ) + """Return whether the source behind this device is currently usable.""" + @dataclass class ExposableEntity: @@ -877,3 +888,323 @@ def _energy_cost_provider(session: Session) -> list[ExposableEntity]: # Register the energy cost provider at module load time. register_provider(_energy_cost_provider) + + +# --------------------------------------------------------------------------- +# M8 source / meter / thermal-cost provider +# --------------------------------------------------------------------------- + +_SOURCE_STALE_AFTER = timedelta(minutes=5) + + +def _utc_now() -> datetime: + """Small clock seam for live-value bounds and deterministic tests.""" + 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 _source_is_online(source: Any, channel: Any | None = None) -> bool: + """Do not turn an old cumulative value into a plausible live HA state.""" + now = _utc_now() + if not source.enabled or source.status != "online" or source.last_seen_at is None: + return False + source_age = now - _as_utc(source.last_seen_at) + if not timedelta(0) <= source_age <= _SOURCE_STALE_AFTER: + return False + if channel is None: + return True + if channel.latest_at is None or channel.latest_quality not in {"valid", "unverifiable"}: + return False + channel_age = now - _as_utc(channel.latest_at) + return timedelta(0) <= channel_age <= _SOURCE_STALE_AFTER + + +def _dsmr_latest(session: Session, source_id: int, *, start: datetime | None = None, + end: datetime | None = None, not_after: datetime | None = None) -> Any: + """Latest DSMR row in the source's (optionally bounded) cumulative domain.""" + from app.models.energy import DsmrReading + from sqlalchemy import select + + query = select(DsmrReading).where(DsmrReading.meter_source_id == source_id) + if start is not None: + query = query.where(DsmrReading.recorded_at >= start) + if end is not None: + query = query.where(DsmrReading.recorded_at < end) + if not_after is not None: + query = query.where(DsmrReading.recorded_at <= not_after) + return session.execute(query.order_by(DsmrReading.recorded_at.desc()).limit(1)).scalar_one_or_none() + + +def _dsmr_total(reading: Any) -> Any: + """Return imported electricity total from a real DSMR telegram, or None.""" + from decimal import Decimal, InvalidOperation + + try: + payload = reading.payload or {} + return Decimal(str(payload["electricity_delivered_1"])) + Decimal( + str(payload["electricity_delivered_2"]) + ) + except (InvalidOperation, KeyError, TypeError, ValueError): + return None + + +def _m8_energy_provider(session: Session) -> list[ExposableEntity]: + """Expose accepted source snapshots and current M8 meters. + + The provider intentionally reads the thermal service's public ``summarize`` + result for money. Keeping formulas in ``meter_cost`` prevents HA from + becoming a second, subtly different billing implementation. + """ + from app.models.energy import Meter + from app.models.meter_source import MeterSource, MeterSourceBinding, MeterSourceChannel + from sqlalchemy import select + + sources = session.execute(select(MeterSource)).scalars().all() + entities: list[ExposableEntity] = [] + + for source in sources: + source_info = DeviceInfo( + identifiers=("meter-source", source.uuid), name=source.name, + availability_id=source.uuid, + availability_getter=lambda sess, source_id=source.id: _source_online_by_id(sess, source_id), + ) + entities.append(ExposableEntity( + key=f"source.{source.uuid}.online", component="binary_sensor", device=source_info, + device_class="connectivity", unit="", name=f"{source.name} Online", + value_getter=lambda sess, source_id=source.id: "ON" if _source_online_by_id(sess, source_id) else "OFF", + )) + + active_meters = session.execute( + select(Meter).where( + Meter.ended_at.is_(None), Meter.commodity.in_(("electricity", "heating", "hot_water")) + ) + ).scalars().all() + active_by_commodity = {meter.commodity: meter for meter in active_meters} + for meter in active_meters: + bound = session.execute( + select(MeterSourceBinding, MeterSourceChannel, MeterSource) + .join(MeterSourceChannel, MeterSourceChannel.id == MeterSourceBinding.channel_id) + .join(MeterSource, MeterSource.id == MeterSourceChannel.source_id) + .where(MeterSourceBinding.meter_id == meter.id, MeterSourceBinding.ended_at.is_(None)) + ).one_or_none() + if bound is None: + continue + binding, channel, source = bound + info = DeviceInfo( + identifiers=("meter", meter.uuid), name=meter.label, + # Keep this opaque and Meter-anchored. In particular, do not use a + # source UUID here: two channels of one source can be independently + # stale/invalid and must not overwrite each other's availability. + availability_id=f"meter-availability-{meter.uuid}", + availability_getter=lambda sess, source_id=source.id, channel_id=channel.id: + _bound_channel_online(sess, source_id, channel_id), + ) + if meter.commodity == "electricity": + unit, device_class = "kWh", "energy" + elif meter.commodity == "heating": + unit, device_class = "GJ", "energy" + else: + unit, device_class = "m³", "volume" + for suffix, getter in ( + ("total", _meter_total_getter(binding.id, source.id, channel.id)), + ("today", _meter_today_getter(binding.id, source.id, channel.id)), + ): + entities.append(ExposableEntity( + key=f"meter.{meter.uuid}.{suffix}", component="sensor", device=info, + device_class=device_class, unit=unit, + name=f"{meter.label} {suffix.title()}", value_getter=getter, + state_class="total_increasing", + )) + + heating, hot_water = active_by_commodity.get("heating"), active_by_commodity.get("hot_water") + if heating is not None and hot_water is not None: + identity = ".".join(sorted((heating.uuid, hot_water.uuid))) + currency = _thermal_currency(session) + cost_info = DeviceInfo( + identifiers=("thermal-cost", identity), name="Thermal Energy Cost", + provides_availability=False, + ) + labels = { + "heating": "Heating", "hot_water_heating": "Hot Water Heating", "water": "Water", + "water_tax": "Water Tax", "fixed": "Fixed", "all_in": "All-in", + } + for suffix, window in (("total", None), ("today", "today")): + for metric, label in labels.items(): + entities.append(ExposableEntity( + key=f"thermal_cost.{identity}.{metric}_{suffix}", component="sensor", device=cost_info, + device_class="monetary", unit=currency, name=f"Thermal {label} {suffix.title()}", + value_getter=_thermal_cost_getter(metric, window), + state_class="total" if suffix == "total" else "total_increasing", + )) + return entities + + +def _source_online_by_id(session: Session, source_id: int) -> bool: + from app.models.meter_source import MeterSource + source = session.get(MeterSource, source_id) + if source is not None and source.kind == "dsmr_mqtt": + now = _utc_now() + # Inspect the actual latest telegram before calculating freshness: a + # clock-skewed future telegram must not make an older one look live. + latest = _dsmr_latest(session, source_id) + if not source.enabled or latest is None: + return False + age = now - _as_utc(latest.recorded_at) + return timedelta(0) <= age <= _SOURCE_STALE_AFTER + return source is not None and _source_is_online(source) + + +def _bound_channel_online(session: Session, source_id: int, channel_id: int) -> bool: + from app.models.meter_source import MeterSource, MeterSourceChannel + source, channel = session.get(MeterSource, source_id), session.get(MeterSourceChannel, channel_id) + if source is not None and source.kind == "dsmr_mqtt": + return _source_online_by_id(session, source_id) + return source is not None and channel is not None and _source_is_online(source, channel) + + +def _meter_total_getter(binding_id: int, source_id: int, channel_id: int) -> Callable[[Session], Any]: + def _getter(session: Session) -> Any: + from app.models.energy import Meter + from app.models.meter_source import MeterSource, MeterSourceBinding, MeterSourceChannel + if not _bound_channel_online(session, source_id, channel_id): + return None + binding = session.get(MeterSourceBinding, binding_id) + if ( + binding is None or binding.ended_at is not None or binding.channel_id != channel_id + or (meter := session.get(Meter, binding.meter_id)) is None or meter.ended_at is not None + ): + return None + source = session.get(MeterSource, source_id) + channel = session.get(MeterSourceChannel, channel_id) + if source is None or channel is None: + return None + start = max(_as_utc(meter.started_at), _as_utc(binding.started_at)) + # Both windows are half-open. The current records have no end, but + # retaining this form makes a future close fail safely. + end = min( + (value for value in (_as_utc(meter.ended_at) if meter.ended_at else None, + _as_utc(binding.ended_at) if binding.ended_at else None) if value is not None), + default=None, + ) + now = _utc_now() + if source.kind == "dsmr_mqtt": + latest = _dsmr_latest(session, source_id, start=start, end=end, not_after=now) + if latest is None: + return None + return _dsmr_total(latest) + if channel.latest_at is None or channel.latest_quality not in {"valid", "unverifiable"}: + return None + latest_at = _as_utc(channel.latest_at) + if latest_at < start or latest_at > now or (end is not None and latest_at >= end): + return None + return channel.latest_value + return _getter + + +def _meter_today_getter(binding_id: int, source_id: int, channel_id: int) -> Callable[[Session], Any]: + def _getter(session: Session) -> Any: + from app.models.energy import Meter + from app.models.meter_source import MeterSource, MeterSourceBinding, WarmteLinkReading + from app.services import timezone as tz + from sqlalchemy import select + binding = session.get(MeterSourceBinding, binding_id) + if binding is None or binding.ended_at is not None or binding.channel_id != channel_id: + return None + meter = session.get(Meter, binding.meter_id) + if meter is None or meter.ended_at is not None: + return None + now = _utc_now() + day = (tz.local_now() - _TODAY_RESET_GRACE).date() + start = max(tz.local_midnight_utc(day), _as_utc(meter.started_at), _as_utc(binding.started_at)) + bounds = [tz.local_midnight_utc(day + timedelta(days=1))] + bounds.extend(value for value in ( + _as_utc(meter.ended_at) if meter.ended_at else None, + _as_utc(binding.ended_at) if binding.ended_at else None, + ) if value is not None) + end = min(bounds) + source = session.get(MeterSource, source_id) + if source is None or not source.enabled: + return None + if source.kind != "dsmr_mqtt" and source.status != "online": + return None + if source.kind == "dsmr_mqtt": + first = _dsmr_latest(session, source_id, start=start, end=end, not_after=now) + if first is None: + return None + from app.models.energy import DsmrReading + from sqlalchemy import select as dsmr_select + rows = session.execute(dsmr_select(DsmrReading).where( + DsmrReading.meter_source_id == source_id, DsmrReading.recorded_at >= start, + DsmrReading.recorded_at < end, + DsmrReading.recorded_at <= now, + ).order_by(DsmrReading.recorded_at)).scalars().all() + if len(rows) < 2: + return None + first_total, last_total = _dsmr_total(rows[0]), _dsmr_total(rows[-1]) + if first_total is None or last_total is None: + return None + value = last_total - first_total + return value if value >= 0 else None + query = select(WarmteLinkReading).where( + WarmteLinkReading.channel_id == channel_id, + WarmteLinkReading.recorded_at >= start, + WarmteLinkReading.quality.in_(("valid", "unverifiable")), + ) + if end is not None: + query = query.where(WarmteLinkReading.recorded_at < end) + query = query.where(WarmteLinkReading.recorded_at <= now) + readings = session.execute(query.order_by(WarmteLinkReading.recorded_at)).scalars().all() + if len(readings) < 2: + return None + value = readings[-1].value - readings[0].value + return value if value >= 0 else None + return _getter + + +def _thermal_currency(session: Session) -> str: + from app.services.contracts import active_contract_versions + versions = active_contract_versions(session, scope="thermal") + return versions[-1].contract.currency if versions else "EUR" + + +def _thermal_cost_getter(metric: str, window: str | None) -> Callable[[Session], Any]: + def _getter(session: Session) -> Any: + from app.services import timezone as tz + from app.services.meter_cost import summarize + now = _utc_now() + from app.models.energy import Meter + meters = session.query(Meter).filter( + Meter.commodity.in_(("heating", "hot_water")), Meter.ended_at.is_(None) + ).all() + if len(meters) != 2: + return None + epoch_start = max(_as_utc(m.started_at) for m in meters) + if window == "today": + day = (tz.local_now() - _TODAY_RESET_GRACE).date() + start = max(tz.local_midnight_utc(day), epoch_start) + # During the reset grace ``day`` is yesterday, whose local midnight + # remains the cap; otherwise do not summarize readings from later today. + end = min(tz.local_midnight_utc(day + timedelta(days=1)), now) + else: + # A combined thermal identity begins when its newest constituent + # meter epoch begins; including pre-swap rows would mix identities. + start, end = epoch_start, now + result = summarize(session, start, end, now=now) + if result["period_count"] == 0 and result["fixed_cost"] == 0: + return None + if metric == "fixed": + return result["fixed_cost"] + if metric == "all_in": + return result["total_cost"] + if metric == "water": + return result["breakdown"]["hot_water"] + if metric == "water_tax": + return result["breakdown"]["hot_water_tax"] + return result["breakdown"][metric] + return _getter + + +register_provider(_m8_energy_provider) diff --git a/app/services/ha_discovery.py b/app/services/ha_discovery.py index 852f56a..3ce6e29 100644 --- a/app/services/ha_discovery.py +++ b/app/services/ha_discovery.py @@ -98,6 +98,15 @@ def _availability_topic(device_uuid: str, prefix: str) -> str: return f"{prefix}/modbus/{node}/availability" +def _availability_id(entity: ExposableEntity) -> str: + """Return the identity which owns this entity's liveness topic. + + M8 meters deliberately retain their own UUID as HA node/unique identity, + while their availability is supplied by a MeterSource UUID. + """ + return entity.device.availability_id or entity.device.identifiers[1] + + def _unique_id(entity: ExposableEntity) -> str: """Stable unique_id — device uuid + metric key (never from mutable fields).""" device_uuid = entity.device.identifiers[1] @@ -139,8 +148,7 @@ def build_discovery_payload( if state_prefix is None: state_prefix = discovery_prefix - device_uuid = entity.device.identifiers[1] - avail_topic = _availability_topic(device_uuid, state_prefix) + avail_topic = _availability_topic(_availability_id(entity), state_prefix) state_t = _state_topic(entity, state_prefix) topic = _discovery_topic(entity, discovery_prefix) @@ -212,6 +220,21 @@ def publish_discovery(session: Session) -> None: logger.exception("publish_discovery: failed to build catalog; aborting") return + # Meter UUIDs are intentionally identity-changing epochs. Discovery config + # is retained, so clear only the precisely enumerable old M8 identities; + # never wildcard a provider/topic and risk removing another source's card. + try: + stale_entities = _stale_m8_entities(session) + except Exception: + logger.exception("publish_discovery: unable to enumerate old M8 identities") + stale_entities = [] + for old_entity in stale_entities: + try: + old_topic, _ = build_discovery_payload(old_entity, discovery_prefix, state_prefix) + mqtt_manager.publish(old_topic, b"", retain=True) + except Exception: + logger.exception("publish_discovery: unable to clear old identity %r", old_entity.key) + for entry in catalog: entity = entry.entity try: @@ -236,6 +259,62 @@ def publish_discovery(session: Session) -> None: ) +def _stale_m8_entities(session: Session) -> list[ExposableEntity]: + """Return synthetic discovery entries for superseded thermal identities. + + This is deliberately a narrow, best-effort cleanup: ended Meter UUIDs and + historically possible thermal combinations only; current identities are excluded. + """ + from app.integrations.expose import DeviceInfo + from app.models.energy import Meter + from sqlalchemy import select + + meters = session.execute(select(Meter).where( + Meter.commodity.in_(("electricity", "heating", "hot_water")) + )).scalars().all() + current = {meter.commodity: meter for meter in meters if meter.ended_at is None} + old = [meter for meter in meters if meter.ended_at is not None] + entities: list[ExposableEntity] = [] + for meter in old: + info = DeviceInfo(identifiers=("meter", meter.uuid), name=meter.label) + for suffix in ("total", "today"): + entities.append(ExposableEntity( + key=f"meter.{meter.uuid}.{suffix}", component="sensor", device=info, + device_class=None, unit="", name="obsolete", + )) + heatings = [meter for meter in meters if meter.commodity == "heating"] + waters = [meter for meter in meters if meter.commodity == "hot_water"] + current_identity = ( + ".".join(sorted((current["heating"].uuid, current["hot_water"].uuid))) + if current.get("heating") is not None and current.get("hot_water") is not None else None + ) + for heating in heatings: + for water in waters: + if heating.ended_at is None and water.ended_at is None: + continue + # A thermal identity can only have been published when both Meter + # epochs were current at the same instant. Do not form a Cartesian + # product of historical records: that would tombstone identities + # which have never existed in HA. + heating_start, water_start = heating.started_at, water.started_at + heating_end, water_end = heating.ended_at, water.ended_at + if (heating_end is not None and water_start >= heating_end) or ( + water_end is not None and heating_start >= water_end + ): + continue + identity = ".".join(sorted((heating.uuid, water.uuid))) + if identity == current_identity: + continue + info = DeviceInfo(identifiers=("thermal-cost", identity), name="obsolete") + for metric in ("heating", "hot_water_heating", "water", "water_tax", "fixed", "all_in"): + for suffix in ("total", "today"): + entities.append(ExposableEntity( + key=f"thermal_cost.{identity}.{metric}_{suffix}", component="sensor", device=info, + device_class=None, unit="", name="obsolete", + )) + return entities + + # --------------------------------------------------------------------------- # Public: publish states # --------------------------------------------------------------------------- @@ -288,7 +367,19 @@ def _publish_entity_state( Also publishes the availability topic for ``binary_sensor`` "online" entities. """ state_t = _state_topic(entity, prefix) - device_uuid = entity.device.identifiers[1] + # Source-backed entities can have a different liveness identity from their + # HA device identity. Publish it before the state; a None value below is + # intentionally not converted to a synthetic zero. + if entity.device.provides_availability and entity.device.availability_getter is not None: + try: + available = bool(entity.device.availability_getter(session)) + mqtt_manager.publish( + _availability_topic(_availability_id(entity), prefix), + "online" if available else "offline", + retain=False, + ) + except Exception: + logger.exception("availability_getter raised for entity %r", entity.key) if entity.component == "binary_sensor" and "online" in entity.key: # The online sensor represents device availability. @@ -303,7 +394,7 @@ def _publish_entity_state( # Default to offline when no reading is available. online = (raw_value == "ON") avail_payload = "online" if online else "offline" - avail_topic = _availability_topic(device_uuid, prefix) + avail_topic = _availability_topic(_availability_id(entity), prefix) mqtt_manager.publish(avail_topic, avail_payload, retain=False) # The state of the binary_sensor itself state_payload = "ON" if online else "OFF" diff --git a/docs/design/m8-warmtelink-energy.md b/docs/design/m8-warmtelink-energy.md index 6de386f..8f33096 100644 --- a/docs/design/m8-warmtelink-energy.md +++ b/docs/design/m8-warmtelink-energy.md @@ -935,7 +935,7 @@ T01～T06 先把现有 DSMR 安全迁到统一 source/binding；T07～T11 再接 ### M8-T17 — Home Assistant Source / Meter / Thermal 实体 [structural] -- **Status**: `todo` +- **Status**: `done` - **Depends**: M8-T11, M8-T16 - **Context**: 在完整采集和成本链上扩展现有 expose provider，保持默认关闭与稳定 identity。 diff --git a/tests/test_energy_expose.py b/tests/test_energy_expose.py index 68b2069..1c926ff 100644 --- a/tests/test_energy_expose.py +++ b/tests/test_energy_expose.py @@ -28,10 +28,12 @@ Coverage: from __future__ import annotations -from datetime import datetime, timezone +from datetime import datetime, timedelta, timezone +from decimal import Decimal from pathlib import Path from typing import Any from unittest.mock import MagicMock, patch +from zoneinfo import ZoneInfo import pytest from alembic import command @@ -2715,3 +2717,237 @@ def test_midnight_state_publish_no_raise_when_mqtt_disabled() -> None: publish_states(sess) # must not raise mock_mgr.publish.assert_not_called() + + +# --------------------------------------------------------------------------- +# M8-T17 source / thermal HA catalog +# --------------------------------------------------------------------------- + + +def _make_thermal_source_and_meter(session: Session, commodity: str, now: datetime) -> Any: + from app.models.energy import Meter + from app.models.meter_source import MeterSource, MeterSourceBinding, MeterSourceChannel + + source = MeterSource( + name=f"{commodity} source", kind="warmtelink_serial", enabled=True, config={}, + status="online", last_seen_at=now, last_error=None, created_at=now, updated_at=now, + ) + session.add(source) + session.flush() + unit = {"electricity": "kWh", "heating": "GJ", "hot_water": "m³"}[commodity] + channel = MeterSourceChannel( + source_id=source.id, channel_key=f"accepted-{commodity}", label=commodity, unit=unit, + latest_value=Decimal("12.5"), latest_at=now, latest_quality="valid", + created_at=now, updated_at=now, + ) + meter = Meter(label=f"{commodity} meter", commodity=commodity, started_at=now - timedelta(days=1), + ended_at=None, reason="initial", note=None, created_at=now) + session.add_all((channel, meter)) + session.flush() + session.add(MeterSourceBinding(meter_id=meter.id, channel_id=channel.id, + started_at=meter.started_at, ended_at=None, + created_at=now, updated_at=now)) + session.flush() + return source, channel, meter + + +def test_m8_catalog_has_source_meter_and_thermal_entities_disabled(energy_db) -> None: + """D13: all new entities are catalogued with safe default toggles and units.""" + from app.integrations.expose import build_catalog + + now = datetime.now(timezone.utc) + with Session(energy_db) as session: + heating_source, _, heating = _make_thermal_source_and_meter(session, "heating", now) + _water_source, _, water = _make_thermal_source_and_meter(session, "hot_water", now) + _electricity_source, _, electricity = _make_thermal_source_and_meter(session, "electricity", now) + session.commit() + catalog = build_catalog(session) + + entries = {entry.entity.key: entry for entry in catalog} + assert entries[f"source.{heating_source.uuid}.online"].entity.device_class == "connectivity" + assert entries[f"meter.{heating.uuid}.total"].entity.unit == "GJ" + assert entries[f"meter.{heating.uuid}.total"].entity.device_class == "energy" + assert entries[f"meter.{water.uuid}.total"].entity.unit == "m³" + assert entries[f"meter.{water.uuid}.total"].entity.device_class == "volume" + assert entries[f"meter.{electricity.uuid}.total"].entity.unit == "kWh" + assert entries[f"meter.{electricity.uuid}.total"].entity.device_class == "energy" + assert entries[f"meter.{electricity.uuid}.today"].entity.state_class == "total_increasing" + thermal = [entry for key, entry in entries.items() if key.startswith("thermal_cost.")] + assert len(thermal) == 12 + assert all(entry.enabled is False for entry in thermal) + assert all(entry.entity.unit == "EUR" and entry.entity.device_class == "monetary" for entry in thermal) + + +def test_m8_meter_getter_hides_stale_or_offline_source(energy_db) -> None: + from app.integrations.expose import build_catalog + + now = datetime.now(timezone.utc) + with Session(energy_db) as session: + source, channel, meter = _make_thermal_source_and_meter(session, "heating", now) + session.commit() + entry = next(item for item in build_catalog(session) if item.entity.key == f"meter.{meter.uuid}.total") + assert entry.entity.value_getter(session) == channel.latest_value + source.status = "error" + assert entry.entity.value_getter(session) is None + source.status = "online" + source.last_seen_at = now - timedelta(minutes=6) + assert entry.entity.value_getter(session) is None + + +def test_m8_meter_today_uses_current_binding_and_never_invents_zero(energy_db) -> None: + """A binding is a half-open cumulative epoch, not merely a channel filter.""" + from app.integrations.expose import build_catalog + from app.models.meter_source import WarmteLinkReading + + now = datetime.now(timezone.utc) + with Session(energy_db) as session: + _source, channel, meter = _make_thermal_source_and_meter(session, "heating", now) + binding = meter.source_bindings[0] + binding.started_at = now - timedelta(minutes=20) + session.add_all(( + WarmteLinkReading(channel_id=channel.id, recorded_at=now - timedelta(minutes=30), + received_at=now, value=Decimal("100"), unit="GJ", quality="valid", + equipment_fingerprint="test"), + WarmteLinkReading(channel_id=channel.id, recorded_at=now - timedelta(minutes=15), + received_at=now, value=Decimal("110"), unit="GJ", quality="valid", + equipment_fingerprint="test"), + WarmteLinkReading(channel_id=channel.id, recorded_at=now - timedelta(minutes=5), + received_at=now, value=Decimal("115"), unit="GJ", quality="valid", + equipment_fingerprint="test"), + )) + session.commit() + entity = next(item.entity for item in build_catalog(session) + if item.entity.key == f"meter.{meter.uuid}.today") + assert entity.value_getter(session) == Decimal("5") + session.query(WarmteLinkReading).filter(WarmteLinkReading.channel_id == channel.id).delete() + session.add(WarmteLinkReading(channel_id=channel.id, recorded_at=now - timedelta(minutes=5), + received_at=now, value=Decimal("115"), unit="GJ", quality="valid", + equipment_fingerprint="test")) + session.flush() + assert entity.value_getter(session) is None + + +def test_m8_dsmr_electricity_meter_reads_real_telegram_domain(energy_db) -> None: + """DSMR does not populate channel latest fields: its telegram is authoritative.""" + from app.integrations.expose import build_catalog + from app.models.energy import DsmrReading + + now = datetime.now(timezone.utc) + with Session(energy_db) as session: + source, channel, meter = _make_thermal_source_and_meter(session, "electricity", now) + source.kind, source.status, source.last_seen_at = "dsmr_mqtt", "unknown", None + channel.latest_at, channel.latest_value, channel.latest_quality = None, None, None + session.add_all(( + DsmrReading(meter_source_id=source.id, recorded_at=now - timedelta(minutes=2), + payload={"electricity_delivered_1": "100", "electricity_delivered_2": "20"}), + DsmrReading(meter_source_id=source.id, recorded_at=now - timedelta(minutes=1), + payload={"electricity_delivered_1": "103", "electricity_delivered_2": "22"}), + )) + session.commit() + entries = {item.entity.key: item.entity for item in build_catalog(session)} + assert entries[f"source.{source.uuid}.online"].value_getter(session) == "ON" + assert entries[f"meter.{meter.uuid}.total"].value_getter(session) == Decimal("125") + assert entries[f"meter.{meter.uuid}.today"].value_getter(session) == Decimal("5") + + +def test_m8_warmtelink_future_snapshot_is_offline_but_today_uses_elapsed_points(energy_db) -> None: + """Future channel snapshots never become current HA state or today's delta.""" + from app.integrations.expose import build_catalog + from app.models.meter_source import WarmteLinkReading + from app.services import timezone as tz + + now = datetime(2026, 1, 15, 10, tzinfo=timezone.utc) + with Session(energy_db) as session: + source, channel, meter = _make_thermal_source_and_meter(session, "heating", now) + channel.latest_at, channel.latest_value = now + timedelta(minutes=1), Decimal("999") + session.add_all(( + WarmteLinkReading(channel_id=channel.id, recorded_at=now - timedelta(minutes=2), + received_at=now, value=Decimal("100"), unit="GJ", quality="valid", + equipment_fingerprint="test"), + WarmteLinkReading(channel_id=channel.id, recorded_at=now, + received_at=now, value=Decimal("110"), unit="GJ", quality="valid", + equipment_fingerprint="test"), + WarmteLinkReading(channel_id=channel.id, recorded_at=now + timedelta(minutes=1), + received_at=now, value=Decimal("999"), unit="GJ", quality="valid", + equipment_fingerprint="test"), + )) + session.commit() + with ( + patch("app.integrations.expose._utc_now", return_value=now), + patch.object(tz, "local_tz", return_value=ZoneInfo("Europe/Amsterdam")), + patch.object(tz, "local_now", return_value=now.astimezone(ZoneInfo("Europe/Amsterdam"))), + ): + entries = {item.entity.key: item.entity for item in build_catalog(session)} + assert entries[f"meter.{meter.uuid}.total"].device.availability_getter(session) is False + assert entries[f"meter.{meter.uuid}.total"].value_getter(session) is None + assert entries[f"meter.{meter.uuid}.today"].value_getter(session) == Decimal("10") + + +def test_m8_dsmr_future_telegram_is_offline_and_excluded_from_today(energy_db) -> None: + """DSMR's latest telegram is not fresh when it is in the future.""" + from app.integrations.expose import build_catalog + from app.models.energy import DsmrReading + from app.services import timezone as tz + + now = datetime(2026, 1, 15, 10, tzinfo=timezone.utc) + local_tz = ZoneInfo("Europe/Amsterdam") + next_midnight = datetime(2026, 1, 15, 23, tzinfo=timezone.utc) + with Session(energy_db) as session: + source, channel, meter = _make_thermal_source_and_meter(session, "electricity", now) + source.kind, source.status, source.last_seen_at = "dsmr_mqtt", "unknown", None + channel.latest_at, channel.latest_value, channel.latest_quality = None, None, None + session.add_all(( + DsmrReading(meter_source_id=source.id, recorded_at=now - timedelta(minutes=2), + payload={"electricity_delivered_1": "100", "electricity_delivered_2": "0"}), + DsmrReading(meter_source_id=source.id, recorded_at=now, + payload={"electricity_delivered_1": "110", "electricity_delivered_2": "0"}), + DsmrReading(meter_source_id=source.id, recorded_at=now + timedelta(minutes=1), + payload={"electricity_delivered_1": "999", "electricity_delivered_2": "0"}), + DsmrReading(meter_source_id=source.id, recorded_at=next_midnight, + payload={"electricity_delivered_1": "1000", "electricity_delivered_2": "0"}), + )) + session.commit() + with ( + patch("app.integrations.expose._utc_now", return_value=now), + patch.object(tz, "local_tz", return_value=local_tz), + patch.object(tz, "local_now", return_value=now.astimezone(local_tz)), + ): + entries = {item.entity.key: item.entity for item in build_catalog(session)} + assert entries[f"source.{source.uuid}.online"].value_getter(session) == "OFF" + assert entries[f"meter.{meter.uuid}.total"].value_getter(session) is None + assert entries[f"meter.{meter.uuid}.today"].value_getter(session) == Decimal("10") + + +def test_m8_thermal_today_summary_ends_at_frozen_now(energy_db) -> None: + """A running local day must not include thermal cost rows from its future.""" + from app.integrations.expose import build_catalog + from app.services import timezone as tz + + now = datetime(2026, 1, 15, 10, tzinfo=timezone.utc) + local_tz = ZoneInfo("Europe/Amsterdam") + captured: dict[str, datetime] = {} + result = { + "period_count": 1, "fixed_cost": Decimal("0"), "total_cost": Decimal("0"), + "breakdown": {key: Decimal("0") for key in ( + "heating", "hot_water_heating", "hot_water", "hot_water_tax", + )}, + } + + def summarize_spy(_session: Session, start: datetime, end: datetime, *, now: datetime) -> dict: + captured.update(start=start, end=end, now=now) + return result + + with Session(energy_db) as session: + _make_thermal_source_and_meter(session, "heating", now) + _make_thermal_source_and_meter(session, "hot_water", now) + session.commit() + with ( + patch("app.integrations.expose._utc_now", return_value=now), + patch.object(tz, "local_tz", return_value=local_tz), + patch.object(tz, "local_now", return_value=now.astimezone(local_tz)), + patch("app.services.meter_cost.summarize", side_effect=summarize_spy), + ): + entity = next(item.entity for item in build_catalog(session) + if item.entity.key.endswith(".heating_today")) + assert entity.value_getter(session) == Decimal("0") + assert captured["end"] == now diff --git a/tests/test_ha_discovery.py b/tests/test_ha_discovery.py index f8c9784..6ad8272 100644 --- a/tests/test_ha_discovery.py +++ b/tests/test_ha_discovery.py @@ -15,7 +15,7 @@ Coverage: from __future__ import annotations -from datetime import datetime, timezone +from datetime import datetime, timedelta, timezone from pathlib import Path from typing import Any from unittest.mock import MagicMock, patch @@ -1361,3 +1361,20 @@ def test_settings_payload_includes_ha_state_topic_prefix() -> None: assert payload["ha_state_topic_prefix"] == "my_prefix", ( f"Expected 'my_prefix', got {payload['ha_state_topic_prefix']!r}" ) + + +def test_stale_m8_entities_includes_ended_electricity_meter(disco_db) -> None: + """An electricity swap clears exactly the old Meter's two discovery configs.""" + from app.models.energy import Meter + from app.services.ha_discovery import _stale_m8_entities + + now = datetime.now(timezone.utc) + with Session(disco_db) as session: + old = Meter(label="old", commodity="electricity", started_at=now - timedelta(days=1), + ended_at=now, reason="meter_swap", note=None, created_at=now) + current = Meter(label="current", commodity="electricity", started_at=now, + ended_at=None, reason="meter_swap", note=None, created_at=now) + session.add_all((old, current)) + session.commit() + keys = {entity.key for entity in _stale_m8_entities(session)} + assert keys == {f"meter.{old.uuid}.total", f"meter.{old.uuid}.today"}