home-automation/app/services/dsmr_ingest.py

"""DSMR telegram ingest service.

Subscribes to the DSMR Reader MQTT topic (``dsmr/json``) and persists
down-sampled DSMR telegram frames to the ``dsmr_reading`` table.

Design decisions
----------------
- **Whole-frame storage**: the entire parsed telegram dict is stored as a JSON
  blob in ``DsmrReading.payload``; no field allow-list is applied.  This lets
  future commodities (gas, heating, three-phase) be accommodated without a
  table-schema change.
- **10-second down-sampling** (configurable via ``dsmr_sample_interval_s``):
  only telegrams whose ``timestamp`` second falls on an exact multiple of the
  interval are persisted.  This reduces write volume from ~60 rows/min to ~6
  rows/min while guaranteeing that every 15-minute boundary (second=00) is
  captured.
- **Idempotency**: the telegram's own ``id`` field is stored as ``source_id``
  with a UNIQUE constraint.  A second delivery of the same telegram (e.g. after
  a broker reconnect) is silently skipped.
- **Network-thread safety**: ``handle_message`` is called from paho's background
  loop thread.  It opens and closes its own short-lived SQLAlchemy session and
  swallows all exceptions so that a buggy payload or transient DB error never
  crashes the paho loop or drops the MQTT connection.
- **Numeric values kept as strings**: the DSMR Reader emits all numeric readings
  as JSON strings (e.g. ``"20915.154"``).  They are stored verbatim; conversion
  to ``Decimal`` is deferred to the billing engine (T07) where precision matters.
- **Null phases**: some telegrams omit certain phase readings (``null`` in JSON);
  these are stored as-is without special handling.
"""

from __future__ import annotations

import json
import logging
from datetime import datetime, timezone
from typing import TYPE_CHECKING

import sqlalchemy.exc
from sqlalchemy import select

from app.db import get_session_local
from app.models.energy import DsmrReading

if TYPE_CHECKING:
    from app.config import Settings

logger = logging.getLogger(__name__)


# Tracks the DSMR topic currently subscribed via the MQTT manager, so a config
# change can unsubscribe the old topic before subscribing the new one.
_current_dsmr_topic: str | None = None


def apply_dsmr_subscription(settings: "Settings") -> None:
    """(Re)apply the DSMR MQTT subscription to match *settings* — restart-free.

    Call this at startup and after every config save.  It makes the live MQTT
    subscription reflect the current ``dsmr_ingest_enabled`` / ``dsmr_mqtt_topic``
    / ``dsmr_sample_interval_s`` settings without an app restart:

    - **Disabled** → unsubscribe any active DSMR subscription.
    - **Enabled** → (re)subscribe to ``dsmr_mqtt_topic`` with a handler bound to
      a *fresh* settings snapshot, so a changed sample interval also takes effect.
    - **Topic changed** → unsubscribe the old topic before subscribing the new one.

    Idempotent and safe to call when MQTT is not connected (the subscription is
    queued in the manager and established on the next connect).
    """
    # Imported here (not at module top) to avoid a circular import at app start.
    from app.integrations.mqtt import mqtt_manager

    global _current_dsmr_topic

    if not settings.dsmr_ingest_enabled:
        if _current_dsmr_topic is not None:
            mqtt_manager.unsubscribe(_current_dsmr_topic)
            logger.info("DSMR ingest disabled — unsubscribed from topic=%s.", _current_dsmr_topic)
            _current_dsmr_topic = None
        return

    topic = settings.dsmr_mqtt_topic
    if _current_dsmr_topic is not None and _current_dsmr_topic != topic:
        mqtt_manager.unsubscribe(_current_dsmr_topic)

    # Re-subscribe (overwrites any existing handler for this topic) with a fresh
    # settings snapshot so dsmr_sample_interval_s changes take effect too.
    snapshot = settings
    mqtt_manager.subscribe(topic, lambda payload: handle_message(payload, snapshot))
    _current_dsmr_topic = topic
    logger.info("DSMR ingest enabled — subscribed to topic=%s.", topic)


def handle_message(payload_bytes: bytes, settings: "Settings") -> None:
    """Parse one DSMR MQTT payload and persist it if it passes the sample filter.

    Called from the paho network thread; *must* swallow all exceptions so that
    a bad payload or transient error does not crash the loop or drop the broker
    connection.

    Parameters
    ----------
    payload_bytes:
        Raw bytes from the MQTT message.
    settings:
        Runtime settings snapshot (captured at subscription time).  Used for
        ``dsmr_sample_interval_s``.
    """
    try:
        _handle_message_inner(payload_bytes, settings)
    except Exception:
        logger.exception("dsmr_ingest.handle_message: unexpected error (swallowed).")


def _handle_message_inner(payload_bytes: bytes, settings: "Settings") -> None:
    """Inner implementation — may raise; caller wraps in try/except."""
    # --- 1. Parse JSON ---
    try:
        data: dict = json.loads(payload_bytes)
    except (json.JSONDecodeError, ValueError):
        logger.debug("dsmr_ingest: invalid JSON payload (skipped).")
        return

    if not isinstance(data, dict):
        logger.debug("dsmr_ingest: payload is not a JSON object (skipped).")
        return

    # --- 2. Parse timestamp ---
    raw_ts = data.get("timestamp")
    if raw_ts is None:
        logger.debug("dsmr_ingest: missing 'timestamp' field (skipped).")
        return

    try:
        # Python 3.11+ accepts the trailing 'Z' directly; for 3.10 compat we
        # replace 'Z' with '+00:00' before parsing.
        ts_str = raw_ts if not isinstance(raw_ts, str) else raw_ts.replace("Z", "+00:00")
        ts_utc: datetime = datetime.fromisoformat(ts_str)
        # Ensure it is timezone-aware UTC.
        if ts_utc.tzinfo is None:
            ts_utc = ts_utc.replace(tzinfo=timezone.utc)
    except (ValueError, TypeError, AttributeError):
        logger.debug(
            "dsmr_ingest: cannot parse 'timestamp' value %r (skipped).", raw_ts
        )
        return

    # --- 3. Down-sample: only persist if second falls on interval boundary ---
    interval = settings.dsmr_sample_interval_s
    if interval > 0 and (ts_utc.second % interval) != 0:
        # This telegram is between sample points; discard silently.
        return

    # --- 4. Extract source_id (telegram's own id) — stored only as a reference,
    #        NOT used for uniqueness/idempotency (it overflows and gets reset). ---
    source_id: int | None = data.get("id")
    if source_id is not None and not isinstance(source_id, int):
        # Unexpected type — treat as missing rather than raising.
        logger.debug(
            "dsmr_ingest: 'id' field has unexpected type %s (ignoring).",
            type(source_id).__name__,
        )
        source_id = None

    # --- 5. Persist to database ---
    # Idempotency is keyed on recorded_at (the telegram timestamp), which is
    # telegram-id-independent: a single P1 meter emits one telegram per second,
    # and down-sampling keeps at most one per interval-aligned second.  The
    # UNIQUE(recorded_at) constraint is the backstop for the IntegrityError race.
    session_local = get_session_local()
    session = session_local()
    try:
        existing = session.scalar(
            select(DsmrReading).where(DsmrReading.recorded_at == ts_utc)
        )
        if existing is not None:
            logger.debug(
                "dsmr_ingest: recorded_at=%s already in DB, skipping.",
                ts_utc.isoformat(),
            )
            return

        reading = DsmrReading(
            recorded_at=ts_utc,
            source_id=source_id,
            payload=data,  # full frame, verbatim
        )
        session.add(reading)
        session.commit()
        logger.debug(
            "dsmr_ingest: persisted reading recorded_at=%s source_id=%s.",
            ts_utc.isoformat(),
            source_id,
        )
    except sqlalchemy.exc.IntegrityError:
        # Race / duplicate: another insert beat us to the same recorded_at.
        session.rollback()
        logger.debug(
            "dsmr_ingest: IntegrityError for recorded_at=%s (duplicate, skipped).",
            ts_utc.isoformat(),
        )
    except Exception:
        session.rollback()
        logger.exception("dsmr_ingest: DB error (swallowed).")
    finally:
        session.close()