home-automation/app/services/warmtelink_ingest.py

"""Privacy-preserving WarmteLink frame admission and minute sampling.

The serial worker added later owns I/O.  This module deliberately only accepts
already parsed :class:`P1Telegram` instances (or a parser callable at its
small convenience entry point), so rejected telegram bytes never enter the
database or an exception message.
"""

from __future__ import annotations

from collections.abc import Callable
from dataclasses import dataclass
from datetime import UTC, datetime, timedelta
from decimal import Decimal
import re
from zoneinfo import ZoneInfo

from sqlalchemy import select
from sqlalchemy.orm import Session

from app.db import get_session_local
from app.integrations.p1 import IntegrityStatus, P1Channel, P1Telegram, parse_telegram
from app.models.meter_source import MeterSource, WarmteLinkReading
from app.services.meter_sources import upsert_discovered_channel


@dataclass(frozen=True)
class _ChannelSample:
    key: str
    label: str
    value: Decimal
    unit: str
    device_type: str | None
    fingerprint: str | None
    identity: tuple[object, ...]


@dataclass(frozen=True)
class _FrameSnapshot:
    recorded_at: datetime
    fingerprint: str | None
    samples: tuple[_ChannelSample, ...]


class WarmteLinkIngestor:
    """Keep unverifiable candidates isolated by source for one worker lifetime."""

    def __init__(
        self,
        *,
        clock: Callable[[], datetime] | None = None,
    ) -> None:
        self._clock = clock or (lambda: datetime.now(UTC))
        self._previous: dict[int, _FrameSnapshot] = {}

    def ingest(
        self,
        session: Session,
        *,
        source_id: int,
        telegram: P1Telegram,
        received_at: datetime | None = None,
    ) -> bool:
        """Apply one parsed telegram in the caller's transaction.

        Returns whether the frame was admitted.  Callers that own a session
        must commit on success; :func:`handle_frame` is the failure-contained
        worker-facing entry point.
        """
        source = session.get(MeterSource, source_id)
        if source is None:
            raise ValueError("WarmteLink source was not found")
        if source.kind != "warmtelink_serial":
            raise ValueError("Source is not a WarmteLink serial source")

        received_at = _utc(self._clock()) if received_at is None else _utc(received_at)
        try:
            integrity = _integrity_status(telegram.integrity)
        except Exception:
            self._previous.pop(source_id, None)
            self._diagnose(source, "WarmteLink frame could not be normalized", now=received_at)
            return False

        if integrity is IntegrityStatus.INVALID:
            self._previous.pop(source_id, None)
            self._diagnose(source, "WarmteLink frame checksum is invalid", now=received_at)
            return False

        try:
            snapshot = _snapshot(telegram, received_at)
            fingerprints = _final_fingerprints(snapshot)
        except Exception:
            # A parser DTO remains an untrusted boundary.  Do not let an
            # unnormalizable DTO bridge two otherwise matching candidates.
            self._previous.pop(source_id, None)
            self._diagnose(source, "WarmteLink frame could not be normalized", now=received_at)
            return False
        if fingerprints is None:
            self._previous.pop(source_id, None)
            self._diagnose(source, "WarmteLink frame fingerprint is invalid", now=received_at)
            return False
        if integrity is IntegrityStatus.UNVERIFIABLE:
            previous = self._previous.get(source_id)
            if previous is None:
                self._previous[source_id] = snapshot
                self._diagnose(
                    source, "Awaiting a second matching unverifiable WarmteLink frame", now=received_at
                )
                return False
            reason = _continuity_problem(previous, snapshot)
            if reason is not None:
                self._previous[source_id] = snapshot
                self._diagnose(source, reason, now=received_at)
                return False
        try:
            self._admit(session, source, snapshot, integrity.value, fingerprints, received_at)
        except Exception:
            # ``ingest`` owns flushes and may be used directly by tests or
            # future callers.  A failed write must never become a speculative
            # predecessor for this source.
            self._previous.pop(source_id, None)
            raise
        if integrity is IntegrityStatus.UNVERIFIABLE:
            # Advance the sliding predecessor only once every database write
            # for this frame has succeeded.  ``handle_frame`` also clears it
            # if the caller's later commit fails.
            self._previous[source_id] = snapshot
        else:
            # A verified frame has no need for a speculative predecessor.
            self._previous.pop(source_id, None)
        return True

    def handle_frame(
        self,
        source_id: int,
        frame: bytes,
        *,
        session_factory: Callable[[], Session] = get_session_local,
        parser: Callable[[bytes], P1Telegram] = parse_telegram,
    ) -> bool:
        """Parse and persist one frame, containing both parse and DB failures.

        A failed write is rolled back before a fresh transaction records only
        a generic source error.  Consequently no partial latest/history update
        survives and the following frame may recover normally.
        """
        received_at = _utc(self._clock())
        try:
            telegram = parser(frame)
        except Exception:
            self._previous.pop(source_id, None)
            self._record_error(
                session_factory, source_id, "WarmteLink frame could not be parsed", now=received_at
            )
            return False
        try:
            with session_factory() as session:
                admitted = self.ingest(
                    session, source_id=source_id, telegram=telegram, received_at=received_at
                )
                session.commit()
                return admitted
        except Exception:
            self._previous.pop(source_id, None)
            self._record_error(session_factory, source_id, "WarmteLink ingest failed", now=received_at)
            return False

    def _admit(
        self,
        session: Session,
        source: MeterSource,
        snapshot: _FrameSnapshot,
        quality: str,
        fingerprints: tuple[str, ...],
        received_at: datetime,
    ) -> None:
        for sample, fingerprint in zip(snapshot.samples, fingerprints, strict=True):
            channel = upsert_discovered_channel(
                session,
                source_id=source.id,
                channel_key=sample.key,
                label=sample.label,
                unit=sample.unit,
                suggested_commodity=_suggestion(sample.unit),
                device_type=sample.device_type,
                fingerprint=fingerprint,
                latest_value=sample.value,
                latest_at=snapshot.recorded_at,
                latest_quality=quality,
            )
            session.flush()
            bucket = snapshot.recorded_at.replace(second=0, microsecond=0)
            exists = session.scalar(
                select(WarmteLinkReading.id)
                .where(
                    WarmteLinkReading.channel_id == channel.id,
                    WarmteLinkReading.recorded_at >= bucket,
                    WarmteLinkReading.recorded_at < bucket + timedelta(minutes=1),
                )
                .limit(1)
            )
            if exists is None:
                session.add(
                    WarmteLinkReading(
                        channel_id=channel.id,
                        recorded_at=snapshot.recorded_at,
                        received_at=received_at,
                        value=sample.value,
                        unit=sample.unit,
                        quality=quality,
                        equipment_fingerprint=fingerprint,
                    )
                )
        source.status = "online"
        source.last_seen_at = received_at
        source.last_error = None
        source.updated_at = received_at

    def _diagnose(self, source: MeterSource, reason: str, *, now: datetime | None = None) -> None:
        now = _utc(self._clock()) if now is None else now
        source.status = "error"
        source.last_error = reason
        source.updated_at = now

    def _record_error(
        self,
        session_factory: Callable[[], Session],
        source_id: int,
        message: str,
        *,
        now: datetime,
    ) -> None:
        try:
            with session_factory() as session:
                source = session.get(MeterSource, source_id)
                if source is not None:
                    self._diagnose(source, message, now=now)
                    session.commit()
        except Exception:
            # Error reporting itself must not kill another source worker.
            return


def _snapshot(telegram: P1Telegram, received_at: datetime) -> _FrameSnapshot:
    recorded_at = _parse_timestamp(telegram.timestamp, received_at)
    samples = tuple(_sample(channel) for channel in telegram.channels)
    if not samples:
        raise ValueError("WarmteLink telegram contains no cumulative channels")
    return _FrameSnapshot(recorded_at, telegram.equipment_fingerprint, samples)


def _sample(channel: P1Channel) -> _ChannelSample:
    profile = _canonical_channel_profile(channel.number)
    if channel.device_type != profile.device_type:
        raise ValueError("WarmteLink channel device type is not canonical")
    if len(channel.readings) != 1:
        raise ValueError("WarmteLink channel has no unambiguous cumulative reading")
    reading = channel.readings[0]
    if reading.code != profile.reading_code or reading.value is None or reading.unit != profile.raw_unit:
        raise ValueError("WarmteLink cumulative reading is incomplete")
    # Parser annotations are not a trust boundary: fake or future parser DTOs
    # must not put a float (including NaN) or a non-finite Decimal into the
    # per-source unverifiable candidate state.  Do not coerce here: accepting
    # another numeric type would make the persistence and continuity paths
    # disagree about the cumulative-value contract.
    if not isinstance(reading.value, Decimal) or not reading.value.is_finite():
        raise ValueError("WarmteLink cumulative reading is not a finite Decimal")
    fingerprint = channel.equipment_fingerprint
    return _ChannelSample(
        key=profile.key,
        label=profile.label,
        value=reading.value,
        unit=profile.unit,
        device_type=profile.device_type,
        fingerprint=fingerprint,
        identity=(channel.number, profile.device_type, fingerprint, profile.reading_code, profile.unit),
    )


@dataclass(frozen=True)
class _CanonicalChannelProfile:
    key: str
    label: str
    device_type: str
    reading_code: str
    raw_unit: str
    unit: str


_CANONICAL_CHANNELS = {
    1: _CanonicalChannelProfile(
        key="channel-1",
        label="WarmteLink channel 1",
        device_type="006",
        reading_code="0-1:24.2.1",
        raw_unit="m3",
        unit="m³",
    ),
    2: _CanonicalChannelProfile(
        key="channel-2",
        label="WarmteLink channel 2",
        device_type="012",
        reading_code="0-2:24.2.1",
        raw_unit="GJ",
        unit="GJ",
    ),
}


def _canonical_channel_profile(number: object) -> _CanonicalChannelProfile:
    # Do not format or coerce parser supplied channel numbers: that could turn
    # an arbitrary object into an identity key before it is rejected.
    if type(number) is not int:
        raise ValueError("WarmteLink channel number is not canonical")
    try:
        return _CANONICAL_CHANNELS[number]
    except KeyError as exc:
        raise ValueError("WarmteLink channel is not supported by the profile") from exc


def _integrity_status(value: object) -> IntegrityStatus:
    """Require a real parser integrity enum, never a look-alike value object."""
    if not isinstance(value, IntegrityStatus):
        raise ValueError("WarmteLink integrity status is not canonical")
    return value


def _continuity_problem(previous: _FrameSnapshot, current: _FrameSnapshot) -> str | None:
    if current.recorded_at <= previous.recorded_at:
        return "Unverifiable WarmteLink timestamp is not strictly increasing"
    if current.recorded_at - previous.recorded_at != timedelta(seconds=10):
        return "Unverifiable WarmteLink frame cadence is not 10 seconds"
    if current.fingerprint != previous.fingerprint:
        return "WarmteLink equipment metadata changed"
    if tuple(sample.identity for sample in current.samples) != tuple(sample.identity for sample in previous.samples):
        return "WarmteLink channel metadata changed or channel set changed"
    old_values = {sample.key: sample.value for sample in previous.samples}
    if any(sample.value < old_values[sample.key] for sample in current.samples):
        return "WarmteLink cumulative value decreased"
    return None


_FINGERPRINT_PATTERN = re.compile(r"[0-9a-f]{64}")


def _final_fingerprints(snapshot: _FrameSnapshot) -> tuple[str, ...] | None:
    """Return only canonical SHA-256 hexdigests safe to persist.

    A parser DTO is an untrusted boundary: even a field named ``fingerprint``
    can contain a raw equipment identifier.  Validate the complete DTO before
    choosing persisted values: the top-level value and every channel value
    must independently be canonical hashes.  This deliberately does not use
    a top-level fallback for an absent or malformed channel fingerprint.
    """
    values = (snapshot.fingerprint, *(sample.fingerprint for sample in snapshot.samples))
    if any(not _is_canonical_fingerprint(value) for value in values):
        return None
    return tuple(sample.fingerprint for sample in snapshot.samples if sample.fingerprint is not None)


def _is_canonical_fingerprint(value: str | None) -> bool:
    return value is not None and _FINGERPRINT_PATTERN.fullmatch(value) is not None


_AMSTERDAM = ZoneInfo("Europe/Amsterdam")
_MAX_CLOCK_SKEW = timedelta(minutes=5)


def _parse_timestamp(value: str | None, received_at: datetime) -> datetime:
    if value is None or len(value) != 13 or value[-1] not in {"S", "W"} or not value[:-1].isdigit():
        raise ValueError("WarmteLink timestamp is unavailable")
    naive = datetime.strptime(value[:-1], "%y%m%d%H%M%S")
    # The S/W marker is only advisory: deployed devices have emitted W while
    # on CEST.  Validate both folds by a UTC round trip, which rejects spring
    # gaps and leaves one (ordinary) or two (fall-back) real instants.
    candidates: list[datetime] = []
    for fold in (0, 1):
        candidate = naive.replace(tzinfo=_AMSTERDAM, fold=fold).astimezone(UTC)
        local = candidate.astimezone(_AMSTERDAM)
        if local.replace(tzinfo=None) == naive and local.fold == fold and candidate not in candidates:
            candidates.append(candidate)
    if not candidates:
        raise ValueError("WarmteLink timestamp is unavailable")
    received_at = _utc(received_at)
    recorded_at = min(candidates, key=lambda candidate: abs(candidate - received_at))
    if abs(recorded_at - received_at) > _MAX_CLOCK_SKEW:
        raise ValueError("WarmteLink timestamp is unavailable")
    return recorded_at


def _suggestion(unit: str) -> str | None:
    return {"GJ": "heating", "m³": "hot_water"}.get(unit)


def _canonical_unit(unit: str) -> str:
    """Map the P1 spelling of cubic metres to the source-profile unit."""
    return "m³" if unit == "m3" else unit


def _utc(value: datetime) -> datetime:
    return value.replace(tzinfo=UTC) if value.tzinfo is None else value.astimezone(UTC)