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
import threading
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

# Tracks the DSMR tariff topic currently subscribed via the MQTT manager.
_current_tariff_topic: str | None = None

# Current electricity tariff: 1 = dal/off-peak, 2 = normal/peak.
# Written by the paho network thread, read by the publish job — guarded by a lock.
_current_tariff: int | None = None
_tariff_lock = threading.Lock()


def get_current_tariff() -> int | None:
    """Return the most recently received electricity tariff (1 or 2), or None."""
    with _tariff_lock:
        return _current_tariff


def set_current_tariff(value: int | None) -> None:
    """Set the current electricity tariff (1 or 2), or clear it with None."""
    with _tariff_lock:
        global _current_tariff
        _current_tariff = value


def handle_tariff_message(payload_bytes: bytes) -> None:
    """Parse one DSMR tariff MQTT payload and update the in-memory tariff state.

    Called from the paho network thread; *must* swallow all exceptions so that
    a bad payload never crashes the loop or drops the broker connection.

    Accepts payload as bytes or str (paho can deliver either).  Ignores
    whitespace.  Only accepts integer values 1 or 2; anything else is discarded
    and the previous known tariff is preserved.

    Parameters
    ----------
    payload_bytes:
        Raw bytes from the MQTT message (may also be a str in some paho versions).
    """
    try:
        # Decode bytes → str if needed; strip surrounding whitespace.
        if isinstance(payload_bytes, (bytes, bytearray)):
            raw = payload_bytes.decode("utf-8", errors="replace").strip()
        else:
            raw = str(payload_bytes).strip()

        value = int(raw)
        if value not in (1, 2):
            logger.debug(
                "dsmr_ingest.handle_tariff_message: unexpected tariff value %d (expected 1 or 2, ignored).",
                value,
            )
            return

        set_current_tariff(value)
        logger.debug("dsmr_ingest.handle_tariff_message: tariff updated to %d.", value)

    except Exception:
        # Malformed payload (e.g. non-numeric); swallow silently to protect network thread.
        logger.debug(
            "dsmr_ingest.handle_tariff_message: could not parse payload %r (ignored).",
            payload_bytes,
        )


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

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

    - **Disabled** → unsubscribe any active DSMR and tariff subscriptions.
    - **Enabled** → (re)subscribe to ``dsmr_mqtt_topic`` with a handler bound to
      a *fresh* settings snapshot, so a changed sample interval also takes effect.
      Also subscribe to ``dsmr_tariff_topic`` when non-empty.
    - **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, _current_tariff_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
        if _current_tariff_topic is not None:
            mqtt_manager.unsubscribe(_current_tariff_topic)
            logger.info(
                "DSMR ingest disabled — unsubscribed from tariff topic=%s.",
                _current_tariff_topic,
            )
            _current_tariff_topic = None
        return

    # --- Main DSMR telegram topic ---
    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)

    # --- DSMR tariff topic (dual-tariff slot indicator) ---
    tariff_topic = settings.dsmr_tariff_topic if settings.dsmr_tariff_topic else ""
    if tariff_topic:
        if _current_tariff_topic is not None and _current_tariff_topic != tariff_topic:
            mqtt_manager.unsubscribe(_current_tariff_topic)
        mqtt_manager.subscribe(tariff_topic, lambda payload: handle_tariff_message(payload))
        _current_tariff_topic = tariff_topic
        logger.info("DSMR tariff topic — subscribed to topic=%s.", tariff_topic)
    else:
        # tariff_topic is empty → unsubscribe any existing tariff subscription.
        if _current_tariff_topic is not None:
            mqtt_manager.unsubscribe(_current_tariff_topic)
            logger.info(
                "DSMR tariff topic cleared — unsubscribed from topic=%s.",
                _current_tariff_topic,
            )
            _current_tariff_topic = None


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