diff --git a/ingestor/engines/active_read.py b/ingestor/engines/active_read.py index 9748517..f42b4f7 100644 --- a/ingestor/engines/active_read.py +++ b/ingestor/engines/active_read.py @@ -40,7 +40,11 @@ class ActiveReadEngine(SientiaMonitoring): ) async def read_now( - self, server_id: str, addr: str, names: list[dict] | None = None, server_cfg: dict | None = None + 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} @@ -72,7 +76,9 @@ class ActiveReadEngine(SientiaMonitoring): 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) + await self.sink.publish_reading( + tag=addr, value=value, timestamp=source_str, names=names + ) return { 'value': value, 'opc_status_code': status_name, @@ -96,7 +102,9 @@ class ActiveReadEngine(SientiaMonitoring): 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) + 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.""" diff --git a/ingestor/engines/write.py b/ingestor/engines/write.py index 2f90d64..3b297d6 100644 --- a/ingestor/engines/write.py +++ b/ingestor/engines/write.py @@ -37,13 +37,24 @@ _VARIANT_TYPES = { # 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', -}) +_SESSION_BAD_NAMES = frozenset( + { + 'BadSessionIdInvalid', + 'BadSessionClosed', + 'BadSessionNotActivated', + 'BadSecureChannelIdInvalid', + 'BadSecureChannelClosed', + 'BadSecureChannelTokenUnknown', + 'BadTcpSecureChannelUnknown', + 'BadServerNotConnected', + 'BadConnectionClosed', + 'BadConnectionRejected', + 'BadNotConnected', + 'BadTimeout', + 'BadRequestTimeout', + 'BadTooManySessions', + } +) @dataclass @@ -174,7 +185,11 @@ class WriteEngine(SientiaMonitoring): 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'}, + 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) diff --git a/ingestor/io/data_sink.py b/ingestor/io/data_sink.py index eef8a50..290f0e4 100644 --- a/ingestor/io/data_sink.py +++ b/ingestor/io/data_sink.py @@ -57,7 +57,11 @@ class DataSink: f'UPDATE {schema}.{table} SET write_status = :status ' # noqa: S608 'WHERE timestamp = :timestamp AND model_id = :model_id' ) - params = {'status': result.status, 'timestamp': cmd['timestamp'], 'model_id': cmd['model_id']} + params = { + 'status': result.status, + 'timestamp': cmd['timestamp'], + 'model_id': cmd['model_id'], + } # ponytail: sync engine call inside async def — same pattern sientia_do's own # Postgres activities use (e.g. export_data_to_postgres); no async pg driver here. with self.postgres.engine.begin() as connection: diff --git a/tests/unit/engines/test_active_read.py b/tests/unit/engines/test_active_read.py index d880a9e..2a7640b 100644 --- a/tests/unit/engines/test_active_read.py +++ b/tests/unit/engines/test_active_read.py @@ -74,7 +74,9 @@ async def test_read_now_persists_and_returns_fields_when_names_given(): 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 + 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'