"""Read-only WarmteLink serial workers and their lifecycle manager. The worker deliberately owns no long-lived SQLAlchemy session and never retains telegram bytes after handing a complete frame to the ingestor. """ from __future__ import annotations from collections.abc import Callable from contextlib import suppress from dataclasses import dataclass, field from datetime import UTC, datetime, timedelta import logging import threading from typing import Protocol import serial from sqlalchemy import select from sqlalchemy.orm import Session from app.db import get_session_local from app.integrations.p1 import TelegramFramer from app.models.meter_source import MeterSource from app.services.warmtelink_ingest import WarmteLinkIngestor logger = logging.getLogger(__name__) _BACKOFF_SECONDS = (1, 2, 4, 8, 16, 32, 60) _JOIN_TIMEOUT_SECONDS = 5 _DISCOVERY_LOCK_TIMEOUT_SECONDS = 0.05 _DISCOVERY_WAIT_SECONDS = 0.1 _DISCOVERY_TIMEOUT_SECONDS = 5 class ReadOnlySerial(Protocol): def read(self, size: int = 1) -> bytes: ... def close(self) -> None: ... SerialFactory = Callable[[dict], ReadOnlySerial] SessionFactory = Callable[[], Session] def _default_session_factory() -> Session: """Resolve the cached sessionmaker at call time, then open one session.""" return get_session_local()() class WorkerClock(Protocol): """Injectable interruptible clock, keeping retry tests deterministic.""" def wait(self, stop_event: threading.Event, seconds: float) -> bool: ... class _EventClock: def wait(self, stop_event: threading.Event, seconds: float) -> bool: return stop_event.wait(seconds) def open_warmtelink_serial(config: dict) -> ReadOnlySerial: """Open the fixed WarmteLink P1 profile; no write-capable API is exposed.""" return serial.Serial( port=config["path"], baudrate=115200, bytesize=serial.SEVENBITS, parity=serial.PARITY_NONE, stopbits=serial.STOPBITS_ONE, timeout=1, ) @dataclass(frozen=True) class _WorkerConfig: source_id: int config: dict @dataclass class DiscoveryRequest: """One source-scoped request, completed only by its serial owner.""" request_id: int source_id: int deadline: datetime status: str = "pending" detail: str | None = None completed: threading.Event = field(default_factory=threading.Event) class WarmteLinkWorker: """One interruptible, read-only serial loop for one meter source.""" def __init__( self, source_id: int, config: dict, *, session_factory: SessionFactory = _default_session_factory, serial_factory: SerialFactory = open_warmtelink_serial, stop_event: threading.Event | None = None, ingestor: WarmteLinkIngestor | None = None, clock: WorkerClock | None = None, ) -> None: self.source_id = source_id self.config = dict(config) self._session_factory = session_factory self._serial_factory = serial_factory self._stop_event = stop_event or threading.Event() self._ingestor = ingestor or WarmteLinkIngestor() self._clock = clock or _EventClock() self._serial: ReadOnlySerial | None = None self._serial_lock = threading.Lock() self._discovery_lock = threading.Lock() self._discoveries: list[DiscoveryRequest] = [] # Never inherit a daemon flag from a caller's background thread: a serial # descriptor and its orderly shutdown must remain visible to the process. self._thread = threading.Thread( target=self._run, name=f"warmtelink-{source_id}", daemon=False ) @property def thread(self) -> threading.Thread: return self._thread def start(self) -> None: self._thread.start() def stop(self) -> None: self._stop_event.set() self._close_serial() def request_discovery(self, request: DiscoveryRequest) -> None: """Queue a read request; this worker remains the sole serial owner.""" with self._discovery_lock: self._discoveries.append(request) def _finish_discoveries(self, status: str, detail: str | None = None) -> None: now = datetime.now(UTC) with self._discovery_lock: pending, self._discoveries = self._discoveries, [] for request in pending: if request.completed.is_set(): continue if request.deadline <= now and status == "completed": request.status, request.detail = "error", "Discovery timed out." else: request.status, request.detail = status, detail request.completed.set() def _expire_discoveries(self) -> None: now = datetime.now(UTC) with self._discovery_lock: expired = [request for request in self._discoveries if request.deadline <= now] self._discoveries = [request for request in self._discoveries if request.deadline > now] for request in expired: if request.completed.is_set(): continue request.status, request.detail = "error", "Discovery timed out." request.completed.set() def join(self, timeout: float = _JOIN_TIMEOUT_SECONDS) -> bool: self._thread.join(timeout) return not self._thread.is_alive() def _close_serial(self) -> None: with self._serial_lock: device, self._serial = self._serial, None if device is not None: with suppress(Exception): device.close() def _record_error(self, message: str) -> None: try: with self._session_factory() as session: source = session.get(MeterSource, self.source_id) if source is not None: source.status = "error" source.last_error = message source.updated_at = datetime.now(UTC) session.commit() except Exception: # A source-status failure must not end another source's worker. return def _run(self) -> None: backoff_index = 0 while not self._stop_event.is_set(): try: device = self._serial_factory(self.config) with self._serial_lock: if self._stop_event.is_set(): with suppress(Exception): device.close() return self._serial = device # A disconnect makes any bytes buffered from the previous # descriptor untrustworthy. In particular, never let a # trailing partial telegram be completed by a newly opened # device. framer = TelegramFramer() while not self._stop_event.is_set(): self._expire_discoveries() chunk = device.read(1024) if not chunk: # ``timeout`` reads are normal, but still yield so a bad # fake/device cannot turn an empty read into a busy spin. self._clock.wait(self._stop_event, 0.05) continue frames = framer.feed(chunk) for frame in frames: if self._stop_event.is_set(): break admitted = self._ingestor.handle_frame( self.source_id, frame, session_factory=self._session_factory ) if admitted: self._finish_discoveries("completed") # A complete frame proves transport recovery even if its # contents are rejected by the privacy/admission layer. backoff_index = 0 except Exception: self._finish_discoveries("error", "WarmteLink discovery failed.") self._record_error("WarmteLink serial connection failed") delay = _BACKOFF_SECONDS[min(backoff_index, len(_BACKOFF_SECONDS) - 1)] backoff_index += 1 self._clock.wait(self._stop_event, delay) finally: self._close_serial() class WarmteLinkWorkerManager: """Reconcile enabled serial sources into exactly one worker each.""" def __init__( self, *, session_factory: SessionFactory = _default_session_factory, serial_factory: SerialFactory = open_warmtelink_serial, worker_factory: Callable[..., WarmteLinkWorker] = WarmteLinkWorker, ) -> None: self._session_factory = session_factory self._serial_factory = serial_factory self._worker_factory = worker_factory self._workers: dict[int, tuple[_WorkerConfig, WarmteLinkWorker]] = {} self._lock = threading.Lock() self._reapers: set[int] = set() self._next_discovery_id = 0 self._shutting_down = False @property def worker_count(self) -> int: with self._lock: return len(self._workers) def reconcile(self) -> None: # Reading desired state under the same lock which applies it prevents a # delayed pre-commit snapshot from rolling a newer commit backwards. with self._lock: if self._shutting_down: return desired = self._read_desired() self._reconcile_locked(desired) def start(self) -> None: """Enable reconciliation for a newly entered application lifespan.""" with self._lock: self._shutting_down = False self.reconcile() def request_discovery(self, source_id: int) -> DiscoveryRequest: """Ask the current source worker for one bounded read/discovery attempt. This intentionally does not reconcile or open a descriptor. Lifecycle convergence remains separate; a request can neither replace nor stop a worker when an HTTP client times out or disconnects. """ now = datetime.now(UTC) request = DiscoveryRequest(0, source_id, now) if not self._lock.acquire(timeout=_DISCOVERY_LOCK_TIMEOUT_SECONDS): request.status, request.detail = "error", "Discovery queue is busy." request.completed.set() return request try: self._next_discovery_id += 1 request.request_id = self._next_discovery_id request.deadline = now + timedelta(seconds=_DISCOVERY_TIMEOUT_SECONDS) if self._shutting_down: request.status, request.detail = "error", "WarmteLink manager is stopped." request.completed.set() elif (entry := self._workers.get(source_id)) is None: request.status, request.detail = "error", "WarmteLink worker is not running." request.completed.set() else: entry[1].request_discovery(request) timer = threading.Timer(_DISCOVERY_TIMEOUT_SECONDS, self._timeout_discovery, args=(request,)) timer.daemon = True timer.start() finally: self._lock.release() # A tiny bounded wait makes an immediately available frame observable, # without turning an HTTP call into serial I/O or an unbounded wait. request.completed.wait(_DISCOVERY_WAIT_SECONDS) return request @staticmethod def _timeout_discovery(request: DiscoveryRequest) -> None: """Resolve a stale HTTP request without touching its healthy worker.""" if not request.completed.is_set(): request.status, request.detail = "error", "Discovery timed out." request.completed.set() def _read_desired(self) -> dict[int, _WorkerConfig]: with self._session_factory() as session: return { source.id: _WorkerConfig(source.id, dict(source.config)) for source in session.execute( select(MeterSource).where( MeterSource.kind == "warmtelink_serial", MeterSource.enabled.is_(True) ) ).scalars() } def _record_manager_error(self, source_id: int) -> None: """Best-effort, deliberately non-sensitive lifecycle failure status.""" try: with self._session_factory() as session: source = session.get(MeterSource, source_id) if source is not None: source.status = "error" source.last_error = "WarmteLink worker failed" source.updated_at = datetime.now(UTC) session.commit() except Exception: return def _reconcile_locked(self, desired: dict[int, _WorkerConfig]) -> None: stale = [ source_id for source_id, (config, _) in self._workers.items() if source_id not in desired or desired[source_id] != config ] blocked: set[int] = set() for source_id in stale: _, worker = self._workers[source_id] try: worker.stop() stopped = worker.join() except Exception: self._record_manager_error(source_id) blocked.add(source_id) continue if stopped: self._workers.pop(source_id, None) else: logger.error("WarmteLink worker did not stop for source %s", source_id) blocked.add(source_id) self._schedule_reaper_locked(source_id, worker) for source_id, config in desired.items(): if source_id in self._workers or source_id in blocked: continue worker: WarmteLinkWorker | None = None try: worker = self._worker_factory( source_id, config.config, session_factory=self._session_factory, serial_factory=self._serial_factory, ) self._workers[source_id] = (config, worker) worker.start() except Exception: self._record_manager_error(source_id) # A failed start normally has no thread. If an unusual worker # did start before raising, keep it tracked until it is reaped. if not self._worker_is_alive(worker): self._workers.pop(source_id, None) else: self._schedule_reaper_locked(source_id, worker) @staticmethod def _worker_is_alive(worker: object | None) -> bool: thread = getattr(worker, "thread", None) return bool(thread is not None and thread.is_alive()) def _schedule_reaper_locked(self, source_id: int, worker: WarmteLinkWorker) -> None: if source_id in self._reapers: return self._reapers.add(source_id) threading.Thread( target=self._reap_worker, args=(source_id, worker), # This bookkeeping watcher must not turn a deliberately bounded # application shutdown into an unbounded process wait. The actual # serial worker itself is explicitly non-daemon. name=f"warmtelink-reaper-{source_id}", daemon=True, ).start() def _reap_worker(self, source_id: int, worker: WarmteLinkWorker) -> None: """Wait for one timed-out worker, then converge without another API call.""" try: while True: try: if worker.join(): break except Exception: self._record_manager_error(source_id) return # A custom worker can report a bounded join timeout immediately; # yield before asking again so its reaper cannot busy-spin. threading.Event().wait(0.05) with self._lock: current = self._workers.get(source_id) if current is not None and current[1] is worker: self._workers.pop(source_id) self._reapers.discard(source_id) should_reconcile = not self._shutting_down if should_reconcile: self.reconcile() finally: with self._lock: self._reapers.discard(source_id) def shutdown(self) -> None: with self._lock: self._shutting_down = True workers = list(self._workers.items()) for source_id, (_, worker) in workers: try: worker.stop() except Exception: self._record_manager_error(source_id) for source_id, (_, worker) in workers: try: stopped = worker.join() except Exception: self._record_manager_error(source_id) continue if not stopped: logger.error("WarmteLink worker did not stop during shutdown for source %s", source_id) with self._lock: self._schedule_reaper_locked(source_id, worker) else: with self._lock: current = self._workers.get(source_id) if current is not None and current[1] is worker: self._workers.pop(source_id) warmtelink_worker_manager = WarmteLinkWorkerManager()