340 lines
13 KiB
Python
340 lines
13 KiB
Python
"""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
|
|
from datetime import UTC, datetime
|
|
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
|
|
|
|
|
|
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
|
|
|
|
|
|
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()
|
|
# 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 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
|
|
framer = TelegramFramer()
|
|
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
|
|
while not self._stop_event.is_set():
|
|
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
|
|
self._ingestor.handle_frame(
|
|
self.source_id, frame, session_factory=self._session_factory
|
|
)
|
|
# A complete frame proves transport recovery even if its
|
|
# contents are rejected by the privacy/admission layer.
|
|
backoff_index = 0
|
|
except Exception:
|
|
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._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 _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()
|