402 lines
16 KiB
Python
402 lines
16 KiB
Python
"""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)
|