125 lines
5.1 KiB
Python
125 lines
5.1 KiB
Python
"""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
|