diff --git a/ingestor/engines/active_read.py b/ingestor/engines/active_read.py new file mode 100644 index 0000000..9748517 --- /dev/null +++ b/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 diff --git a/tests/unit/engines/test_active_read.py b/tests/unit/engines/test_active_read.py new file mode 100644 index 0000000..d880a9e --- /dev/null +++ b/tests/unit/engines/test_active_read.py @@ -0,0 +1,177 @@ +from datetime import datetime +from unittest.mock import AsyncMock, MagicMock + +import pytest +from asyncua import ua + +import ingestor.metrics as metrics +from ingestor.engines.active_read import ActiveReadEngine + + +def make_engine(): + eng = ActiveReadEngine( + pool=AsyncMock(), + sink=AsyncMock(), + logger=MagicMock(), + notification_handler=MagicMock(), + metrics_controller=AsyncMock(), + ) + eng.emit_metric = AsyncMock() + return eng + + +def make_dv(value=42.0, status='Good', source_ts=None, server_ts=None): + source_ts = source_ts or datetime(2026, 7, 13, 12, 0, 0) + server_ts = server_ts or source_ts + return ua.DataValue( + ua.Variant(value), + StatusCode_=ua.StatusCode(getattr(ua.StatusCodes, status)), + SourceTimestamp=source_ts, + ServerTimestamp=server_ts, + ) + + +def make_client(dv=None): + node = MagicMock() + node.read_data_value = AsyncMock(return_value=dv or make_dv()) + client = MagicMock() + client.get_node = MagicMock(return_value=node) + return client, node + + +NAMES = [{'tag_name': 'a', 'model_id': 'm1', 'collection': 'coll'}] + + +@pytest.mark.asyncio +async def test_read_now_persists_and_returns_fields_when_names_given(): + eng = make_engine() + client, node = make_client() + eng.pool.get.return_value = client + + result = await eng.read_now('srv1', 'ns=2;i=5', names=NAMES) + + client.get_node.assert_called_once_with('ns=2;i=5') + node.read_data_value.assert_awaited_once() + eng.sink.publish_reading.assert_awaited_once() + _, kwargs = eng.sink.publish_reading.call_args + assert kwargs['tag'] == 'ns=2;i=5' + assert kwargs['value'] == 42.0 + assert kwargs['names'] == NAMES + assert result['value'] == 42.0 + assert result['opc_status_code'] == 'Good' + assert result['persisted'] is True + assert result['source_timestamp'].startswith('2026-07-13 12:00:00') + assert result['server_timestamp'].startswith('2026-07-13 12:00:00') + eng.emit_metric.assert_any_await( + metric_object=metrics.OPC_ACTIVE_READ_TOTAL, + tags={'pod_id': eng.pod_id, 'server_name': 'srv1', 'result': 'success'}, + ) + latency_call = next( + c + for c in eng.emit_metric.await_args_list + if c.kwargs['metric_object'] is metrics.OPC_ACTIVE_READ_LATENCY + ) + assert latency_call.kwargs['method'] == 'observe' + assert latency_call.kwargs['value'] >= 0 + lag_call = next( + c for c in eng.emit_metric.await_args_list if c.kwargs['metric_object'] is metrics.OPC_READ_LAG + ) + assert lag_call.kwargs['method'] == 'set' + + +@pytest.mark.asyncio +async def test_read_now_without_names_does_not_persist(): + eng = make_engine() + client, _ = make_client() + eng.pool.get.return_value = client + + result = await eng.read_now('srv1', 'ns=2;i=5') + + eng.sink.publish_reading.assert_not_awaited() + assert result['persisted'] is False + + +@pytest.mark.asyncio +async def test_read_now_error_emits_metric_and_reraises(): + eng = make_engine() + node = MagicMock() + node.read_data_value = AsyncMock(side_effect=RuntimeError('boom')) + client = MagicMock() + client.get_node = MagicMock(return_value=node) + eng.pool.get.return_value = client + + with pytest.raises(RuntimeError, match='boom'): + await eng.read_now('srv1', 'ns=2;i=5') + + eng.emit_metric.assert_any_await( + metric_object=metrics.OPC_ACTIVE_READ_TOTAL, + tags={'pod_id': eng.pod_id, 'server_name': 'srv1', 'result': 'error'}, + ) + eng.sink.publish_reading.assert_not_awaited() + + +def make_slot(cycle_multiple=None, enabled=True): + active_read = {'enabled': enabled} + if cycle_multiple is not None: + active_read['cycle_multiple'] = cycle_multiple + return { + 'srv_name': { + 'server_id': 'srv1', + 'url': 'opc.tcp://x', + 'tags': { + 'ns=2;i=5': {'frequency': 1000, 'names': NAMES, 'active_read': active_read}, + }, + } + } + + +@pytest.mark.asyncio +async def test_scheduled_tick_respects_cycle_multiple(): + eng = make_engine() + eng.read_now = AsyncMock() + slot = make_slot(cycle_multiple=3) + + await eng.scheduled_tick(slot) + await eng.scheduled_tick(slot) + eng.read_now.assert_not_awaited() + await eng.scheduled_tick(slot) + eng.read_now.assert_awaited_once_with( + 'srv1', 'ns=2;i=5', names=NAMES, server_cfg={'server_id': 'srv1', 'url': 'opc.tcp://x'} + ) + + +@pytest.mark.asyncio +async def test_scheduled_tick_skips_disabled_tags(): + eng = make_engine() + eng.read_now = AsyncMock() + slot = make_slot(enabled=False) + + await eng.scheduled_tick(slot) + + eng.read_now.assert_not_awaited() + + +@pytest.mark.asyncio +async def test_scheduled_tick_continues_past_tag_error(): + eng = make_engine() + bad_slot = make_slot(cycle_multiple=1) + good_names = [{'tag_name': 'b', 'model_id': 'm2', 'collection': 'coll'}] + bad_slot['srv_name']['tags']['ns=2;i=6'] = { + 'frequency': 1000, + 'names': good_names, + 'active_read': {'enabled': True, 'cycle_multiple': 1}, + } + + calls = [] + + async def fake_read_now(server_id, addr, names=None, server_cfg=None): + calls.append(addr) + if addr == 'ns=2;i=5': + raise RuntimeError('boom') + + eng.read_now = AsyncMock(side_effect=fake_read_now) + + await eng.scheduled_tick(bad_slot) + + assert set(calls) == {'ns=2;i=5', 'ns=2;i=6'} + eng.logger.error.assert_called_once()