sientia-dataops-opc-ingestor/ingestor/engines/active_read.py

"""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 names:
            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