feat(gateway): ActiveReadEngine (scheduled + read_now)

ingestor/engines/active_read.py

@@ -0,0 +1,124 @@
+"""Active OPC reads (F3): on-demand (`read_now`, used by the gRPC server -
+Task 16) and scheduled (`scheduled_tick`, driven by the app main loop each
+collect cycle). Emits metrics only -- unlike WriteEngine there is no
+notification path here."""
+
+from sientia_do.notifications.handlers import CoreNotificationHandler as NotificationHandler
+from sientia_do.observability.logger import Logger
+from sientia_do.observability.metrics_controller import MetricsController
+from sientia_do.observability.sientia_monitoring import SientiaMonitoring
+from sientia_do.temporal.constants import DATETIME_FORMAT_WITH_TZ, OPC_TIMEZONE
+
+import ingestor.metrics as metrics
+from ingestor.connection.clock import MonotonicClock
+from ingestor.connection.pool import OpcConnectionPool
+from ingestor.io.data_sink import DataSink
+
+
+class ActiveReadEngine(SientiaMonitoring):
+    """On-demand + scheduled OPC reads, sharing the pool/sink with WriteEngine
+    and SubscriptionEngine."""
+
+    def __init__(
+        self,
+        pool: OpcConnectionPool,
+        sink: DataSink,
+        logger: Logger,
+        notification_handler: NotificationHandler,
+        metrics_controller: MetricsController,
+    ):
+        self.pool = pool
+        self.sink = sink
+        self.clock = MonotonicClock()
+        self._ticks: dict[tuple[str, str], int] = {}
+
+        SientiaMonitoring.__init__(
+            self,
+            logger=logger,
+            notification_handler=notification_handler,
+            metrics_controller=metrics_controller,
+        )
+
+    async def read_now(
+        self, server_id: str, addr: str, names: list[dict] | None = None, server_cfg: dict | None = None
+    ) -> dict:
+        """Read `addr` on `server_id` right now; persist via sink only if `names` given."""
+        tags = {'pod_id': self.pod_id, 'server_name': server_id}
+        start = self.clock.elapsed()
+        try:
+            client = await self.pool.get(server_id, server_cfg or {})
+            dv = await client.get_node(addr).read_data_value()
+        except Exception as e:
+            await self.emit_metric(
+                metric_object=metrics.OPC_ACTIVE_READ_TOTAL, tags={**tags, 'result': 'error'}
+            )
+            self.logger.error(f'Active read failed for {server_id}:{addr}: {e}')
+            raise
+
+        await self.emit_metric(
+            metric_object=metrics.OPC_ACTIVE_READ_LATENCY,
+            method='observe',
+            value=self.clock.elapsed() - start,
+            tags=tags,
+        )
+        await self.emit_metric(
+            metric_object=metrics.OPC_ACTIVE_READ_TOTAL, tags={**tags, 'result': 'success'}
+        )
+        return await self._build_result(dv, addr, names, tags)
+
+    async def _build_result(self, dv, addr: str, names: list[dict] | None, tags: dict) -> dict:
+        value, status_name, source_ts, server_ts = self._extract(dv)
+        await self._record_read_lag(source_ts, tags)
+        source_str = self._format_ts(source_ts)
+        persisted = bool(names)
+        if persisted:
+            await self.sink.publish_reading(tag=addr, value=value, timestamp=source_str, names=names)
+        return {
+            'value': value,
+            'opc_status_code': status_name,
+            'source_timestamp': source_str,
+            'server_timestamp': self._format_ts(server_ts),
+            'persisted': persisted,
+        }
+
+    @staticmethod
+    def _extract(dv):
+        status_name = dv.StatusCode_.name if dv.StatusCode_ is not None else None
+        return dv.Value.Value, status_name, dv.SourceTimestamp, dv.ServerTimestamp
+
+    @staticmethod
+    def _format_ts(ts) -> str | None:
+        if ts is None:
+            return None
+        return ts.replace(tzinfo=OPC_TIMEZONE).strftime(DATETIME_FORMAT_WITH_TZ)
+
+    async def _record_read_lag(self, source_ts, tags: dict) -> None:
+        if source_ts is None:
+            return
+        lag = MonotonicClock.now_wall() - source_ts.replace(tzinfo=OPC_TIMEZONE).timestamp()
+        await self.emit_metric(metric_object=metrics.OPC_READ_LAG, method='set', value=lag, tags=tags)
+
+    async def scheduled_tick(self, slot: dict) -> None:
+        """Sweep every server/tag in a parsed slot, reading tags due for an active read."""
+        for server in slot.values():
+            server_id = server['server_id']
+            server_cfg = {k: v for k, v in server.items() if k != 'tags'}
+            for addr, tag in server['tags'].items():
+                await self._tick_tag(server_id, addr, tag, server_cfg)
+
+    async def _tick_tag(self, server_id: str, addr: str, tag: dict, server_cfg: dict) -> None:
+        active_read = tag.get('active_read') or {}
+        if not active_read.get('enabled'):
+            return
+        if not self._should_read(server_id, addr, active_read.get('cycle_multiple', 1)):
+            return
+        try:
+            await self.read_now(server_id, addr, names=tag['names'], server_cfg=server_cfg)
+        except Exception as e:
+            self.logger.error(f'Scheduled active read failed for {server_id}:{addr}: {e}')
+
+    def _should_read(self, server_id: str, addr: str, cycle_multiple: int) -> bool:
+        key = (server_id, addr)
+        count = self._ticks.get(key, 0) + 1
+        self._ticks[key] = count
+        return count % max(cycle_multiple, 1) == 0