M8-T17: add Home Assistant thermal entities

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)

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"

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。
 

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

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"}