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__)


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 field, for idempotency) ---
    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 ---
    session_local = get_session_local()
    session = session_local()
    try:
        # Idempotency check: if source_id is known and already exists, skip.
        if source_id is not None:
            existing = session.scalar(
                select(DsmrReading).where(DsmrReading.source_id == source_id)
            )
            if existing is not None:
                logger.debug(
                    "dsmr_ingest: source_id=%d already in DB, skipping.", source_id
                )
                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 condition: another handler inserted the same source_id between our
        # check and our insert.  Roll back and ignore.
        session.rollback()
        logger.debug(
            "dsmr_ingest: IntegrityError for source_id=%s (duplicate, skipped).",
            source_id,
        )
    except Exception:
        session.rollback()
        logger.exception("dsmr_ingest: DB error (swallowed).")
    finally:
        session.close()