gitea_admin/sientia-dataops-opc-ingestor
1
0
Fork 0
Code Issues Pull Requests Actions 2 Packages Projects Releases Wiki Activity

fix(gateway): adapt DataValue construction to asyncua 2.x

Browse Source 
main's requirements.txt unpins asyncua (PR #30), so CI installs 2.0.1,
where DataValue's field was renamed StatusCode_ -> StatusCode. The old
kwarg made every OPC write raise TypeError at runtime - masked locally
by a stale asyncua 1.0.6 in the dev conda env. Also uses ua.DateTime
(a datetime subclass) and ua.UInt32 to satisfy the new stricter types,
with no behavior change (verified via binary encode/decode round-trip).
This commit is contained in:
PedroHMCosme
2026-09-22 12:02:29 -03:00
parent 4f4037b4ae
commit 673b13cc6e
4 changed files with 11 additions and 12 deletions
Download Patch File Download Diff File Expand all files Collapse all files

2
ingestor/engines/active_read.py
View File

@@ -89,7 +89,7 @@ class ActiveReadEngine(SientiaMonitoring):
@staticmethod
def _extract(dv):
status_name = dv.StatusCode_.name if dv.StatusCode_ is not None else None
status_name = dv.StatusCode.name if dv.StatusCode is not None else None
return dv.Value.Value, status_name, dv.SourceTimestamp, dv.ServerTimestamp
@staticmethod

11
ingestor/engines/write.py
View File

@@ -7,7 +7,6 @@ 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
@@ -146,12 +145,12 @@ class WriteEngine(SientiaMonitoring):
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
# 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)
status_code = ua.StatusCode(ua.UInt32(ua.StatusCodes.Good)) if write_quality else None
source_ts = ua.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