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

"""OPC write execution (F2 capstone) from a NATS C2 command dict.

Fixes 1879 correctness gaps: safe value coercion (never raises on bad
input), StatusCode/SourceTimestamp fidelity on the DataValue actually sent
to the server, a rate-limit floor between writes, and OPC error typing.
"""

import asyncio
from dataclasses import dataclass
from datetime import datetime

from asyncua import ua
from asyncua.ua.uaerrors import UaStatusCodeError
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

import ingestor.metrics as metrics
from ingestor.connection.clock import MonotonicClock
from ingestor.connection.pool import OpcConnectionPool
from ingestor.converters import ConversionError, to_typed_value
from ingestor.io.data_sink import DataSink

_VARIANT_TYPES = {
    'float': ua.VariantType.Float,
    'double': ua.VariantType.Double,
    'real': ua.VariantType.Double,
    'int': ua.VariantType.Int32,
    'integer': ua.VariantType.Int32,
    'bool': ua.VariantType.Boolean,
    'boolean': ua.VariantType.Boolean,
    'str': ua.VariantType.String,
    'string': ua.VariantType.String,
}

# ponytail: same reconnect-worthy Bad*-status subset as laborious's
# opc_repository.py (different repo, not importable here); keep in sync by
# hand or promote to a shared lib if it drifts.
_SESSION_BAD_NAMES = frozenset(
    {
        'BadSessionIdInvalid',
        'BadSessionClosed',
        'BadSessionNotActivated',
        'BadSecureChannelIdInvalid',
        'BadSecureChannelClosed',
        'BadSecureChannelTokenUnknown',
        'BadTcpSecureChannelUnknown',
        'BadServerNotConnected',
        'BadConnectionClosed',
        'BadConnectionRejected',
        'BadNotConnected',
        'BadTimeout',
        'BadRequestTimeout',
        'BadTooManySessions',
    }
)


@dataclass
class WriteResult:
    ok: bool
    status: str  # 'ok' | 'error' | 'session_bad'
    opc_status_code: str
    message: str


class WriteEngine(SientiaMonitoring):
    """Executes one OPC write from a NATS C2 command."""

    def __init__(
        self,
        pool: OpcConnectionPool,
        sink: DataSink,
        logger: Logger,
        notification_handler: NotificationHandler,
        metrics_controller: MetricsController,
        write_timestamp: bool = True,
        write_quality: bool = True,
        min_pause_s: float = 0.0,
    ):
        self.pool = pool
        self.sink = sink
        self.write_timestamp = write_timestamp
        self.write_quality = write_quality
        self.min_pause_s = min_pause_s
        self.clock = MonotonicClock()
        self._last_write_at: float | None = None

        SientiaMonitoring.__init__(
            self,
            logger=logger,
            notification_handler=notification_handler,
            metrics_controller=metrics_controller,
        )

    async def write(self, cmd: dict) -> WriteResult:
        await self._rate_limit()

        missing = [k for k in ('value', 'data_type', 'server_id', 'addr') if k not in cmd]
        if missing:
            return await self._emit_error(cmd, 'bad_command', f'missing fields: {missing}', 'error')

        try:
            typed = to_typed_value(cmd['value'], cmd['data_type'])
        except ConversionError as e:
            return await self._emit_error(cmd, 'conversion', str(e), 'error')

        dv = self._build_datavalue(cmd, typed)

        start = self.clock.elapsed()
        try:
            client = await self.pool.get(cmd['server_id'], self._server_cfg(cmd))
            node = client.get_node(cmd['addr'])
            await node.write_value(dv)
        except Exception as e:
            status = 'session_bad' if self._is_session_bad(e) else 'error'
            return await self._emit_error(cmd, type(e).__name__, str(e), status)

        return await self._emit_success(cmd, start)

    async def _rate_limit(self) -> None:
        now = self.clock.elapsed()
        if self.min_pause_s > 0 and self._last_write_at is not None:
            remaining = self.min_pause_s - (now - self._last_write_at)
            if remaining > 0:
                await asyncio.sleep(remaining)
                now = self.clock.elapsed()
        self._last_write_at = now

    def _server_cfg(self, cmd: dict) -> dict:
        # ponytail: minimal cfg built straight from the C2 message. Resolving
        # the full connector config (url/security) from the connectors DB by
        # server_id is F4 wiring, not this task — pool is mocked in tests.
        return {
            'url': cmd.get('url'),
            'use_security': cmd.get('use_security', False),
            'server_uri': cmd.get('server_uri'),
        }

    def _variant_type(self, data_type: str) -> ua.VariantType:
        # direct index: unknown dtypes already became ConversionError in to_typed_value
        return _VARIANT_TYPES[data_type.strip().lower()]

    def _build_datavalue(self, cmd: dict, typed_value) -> ua.DataValue:
        variant = ua.Variant(typed_value, self._variant_type(cmd['data_type']))
        write_quality = cmd.get('write_quality', self.write_quality)
        write_timestamp = cmd.get('write_timestamp', self.write_timestamp)
        # NOTE: DataValue.StatusCode_ default_factory populates a Good status
        # even when omitted, so "off" must pass StatusCode_=None explicitly
        # or the flag silently no-ops.
        status_code = ua.StatusCode(ua.StatusCodes.Good) if write_quality else None
        source_ts = datetime.fromisoformat(cmd['timestamp']) if write_timestamp else None
        return ua.DataValue(variant, StatusCode_=status_code, SourceTimestamp=source_ts)

    def _is_session_bad(self, exc: Exception) -> bool:
        return isinstance(exc, UaStatusCodeError) and type(exc).__name__ in _SESSION_BAD_NAMES

    async def _emit_error(self, cmd: dict, error_type: str, message: str, status: str):
        tags = {'pod_id': self.pod_id, 'server_name': cmd.get('server_id', 'unknown')}
        await self.emit_metric(
            metric_object=metrics.OPC_ERROR_BY_TYPE, tags={**tags, 'error_type': error_type}
        )
        await self.emit_metric(
            metric_object=metrics.OPC_WRITE_TOTAL,
            tags={**tags, 'model_id': cmd.get('model_id', 'unknown'), 'result': 'error'},
        )
        result = WriteResult(ok=False, status=status, opc_status_code=error_type, message=message)
        await self.sink.update_write_status(cmd, result)
        return result

    async def _emit_success(self, cmd: dict, start: float):
        tags = {'pod_id': self.pod_id, 'server_name': cmd.get('server_id', 'unknown')}
        await self.emit_metric(
            metric_object=metrics.OPC_WRITE_TOTAL,
            tags={**tags, 'model_id': cmd.get('model_id', 'unknown'), 'result': 'success'},
        )
        await self.emit_metric(
            metric_object=metrics.OPC_WRITE_LATENCY,
            method='observe',
            value=self.clock.elapsed() - start,
            tags=tags,
        )
        await self.emit_metric(
            metric_object=metrics.OPC_WRITE_STATUS_BY_MODEL,
            method='set',
            value=1,
            tags={
                'pod_id': self.pod_id,
                'model_id': cmd.get('model_id', 'unknown'),
                'status': 'ok',
            },
        )
        result = WriteResult(ok=True, status='ok', opc_status_code='Good', message='')
        await self.sink.update_write_status(cmd, result)
        return result