"""Expose framework: ExposableEntity, provider protocol, and build_catalog. This module defines: - ``ExposableEntity`` — a dataclass describing one publishable MQTT/HA entity. - Provider protocol — a simple callable ``(session) -> list[ExposableEntity]``. - A provider registry and registration helper. - ``build_catalog(session)`` — merges all registered providers' entities with per-key toggle state from the ``exposed_entity_toggle`` table. Design notes ------------ - **No over-engineering**: providers are plain callables, not abstract base classes. This project is personal / single-user; keep it flat. - **Key stability**: entity keys MUST be derived from device uuid + metric key (e.g. ``"modbus..voltage"``), never from auto-increment IDs. Stable keys mean the toggle table survives device removal/re-addition without drift. - **Metadata from profile**: the modbus provider reads ``device_class``, ``unit``, and ``component`` directly from the YAML profile's ``metrics[]`` rather than maintaining a separate mapping. - **value_getter**: a callable ``() -> object | None`` or ``None`` if not yet implemented (T11 will wire real getters). Presence of the field lets T11 call it without changing the dataclass interface. """ from __future__ import annotations import logging from dataclasses import dataclass, field from typing import Any, Callable, Optional, Protocol from sqlalchemy.orm import Session logger = logging.getLogger(__name__) # --------------------------------------------------------------------------- # ExposableEntity # --------------------------------------------------------------------------- @dataclass class DeviceInfo: """Grouping metadata that maps an entity to a logical HA device. ``identifiers`` corresponds to ``device.identifiers`` in HA Discovery config — a stable tuple used to group entities under one device card. ``name`` is the human-readable device name (friendly_name). """ identifiers: tuple[str, ...] """Stable identifiers for HA device grouping (e.g. ``("modbus", "")``). These must not change over the device lifetime; they anchor the HA device card even when friendly_name changes. """ name: str """Human-readable device name (may change; triggers HA discovery re-publish).""" provides_availability: bool = True """Whether this device publishes an availability ("online"/"offline") heartbeat. When True (the default — e.g. Modbus devices, which have an ``online`` binary_sensor driving availability), the HA Discovery config for each entity declares an ``availability`` topic and HA marks the entity unavailable until "online" is published. When False (e.g. the energy-cost device, which has only sensors and no heartbeat source), the config omits ``availability`` so HA treats the entity as *always available* and shows its state as soon as one arrives. Without this, such entities would stay perpetually ``unavailable`` in HA even though their state is being published. """ @dataclass class ExposableEntity: """One publishable entity in the MQTT / HA Discovery expose framework. Fields ------ key Stable unique identifier for this entity. Used as the toggle table key and as part of the HA Discovery ``unique_id`` derivation. Convention: ``".."``, e.g. ``"modbus.abc123.voltage"`` or ``"modbus.abc123.online"``. component HA MQTT component type: ``"sensor"``, ``"binary_sensor"``, ``"switch"``, etc. device Grouping info (DeviceInfo) that determines which HA device card this entity belongs to. device_class HA ``device_class`` string (e.g. ``"voltage"``, ``"power"``, ``"None"``). Empty string or ``None`` means no device_class (e.g. for the online sensor). unit Physical unit string (e.g. ``"V"``, ``"A"``, ``"kWh"``). Empty string for dimensionless. name Human-readable entity label (friendly display name, may include device context, e.g. ``"SDM120 Voltage"``). value_getter Optional callable ``(session) -> object | None`` that returns the current entity value given an open SQLAlchemy session. ``None`` means "not yet wired". T11 calls this to obtain the current state before publishing. Signature: ``Callable[[Session], Any]``. state_class HA ``state_class`` string (e.g. ``"measurement"``, ``"total_increasing"``). ``None`` means not set. """ key: str component: str device: DeviceInfo device_class: Optional[str] unit: str name: str value_getter: Optional[Callable[["Session"], Any]] = field(default=None, repr=False) state_class: Optional[str] = None # --------------------------------------------------------------------------- # Provider protocol # --------------------------------------------------------------------------- class EntityProvider(Protocol): """Protocol for entity provider callables. A provider is any callable that accepts a ``Session`` and returns a list of ``ExposableEntity`` objects. Using a Protocol (not ABC) keeps the implementation lightweight — any matching callable qualifies. """ def __call__(self, session: Session) -> list[ExposableEntity]: ... # --------------------------------------------------------------------------- # Provider registry # --------------------------------------------------------------------------- _REGISTRY: list[EntityProvider] = [] def register_provider(provider: EntityProvider) -> EntityProvider: """Register an entity provider. Can be used as a decorator:: @register_provider def my_provider(session: Session) -> list[ExposableEntity]: ... Or called directly:: register_provider(my_provider) Returns the provider unchanged (for decorator compatibility). """ _REGISTRY.append(provider) return provider def get_providers() -> list[EntityProvider]: """Return a snapshot of the currently registered providers.""" return list(_REGISTRY) # --------------------------------------------------------------------------- # Catalog builder # --------------------------------------------------------------------------- @dataclass class CatalogEntry: """An entity from the catalog, enriched with its current toggle state.""" entity: ExposableEntity enabled: bool """True if this entity is currently enabled in the toggle table.""" def build_catalog(session: Session) -> list[CatalogEntry]: """Enumerate all entities from registered providers and attach toggle state. For each entity key, the toggle state is looked up in ``exposed_entity_toggle``. Entities with no toggle row default to ``enabled=False``. Parameters ---------- session: Active SQLAlchemy session for DB access. Returns ------- list[CatalogEntry] All entities from all registered providers, each paired with its current ``enabled`` state. """ from app.models.expose import ExposedEntityToggle # local import to avoid circular # Collect all entities from all providers. all_entities: list[ExposableEntity] = [] for provider in _REGISTRY: try: entities = provider(session) all_entities.extend(entities) except Exception: logger.exception("Provider %r raised an error; skipping its entities", provider) if not all_entities: return [] # Load all toggle rows in one query for efficiency. keys = [e.key for e in all_entities] toggle_rows = session.query(ExposedEntityToggle).filter( ExposedEntityToggle.key.in_(keys) ).all() toggle_map: dict[str, bool] = {row.key: row.enabled for row in toggle_rows} return [ CatalogEntry(entity=entity, enabled=toggle_map.get(entity.key, False)) for entity in all_entities ] # --------------------------------------------------------------------------- # Modbus provider # --------------------------------------------------------------------------- def _modbus_provider(session: Session) -> list[ExposableEntity]: """Enumerate ExposableEntity objects for all enabled Modbus devices. For each enabled ``ModbusDevice``: - Loads its YAML profile. - Produces one ``sensor`` entity per metric in ``profile.metrics`` (device_class / unit / component taken from the metric spec). - Produces one ``binary_sensor`` entity named "online" (derived from ``device.last_poll_ok``). Entity keys follow the pattern ``"modbus.."``, which is stable across device renames and DB rebuilds. """ from app.models.modbus import ModbusDevice # local import from app.integrations.modbus.profiles import ( load_profile, ProfileNotFoundError, ProfileValidationError, ) devices: list[ModbusDevice] = session.query(ModbusDevice).filter( ModbusDevice.enabled.is_(True) ).all() entities: list[ExposableEntity] = [] for device in devices: device_info = DeviceInfo( identifiers=("modbus", device.uuid), name=device.friendly_name, ) # Load the profile to get metric metadata. try: profile = load_profile(device.profile) except (ProfileNotFoundError, ProfileValidationError) as exc: logger.warning( "Skipping device %r (uuid=%s): cannot load profile %r: %s", device.friendly_name, device.uuid, device.profile, exc, ) continue # One sensor entity per profile metric. for metric in profile.metrics: entity_key = f"modbus.{device.uuid}.{metric.key}" # Capture device id and metric key for the value_getter closure. # The getter queries the most recent ModbusReading for this device # and extracts the metric value from its payload. _device_id = device.id _metric_key = metric.key def _make_value_getter( dev_id: int, m_key: str ) -> Callable[["Session"], Any]: """Return a getter that fetches the latest reading value from DB.""" def _getter(sess: "Session") -> Any: from app.models.modbus import ModbusReading from sqlalchemy import desc reading = ( sess.query(ModbusReading) .filter(ModbusReading.device_id == dev_id) .order_by(desc(ModbusReading.recorded_at)) .first() ) if reading is None: return None return reading.payload.get(m_key) return _getter entities.append( ExposableEntity( key=entity_key, component=metric.ha_component, device=device_info, device_class=metric.device_class or None, unit=metric.unit, name=f"{device.friendly_name} {metric.key.replace('_', ' ').title()}", value_getter=_make_value_getter(_device_id, _metric_key), state_class=metric.state_class, ) ) # One binary_sensor entity for device "online" status. online_key = f"modbus.{device.uuid}.online" _dev_id_online = device.id def _make_online_getter(dev_id: int) -> Callable[["Session"], Any]: """Return a getter that reports the device online status from DB.""" def _getter(sess: "Session") -> Any: from app.models.modbus import ModbusDevice dev = sess.query(ModbusDevice).filter( ModbusDevice.id == dev_id ).first() if dev is None: return None # Map last_poll_ok to HA binary_sensor payload strings. if dev.last_poll_ok is True: return "ON" return "OFF" return _getter entities.append( ExposableEntity( key=online_key, component="binary_sensor", device=device_info, device_class="connectivity", unit="", name=f"{device.friendly_name} Online", value_getter=_make_online_getter(_dev_id_online), state_class=None, ) ) return entities # Register the modbus provider at module load time. register_provider(_modbus_provider) # --------------------------------------------------------------------------- # Energy Cost provider # --------------------------------------------------------------------------- def _energy_cost_provider(session: Session) -> list[ExposableEntity]: """Enumerate ExposableEntity objects for the energy cost subsystem. Produces 4 sensor entities grouped under a single HA device "Energy Cost": - ``buy_price_now`` — current effective buy price (EUR/kWh or local currency). - ``sell_price_now`` — current effective sell price (EUR/kWh or local currency). - ``import_cost_total`` — cumulative import cost (total_increasing, monetary). - ``export_revenue_total`` — cumulative export revenue (total_increasing, monetary). Current-price algorithm (source-agnostic, with fallback) --------------------------------------------------------- Strategy A: read the ``pricing`` snapshot from the most recent non-degraded ``energy_cost_period`` row. - For ``kind="tibber"``: the snapshot contains ``"buy"`` and ``"sell"`` keys (per-unit prices in the contract currency). - For ``kind="manual"``: ``"buy_normal"`` and ``"sell_normal"`` are used as representative effective per-unit prices. (Both tariff-slot prices differ only by the base rate; energy_tax and ODE are the same for both, so buy_normal is the higher/conservative single representative value.) When no non-degraded period exists or the pricing snapshot lacks the expected keys, ``value_getter`` returns ``None`` (``publish_states`` skips None values automatically). Cumulative totals ----------------- ``SUM(import_cost)`` and ``SUM(export_revenue)`` over **all non-degraded** ``energy_cost_period`` rows. Degraded rows carry 0 costs and are excluded to avoid double-counting when they are later overwritten by real values. Currency -------- Taken from the most recent non-degraded period's ``currency`` column. Falls back to ``"EUR"`` when no such row exists. Key convention -------------- Fixed stable string keys (not derived from any mutable field or DB id): - ``"energy.buy_price_now"`` - ``"energy.sell_price_now"`` - ``"energy.import_cost_total"`` - ``"energy.export_revenue_total"`` DeviceInfo identifiers ---------------------- **Two-element tuple** ``("energy-cost", "energy-cost")`` so that ``ha_discovery.py``'s ``entity.device.identifiers[1]`` is always valid (the service uses index [1] as the node_id throughout). """ from app.models.energy import EnergyCostPeriod # local import to avoid circular # --- Determine currency and representative pricing from the latest non-degraded row --- latest_period: EnergyCostPeriod | None = ( session.query(EnergyCostPeriod) .filter(EnergyCostPeriod.degraded.is_(False)) .order_by(EnergyCostPeriod.period_start.desc()) .first() ) currency: str = "EUR" if latest_period is not None and latest_period.currency: currency = latest_period.currency # --- Shared DeviceInfo (2-element identifiers — required by ha_discovery.py [1] access) --- # provides_availability=False: the energy-cost device has only sensors and no # online/offline heartbeat, so its entities must be "always available" in HA. # (Otherwise HA shows them unavailable despite state being published.) device_info = DeviceInfo( identifiers=("energy-cost", "energy-cost"), name="Energy Cost", provides_availability=False, ) # --- value_getter: current buy price --- def _make_buy_price_getter() -> Callable[["Session"], Any]: """Return a getter for the current buy price per kWh. For ``kind="manual"`` contracts the getter selects the price slot that matches the current DSMR electricity tariff: - tariff == 1 (dal / off-peak) → ``buy_dal`` - tariff == 2 (normal / peak) → ``buy_normal`` - tariff is None / unknown → ``buy_normal`` (safe fallback = current status quo) For ``kind="tibber"`` (hourly dynamic pricing) the tariff slot is not applicable; the getter always uses the ``buy`` key from the snapshot. """ def _getter(sess: "Session") -> Any: from app.models.energy import EnergyCostPeriod as _ECP from app.services.dsmr_ingest import get_current_tariff period = ( sess.query(_ECP) .filter(_ECP.degraded.is_(False)) .order_by(_ECP.period_start.desc()) .first() ) if period is None or not period.pricing: return None snap = period.pricing kind = snap.get("kind") if kind == "tibber": raw = snap.get("buy") elif kind == "manual": tariff = get_current_tariff() if tariff == 1: raw = snap.get("buy_dal") else: # tariff == 2, None, or any unexpected value → normal (peak) rate. raw = snap.get("buy_normal") else: # Unknown kind — attempt common keys gracefully. raw = snap.get("buy") or snap.get("buy_normal") if raw is None: return None try: return float(raw) except (TypeError, ValueError): return None return _getter # --- value_getter: current sell price --- def _make_sell_price_getter() -> Callable[["Session"], Any]: """Return a getter for the current sell price per kWh. For ``kind="manual"`` contracts the getter selects the price slot that matches the current DSMR electricity tariff: - tariff == 1 (dal / off-peak) → ``sell_dal`` - tariff == 2 (normal / peak) → ``sell_normal`` - tariff is None / unknown → ``sell_normal`` (safe fallback = current status quo) For ``kind="tibber"`` the tariff slot is not applicable; always uses ``sell``. """ def _getter(sess: "Session") -> Any: from app.models.energy import EnergyCostPeriod as _ECP from app.services.dsmr_ingest import get_current_tariff period = ( sess.query(_ECP) .filter(_ECP.degraded.is_(False)) .order_by(_ECP.period_start.desc()) .first() ) if period is None or not period.pricing: return None snap = period.pricing kind = snap.get("kind") if kind == "tibber": raw = snap.get("sell") elif kind == "manual": tariff = get_current_tariff() if tariff == 1: raw = snap.get("sell_dal") else: # tariff == 2, None, or any unexpected value → normal (peak) rate. raw = snap.get("sell_normal") else: raw = snap.get("sell") or snap.get("sell_normal") if raw is None: return None try: return float(raw) except (TypeError, ValueError): return None return _getter # --- value_getter: cumulative import cost (incl. standing charges) --- def _make_import_cost_getter() -> Callable[["Session"], Any]: """Return a getter for the cumulative import cost plus prorated standing charges. D2 (M7): anchor = current active electricity meter's ``started_at``. After a meter swap the cumulative resets to zero for the new meter — old-meter periods have ``period_start < new_meter.started_at`` and fall outside the [anchor, now) window, so they are naturally excluded. Cross-meter boundary periods are already degraded and also excluded. No active electricity meter → returns None (cannot anchor; safer than returning a stale or wrong value). The check is done via an inline query (``ended_at IS NULL``) rather than ``meter_at(now)`` to be robust against the edge case where the active meter's ``started_at`` is in the future (``meter_at(now)`` would return None in that scenario). No non-degraded periods at all → returns None (has_data guard). No active contract → ``summarize`` still runs but fixed_costs = 0; the return value is the pure metered sum within the current meter window. This is consistent with D2: the anchor is the meter, not the contract. value = summarize(anchor → now).metered_import + summarize(anchor → now).fixed_costs """ def _getter(sess: "Session") -> Any: from datetime import UTC, datetime as _dt from app.models.energy import EnergyCostPeriod as _ECP, Meter as _Meter from app.services.energy_cost import summarize as _summarize from sqlalchemy import func as _func, select as _select # Quick check: any non-degraded period at all? has_data = sess.query(_func.sum(_ECP.import_cost)).filter( _ECP.degraded.is_(False) ).scalar() if has_data is None: return None # D2: anchor = active electricity meter's started_at. # Inline query (ended_at IS NULL) is more robust than meter_at(now) # because it avoids the edge case where started_at is in the future. active_meter = sess.execute( _select(_Meter) .where( _Meter.commodity == "electricity", _Meter.ended_at.is_(None), ) .limit(1) ).scalar_one_or_none() if active_meter is None: # No active electricity meter — cannot anchor; return None. return None from app.services.contracts import _as_utc as _cu anchor_utc = _cu(active_meter.started_at) now_utc = _dt.now(UTC) result = _summarize(sess, anchor_utc, now_utc) return result["metered_import"] + result["fixed_costs"] return _getter # --- value_getter: cumulative export revenue (incl. tax credit) --- def _make_export_revenue_getter() -> Callable[["Session"], Any]: """Return a getter for the cumulative export revenue plus prorated tax credit. D2 (M7): anchor = current active electricity meter's ``started_at``. Same reasoning as the import cost getter — see its docstring. value = summarize(anchor → now).metered_export + summarize(anchor → now).credits Returns None when no non-degraded period exists or no active electricity meter. """ def _getter(sess: "Session") -> Any: from datetime import UTC, datetime as _dt from app.models.energy import EnergyCostPeriod as _ECP, Meter as _Meter from app.services.energy_cost import summarize as _summarize from sqlalchemy import func as _func, select as _select # Quick check: any non-degraded period at all? has_data = sess.query(_func.sum(_ECP.export_revenue)).filter( _ECP.degraded.is_(False) ).scalar() if has_data is None: return None # D2: anchor = active electricity meter's started_at. active_meter = sess.execute( _select(_Meter) .where( _Meter.commodity == "electricity", _Meter.ended_at.is_(None), ) .limit(1) ).scalar_one_or_none() if active_meter is None: # No active electricity meter — cannot anchor; return None. return None from app.services.contracts import _as_utc as _cu anchor_utc = _cu(active_meter.started_at) now_utc = _dt.now(UTC) result = _summarize(sess, anchor_utc, now_utc) return result["metered_export"] + result["credits"] return _getter # --- value_getter: today's import cost (local-day window, resets at local midnight) --- def _make_import_cost_today_getter() -> Callable[["Session"], Any]: """Return a getter for today's import cost (metered + standing for today). Window: [local today 00:00, local tomorrow 00:00) in UTC. After local midnight the window rolls to the new day → value resets to that day's running cost, implementing "resets at local midnight" semantics for HA. Uses timezone module via module-attribute access to preserve monkeypatch safety. Returns None when no non-degraded period exists or no active contract. """ def _getter(sess: "Session") -> Any: from app.services.contracts import active_contract_versions from app.models.energy import EnergyCostPeriod as _ECP from app.services.energy_cost import summarize as _summarize from app.services import timezone as _tz_mod from sqlalchemy import func as _func from datetime import timedelta as _td # Quick check: any non-degraded period at all? has_data = sess.query(_func.sum(_ECP.import_cost)).filter( _ECP.degraded.is_(False) ).scalar() if has_data is None: return None versions = active_contract_versions(sess) if not versions: return None # Today's window in UTC, using monkeypatch-safe module attribute calls. today_local = _tz_mod.local_now().date() tomorrow_local = today_local + _td(days=1) today_start_utc = _tz_mod.local_midnight_utc(today_local) tomorrow_start_utc = _tz_mod.local_midnight_utc(tomorrow_local) result = _summarize(sess, today_start_utc, tomorrow_start_utc) return result["metered_import"] + result["fixed_costs"] return _getter # --- value_getter: today's export revenue (local-day window, resets at local midnight) --- def _make_export_revenue_today_getter() -> Callable[["Session"], Any]: """Return a getter for today's export revenue (metered + tax credit for today). Same window semantics as import_cost_today. Returns None when no non-degraded period exists or no active contract. """ def _getter(sess: "Session") -> Any: from app.services.contracts import active_contract_versions from app.models.energy import EnergyCostPeriod as _ECP from app.services.energy_cost import summarize as _summarize from app.services import timezone as _tz_mod from sqlalchemy import func as _func from datetime import timedelta as _td # Quick check: any non-degraded period at all? has_data = sess.query(_func.sum(_ECP.export_revenue)).filter( _ECP.degraded.is_(False) ).scalar() if has_data is None: return None versions = active_contract_versions(sess) if not versions: return None today_local = _tz_mod.local_now().date() tomorrow_local = today_local + _td(days=1) today_start_utc = _tz_mod.local_midnight_utc(today_local) tomorrow_start_utc = _tz_mod.local_midnight_utc(tomorrow_local) result = _summarize(sess, today_start_utc, tomorrow_start_utc) return result["metered_export"] + result["credits"] return _getter # Price unit string: "/kWh" price_unit = f"{currency}/kWh" entities: list[ExposableEntity] = [ ExposableEntity( key="energy.buy_price_now", component="sensor", device=device_info, device_class=None, unit=price_unit, name="Energy Buy Price Now", value_getter=_make_buy_price_getter(), state_class="measurement", ), ExposableEntity( key="energy.sell_price_now", component="sensor", device=device_info, device_class=None, unit=price_unit, name="Energy Sell Price Now", value_getter=_make_sell_price_getter(), state_class="measurement", ), ExposableEntity( key="energy.import_cost_total", component="sensor", device=device_info, device_class="monetary", unit=currency, name="Energy Import Cost (incl. standing)", value_getter=_make_import_cost_getter(), state_class="total", ), ExposableEntity( key="energy.export_revenue_total", component="sensor", device=device_info, device_class="monetary", unit=currency, name="Energy Export Revenue (incl. tax credit)", value_getter=_make_export_revenue_getter(), state_class="total", ), # Daily entities: reset at local midnight, state_class=total_increasing. # # device_class=monetary is correct here. The HA developer docs (long-term # statistics section) prohibit MONETARY from being combined with # state_class=measurement — they do NOT prohibit it with total or # total_increasing. monetary+total_increasing is valid (the cumulative # entities above use monetary+total with no issue). # # total_increasing is appropriate because these daily quantities are # non-negative (import/export metered cost ≥ 0, fixed standing fee/tax # credit ≥ 0) and monotonically non-decreasing within any given local day. # When the local midnight rolls over the value drops back to ~1 day's fixed # fee; HA interprets that decrease as a new cycle, giving correct daily # aggregation without any explicit last_reset pipeline. ExposableEntity( key="energy.import_cost_today", component="sensor", device=device_info, device_class="monetary", unit=currency, name="Energy Import Cost Today (incl. standing)", value_getter=_make_import_cost_today_getter(), state_class="total_increasing", ), ExposableEntity( key="energy.export_revenue_today", component="sensor", device=device_info, device_class="monetary", unit=currency, name="Energy Export Revenue Today (incl. tax credit)", value_getter=_make_export_revenue_today_getter(), state_class="total_increasing", ), ] return entities # Register the energy cost provider at module load time. register_provider(_energy_cost_provider)