M5-T11: add HA discovery and state publishing wired to polling

app/integrations/expose.py

@@ -91,9 +91,10 @@ class ExposableEntity:
         context, e.g. ``"SDM120 Voltage"``).
 
     value_getter
-        Optional callable ``() -> object | None`` that returns the current
-        entity value.  ``None`` means "not yet wired" (T11 will populate this).
-        T11 calls this to obtain the current state before publishing.
+        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"``).
@@ -106,7 +107,7 @@ class ExposableEntity:
     device_class: Optional[str]
     unit: str
     name: str
-    value_getter: Optional[Callable[[], Any]] = field(default=None, repr=False)
+    value_getter: Optional[Callable[["Session"], Any]] = field(default=None, repr=False)
     state_class: Optional[str] = None
 
 
@@ -269,16 +270,29 @@ def _modbus_provider(session: Session) -> list[ExposableEntity]:
         for metric in profile.metrics:
             entity_key = f"modbus.{device.uuid}.{metric.key}"
 
-            # Capture metric and device for the value_getter closure.
-            _device_uuid = device.uuid
+            # 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_uuid: str, m_key: str) -> Callable[[], Any]:
-                """Return a getter stub for this metric; T11 will wire real retrieval."""
-                def _getter() -> Any:
-                    # Stub — T11 will call build_catalog with a live session
-                    # and replace this with a real value-fetch callable.
-                    return None
+            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
 
@@ -290,19 +304,30 @@ def _modbus_provider(session: Session) -> list[ExposableEntity]:
                     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_uuid, _metric_key),
+                    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_uuid_online = device.uuid
+        _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"
 
-        def _make_online_getter(dev_uuid: str) -> Callable[[], Any]:
-            def _getter() -> Any:
-                # T11 will wire this to last_poll_ok.
-                return None
             return _getter
 
         entities.append(
@@ -313,7 +338,7 @@ def _modbus_provider(session: Session) -> list[ExposableEntity]:
                 device_class="connectivity",
                 unit="",
                 name=f"{device.friendly_name} Online",
-                value_getter=_make_online_getter(_dev_uuid_online),
+                value_getter=_make_online_getter(_dev_id_online),
                 state_class=None,
             )
         )

app/main.py

@@ -28,6 +28,7 @@ from app.services.auth import AuthBootstrapError, initialize_auth_schema
 from app.services.config_page import seed_missing_config_from_bootstrap, sync_app_hostname_from_bootstrap
 from app.services.public_ip import check_public_ipv4_and_notify
 from app.services.modbus_poll import poll_all_enabled_devices, BASE_POLL_TICK_SECONDS
+from app.services.ha_discovery import publish_discovery, publish_states
 from scripts.app_db_adopt import AppDatabaseAdoptionError, validate_app_runtime_db
 
 logger = logging.getLogger(__name__)
@@ -60,6 +61,30 @@ def _run_scheduled_modbus_poll() -> None:
         session.close()
 
 
+def _run_scheduled_ha_state_publish() -> None:
+    """Periodic job: publish discovery configs + state + availability for all enabled exposed entities.
+
+    Runs every 60 seconds.  When the MQTT broker is connected:
+    - Publishes (or re-publishes) all HA Discovery configs (retained, idempotent).
+    - Publishes the current state / availability for all enabled entities.
+
+    This also serves as the reliable "publish discovery after connect" mechanism:
+    because paho connects asynchronously, a synchronous call immediately after
+    ``mqtt_manager.connect()`` would fire before the TCP handshake completes and
+    be a no-op.  Instead, this periodic job picks it up within 60 seconds of the
+    broker becoming available — retained payloads make repeated publishes harmless.
+    """
+    session_local = get_session_local()
+    session: Session = session_local()
+    try:
+        publish_discovery(session)
+        publish_states(session)
+    except Exception:
+        logger.exception("_run_scheduled_ha_state_publish: unexpected error")
+    finally:
+        session.close()
+
+
 def ensure_auth_db_ready() -> None:
     session_local = get_session_local()
     session: Session = session_local()
@@ -103,9 +128,20 @@ async def lifespan(_: FastAPI):
         max_instances=1,
         coalesce=True,
     )
+    # Periodic HA state / availability publish (60-second fallback sweep).
+    scheduler.add_job(
+        _run_scheduled_ha_state_publish,
+        trigger=IntervalTrigger(seconds=60),
+        id="ha-state-publish",
+        replace_existing=True,
+        max_instances=1,
+        coalesce=True,
+    )
     scheduler.start()
 
     # MQTT: connect if configured.
+    # Discovery will be published by the first run of _run_scheduled_ha_state_publish
+    # (within 60 s of startup), after the async paho handshake completes.
     mqtt_manager.connect(get_settings())
 
     yield

app/services/ha_discovery.py

@@ -0,0 +1,409 @@
+"""HA MQTT Discovery service — build and publish discovery configs + state.
+
+This module handles:
+
+1. Building HA MQTT Discovery payloads for each ``ExposableEntity``.
+2. Publishing discovery configs (retained) for enabled entities.
+3. Clearing (empty-payload) discovery configs for disabled entities.
+4. Publishing current entity state values.
+5. A per-device helper for use in the Modbus poll loop.
+
+Design
+------
+- **No-op when MQTT / discovery is not configured**: every public function
+  checks ``ha_discovery_enabled`` and ``mqtt_manager.is_connected`` before
+  doing any work.  Callers never need to guard these conditions.
+- **Stable unique_id**: derived from device ``uuid`` + metric ``key``, never
+  from mutable fields like ``friendly_name`` or auto-increment DB ids.
+- **Discovery topic format**: ``<prefix>/<component>/<node_id>/<object_id>/config``
+  where ``node_id`` is the device uuid (slugified to be safe) and
+  ``object_id`` is a uuid-prefixed stable string.
+- **State topic format**: ``<prefix>/<component>/<node_id>/<object_id>/state``
+- **Availability topic**: ``<prefix>/modbus/<node_id>/availability`` (shared
+  across all entities of the same device; publishes "online"/"offline").
+- **best-effort**: functions catch all exceptions internally so that callers
+  (e.g. the Modbus poll loop) never crash due to MQTT publish errors.
+"""
+
+from __future__ import annotations
+
+import json
+import logging
+from typing import Any
+
+from sqlalchemy.orm import Session
+
+from app.integrations.expose import ExposableEntity, build_catalog
+from app.integrations.mqtt import mqtt_manager
+
+logger = logging.getLogger(__name__)
+
+
+# ---------------------------------------------------------------------------
+# Internal helpers
+# ---------------------------------------------------------------------------
+
+
+def _get_settings() -> Any:
+    """Return the current runtime settings (cached singleton)."""
+    from app.config import get_settings
+    return get_settings()
+
+
+def _should_publish(settings: Any) -> bool:
+    """Return True only if MQTT and HA Discovery are both enabled and connected."""
+    return bool(
+        settings.mqtt_enabled
+        and settings.ha_discovery_enabled
+        and mqtt_manager.is_connected
+    )
+
+
+def _node_id(device_uuid: str) -> str:
+    """Stable MQTT node_id derived from device uuid (replace hyphens for safety)."""
+    return device_uuid.replace("-", "_")
+
+
+def _object_id(entity: ExposableEntity) -> str:
+    """Stable MQTT object_id derived from entity key (dots + hyphens → underscores)."""
+    return entity.key.replace(".", "_").replace("-", "_")
+
+
+def _discovery_topic(entity: ExposableEntity, prefix: str) -> str:
+    """Build the HA Discovery config topic for *entity*.
+
+    Format: ``<prefix>/<component>/<node_id>/<object_id>/config``
+    """
+    node = _node_id(entity.device.identifiers[1])  # device uuid
+    obj = _object_id(entity)
+    return f"{prefix}/{entity.component}/{node}/{obj}/config"
+
+
+def _state_topic(entity: ExposableEntity, prefix: str) -> str:
+    """Build the state publish topic for *entity*.
+
+    Format: ``<prefix>/<component>/<node_id>/<object_id>/state``
+    """
+    node = _node_id(entity.device.identifiers[1])
+    obj = _object_id(entity)
+    return f"{prefix}/{entity.component}/{node}/{obj}/state"
+
+
+def _availability_topic(device_uuid: str, prefix: str) -> str:
+    """Shared availability topic for all entities of a device.
+
+    Format: ``<prefix>/modbus/<node_id>/availability``
+    """
+    node = _node_id(device_uuid)
+    return f"{prefix}/modbus/{node}/availability"
+
+
+def _unique_id(entity: ExposableEntity) -> str:
+    """Stable unique_id — device uuid + metric key (never from mutable fields)."""
+    device_uuid = entity.device.identifiers[1]
+    # entity.key is already "modbus.<uuid>.<metric_key>" — use it as the seed
+    return f"{device_uuid}_{entity.key.replace('.', '_')}"
+
+
+# ---------------------------------------------------------------------------
+# Public: build discovery payload
+# ---------------------------------------------------------------------------
+
+
+def build_discovery_payload(
+    entity: ExposableEntity,
+    prefix: str,
+) -> tuple[str, dict[str, Any]]:
+    """Build the HA MQTT Discovery config topic and payload dict for *entity*.
+
+    Parameters
+    ----------
+    entity:
+        An ``ExposableEntity`` (from ``build_catalog``).
+    prefix:
+        The discovery prefix (e.g. ``"homeassistant"``).
+
+    Returns
+    -------
+    tuple[str, dict]
+        ``(topic, config_dict)`` — the config topic and the HA Discovery
+        payload (not yet JSON-serialised).
+    """
+    device_uuid = entity.device.identifiers[1]
+    avail_topic = _availability_topic(device_uuid, prefix)
+    state_t = _state_topic(entity, prefix)
+    topic = _discovery_topic(entity, prefix)
+
+    config: dict[str, Any] = {
+        "unique_id": _unique_id(entity),
+        "name": entity.name,
+        "state_topic": state_t,
+        "availability": [{"topic": avail_topic}],
+        "availability_mode": "all",
+        "device": {
+            "identifiers": list(entity.device.identifiers),
+            "name": entity.device.name,
+        },
+    }
+
+    # Component-specific fields
+    if entity.component == "binary_sensor":
+        # "online" binary sensor: payload ON/OFF
+        config["payload_on"] = "ON"
+        config["payload_off"] = "OFF"
+        if entity.device_class:
+            config["device_class"] = entity.device_class
+    else:
+        # sensor (and others)
+        if entity.device_class:
+            config["device_class"] = entity.device_class
+        if entity.unit:
+            config["unit_of_measurement"] = entity.unit
+        if entity.state_class:
+            config["state_class"] = entity.state_class
+
+    return topic, config
+
+
+# ---------------------------------------------------------------------------
+# Public: publish discovery
+# ---------------------------------------------------------------------------
+
+
+def publish_discovery(session: Session) -> None:
+    """Publish HA Discovery configs for all entities in the catalog.
+
+    - Enabled entities: publish retained JSON config payload.
+    - Disabled entities: publish empty (b"") payload to clear any previously
+      retained config from HA.
+
+    No-op if MQTT / discovery is not enabled or the client is not connected.
+    All MQTT errors are caught internally — this function never raises.
+    """
+    settings = _get_settings()
+    if not _should_publish(settings):
+        logger.debug("publish_discovery: skipped (MQTT/Discovery not enabled or not connected)")
+        return
+
+    prefix = settings.ha_discovery_prefix
+
+    try:
+        catalog = build_catalog(session)
+    except Exception:
+        logger.exception("publish_discovery: failed to build catalog; aborting")
+        return
+
+    for entry in catalog:
+        entity = entry.entity
+        try:
+            topic, config = build_discovery_payload(entity, prefix)
+            if entry.enabled:
+                payload = json.dumps(config)
+                mqtt_manager.publish(topic, payload, retain=True)
+                logger.debug(
+                    "publish_discovery: published config for %r → %s", entity.key, topic
+                )
+            else:
+                # Clear retained config for disabled entities.
+                mqtt_manager.publish(topic, b"", retain=True)
+                logger.debug(
+                    "publish_discovery: cleared config for disabled entity %r → %s",
+                    entity.key,
+                    topic,
+                )
+        except Exception:
+            logger.exception(
+                "publish_discovery: error processing entity %r; continuing", entity.key
+            )
+
+
+# ---------------------------------------------------------------------------
+# Public: publish states
+# ---------------------------------------------------------------------------
+
+
+def publish_states(session: Session) -> None:
+    """Publish current state values for all enabled entities.
+
+    Calls each entity's ``value_getter`` to obtain the current value and
+    publishes it to the entity's state topic.  Also publishes availability
+    (online/offline) for each device.
+
+    No-op if MQTT / discovery is not enabled or the client is not connected.
+    All errors are caught internally — this function never raises.
+    """
+    settings = _get_settings()
+    if not _should_publish(settings):
+        logger.debug("publish_states: skipped (MQTT/Discovery not enabled or not connected)")
+        return
+
+    prefix = settings.ha_discovery_prefix
+
+    try:
+        catalog = build_catalog(session)
+    except Exception:
+        logger.exception("publish_states: failed to build catalog; aborting")
+        return
+
+    # Collect enabled entries and publish
+    for entry in catalog:
+        if not entry.enabled:
+            continue
+        entity = entry.entity
+        try:
+            _publish_entity_state(entity, prefix, session)
+        except Exception:
+            logger.exception(
+                "publish_states: error publishing state for %r; continuing", entity.key
+            )
+
+
+def _publish_entity_state(
+    entity: ExposableEntity,
+    prefix: str,
+    session: Session,
+) -> None:
+    """Publish one entity's current state value to its state topic.
+
+    Also publishes the availability topic for ``binary_sensor`` "online" entities.
+    """
+    state_t = _state_topic(entity, prefix)
+    device_uuid = entity.device.identifiers[1]
+
+    if entity.component == "binary_sensor" and "online" in entity.key:
+        # The online sensor represents device availability.
+        # Ask the value_getter for the current ON/OFF string, then also publish
+        # the shared availability topic (online/offline).
+        raw_value = None
+        if entity.value_getter is not None:
+            try:
+                raw_value = entity.value_getter(session)
+            except Exception:
+                logger.exception("value_getter raised for online entity %r", entity.key)
+        # 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)
+        mqtt_manager.publish(avail_topic, avail_payload, retain=False)
+        # The state of the binary_sensor itself
+        state_payload = "ON" if online else "OFF"
+        mqtt_manager.publish(state_t, state_payload, retain=False)
+    else:
+        # Regular sensor: call value_getter(session) if available.
+        value = None
+        if entity.value_getter is not None:
+            try:
+                value = entity.value_getter(session)
+            except Exception:
+                logger.exception("value_getter raised for entity %r", entity.key)
+
+        if value is None:
+            logger.debug("publish_states: no value for %r — skipping state publish", entity.key)
+            return
+
+        mqtt_manager.publish(state_t, str(value), retain=False)
+
+
+# ---------------------------------------------------------------------------
+# Public: per-device state push (called by modbus_poll after each poll)
+# ---------------------------------------------------------------------------
+
+
+def publish_device_state(session: Session, device: Any) -> None:
+    """Publish state + availability for all enabled entities of *device*.
+
+    Called by ``modbus_poll.poll_device`` after a successful (or failed) poll.
+    Publishes:
+    - The current availability topic ("online" / "offline") for the device.
+    - The state topic for each **enabled** entity of the device.
+
+    No-op if MQTT / discovery is not enabled or the client is not connected.
+    All errors are caught internally — never raises.
+    """
+    settings = _get_settings()
+    if not _should_publish(settings):
+        return
+
+    prefix = settings.ha_discovery_prefix
+    device_uuid = device.uuid
+    online = bool(device.last_poll_ok)
+
+    # Publish availability topic first.
+    try:
+        avail_topic = _availability_topic(device_uuid, prefix)
+        avail_payload = "online" if online else "offline"
+        mqtt_manager.publish(avail_topic, avail_payload, retain=False)
+    except Exception:
+        logger.exception(
+            "publish_device_state: failed to publish availability for device uuid=%s", device_uuid
+        )
+
+    if not online:
+        # Device is offline — no point pushing stale state values.
+        return
+
+    # Publish state for each enabled entity of this device.
+    try:
+        catalog = build_catalog(session)
+    except Exception:
+        logger.exception(
+            "publish_device_state: failed to build catalog for device uuid=%s", device_uuid
+        )
+        return
+
+    for entry in catalog:
+        if not entry.enabled:
+            continue
+        entity = entry.entity
+        # Only process entities belonging to this device.
+        if entity.device.identifiers[1] != device_uuid:
+            continue
+        # Skip the online binary_sensor itself (availability handled above).
+        if entity.component == "binary_sensor" and "online" in entity.key:
+            # Publish the binary_sensor state too.
+            try:
+                state_t = _state_topic(entity, prefix)
+                mqtt_manager.publish(state_t, "ON", retain=False)
+            except Exception:
+                logger.exception(
+                    "publish_device_state: error publishing online sensor state for %r",
+                    entity.key,
+                )
+            continue
+
+        # Regular sensor — call value_getter(session).
+        try:
+            value = None
+            if entity.value_getter is not None:
+                value = entity.value_getter(session)
+            if value is None:
+                continue
+            state_t = _state_topic(entity, prefix)
+            mqtt_manager.publish(state_t, str(value), retain=False)
+        except Exception:
+            logger.exception(
+                "publish_device_state: error publishing state for entity %r", entity.key
+            )
+
+
+def publish_device_offline(device_uuid: str) -> None:
+    """Publish an "offline" availability payload for *device_uuid*.
+
+    Called by ``modbus_poll.poll_device`` when a poll fails, to immediately
+    reflect the device going offline in HA (before the next full state sweep).
+
+    No-op if MQTT / discovery is not enabled or the client is not connected.
+    Never raises.
+    """
+    settings = _get_settings()
+    if not _should_publish(settings):
+        return
+
+    try:
+        prefix = settings.ha_discovery_prefix
+        avail_topic = _availability_topic(device_uuid, prefix)
+        mqtt_manager.publish(avail_topic, "offline", retain=False)
+    except Exception:
+        logger.exception(
+            "publish_device_offline: failed for device uuid=%s", device_uuid
+        )

app/services/modbus_poll.py

@@ -31,6 +31,7 @@ from app.config import Settings
 from app.integrations.modbus import driver, profiles
 from app.models.modbus import ModbusDevice, ModbusReading
 from app.services.config_page import build_runtime_settings
+from app.services.ha_discovery import publish_device_offline, publish_device_state
 
 logger = logging.getLogger(__name__)
 
@@ -84,6 +85,19 @@ def poll_device(session: Session, device: ModbusDevice) -> ModbusReading | None:
             len(payload),
             now.isoformat(),
         )
+
+        # Best-effort: push MQTT state after successful poll.
+        # Must never let MQTT errors crash the poll loop.
+        try:
+            publish_device_state(session, device)
+        except Exception:  # noqa: BLE001
+            logger.exception(
+                "poll_device: MQTT state publish failed for device %r (id=%d); "
+                "continuing (poll itself succeeded)",
+                device.friendly_name,
+                device.id,
+            )
+
         return reading
 
     except Exception:  # noqa: BLE001
@@ -115,6 +129,17 @@ def poll_device(session: Session, device: ModbusDevice) -> ModbusReading | None:
             logger.exception(
                 "Also failed to persist last_poll_ok=False for device id=%d", device.id
             )
+
+        # Best-effort: publish "offline" availability after failed poll.
+        try:
+            publish_device_offline(device.uuid)
+        except Exception:  # noqa: BLE001
+            logger.exception(
+                "poll_device: MQTT offline publish failed for device %r (id=%d)",
+                device.friendly_name,
+                device.id,
+            )
+
         return None
 
 

docs/design/m5-iot-energy.md

@@ -432,7 +432,7 @@ Phase C（MQTT / Discovery，依赖 B 的设备数据与 provider 接口）
 - **Reviewer checklist**: 断网/连接失败不崩主进程；线程在 shutdown 正确停止；`requirements.txt` 同步锁定 `paho-mqtt`；密码不进日志。
 
 ### M5-T11 — Discovery 发布 + state 发布
-- **Status**: `todo` · **Depends**: M5-T09, M5-T10
+- **Status**: `done` · **Depends**: M5-T09, M5-T10
 - **Context**: 把 enabled 实体发成 HA discovery config（retained）并周期推 state；采集轮询后推最新值。
 - **Files**: `create app/services/ha_discovery.py`；`modify app/services/modbus_poll.py`（轮询后推 state）、`app/main.py`（state 周期 job + 连接后/勾选变更后发 discovery）、`app/api/routes/api/...`（`/api/expose/republish` 在 T12 接，本任务提供 service）；`create tests/test_ha_discovery.py`
 - **Steps**:

tests/test_expose_catalog.py

@@ -108,11 +108,17 @@ def test_exposable_entity_fields():
 
 
 def test_exposable_entity_with_value_getter():
-    """ExposableEntity.value_getter stores and calls the provided callable."""
+    """ExposableEntity.value_getter stores and calls the provided callable.
+
+    value_getter now accepts a Session argument (updated in T11-rework-1).
+    """
     from app.integrations.expose import DeviceInfo, ExposableEntity
+    from sqlalchemy import create_engine
+    from sqlalchemy.orm import Session
 
     device = DeviceInfo(identifiers=("modbus", "dev1"), name="D1")
-    getter = lambda: 230.5  # noqa: E731
+    # The value_getter receives a Session; use a lambda that ignores it.
+    getter = lambda _sess: 230.5  # noqa: E731
     entity = ExposableEntity(
         key="modbus.dev1.voltage",
         component="sensor",
@@ -123,7 +129,9 @@ def test_exposable_entity_with_value_getter():
         value_getter=getter,
     )
     assert entity.value_getter is not None
-    assert entity.value_getter() == 230.5
+    eng = create_engine("sqlite:///:memory:")
+    with Session(eng) as session:
+        assert entity.value_getter(session) == 230.5
 
 
 def test_entity_key_stability_uses_uuid_not_id():