style: ruff format (new quality gate check picked up from main)
This commit is contained in:
@@ -40,7 +40,11 @@ class ActiveReadEngine(SientiaMonitoring):
|
|||||||
)
|
)
|
||||||
|
|
||||||
async def read_now(
|
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:
|
) -> dict:
|
||||||
"""Read `addr` on `server_id` right now; persist via sink only if `names` given."""
|
"""Read `addr` on `server_id` right now; persist via sink only if `names` given."""
|
||||||
tags = {'pod_id': self.pod_id, 'server_name': server_id}
|
tags = {'pod_id': self.pod_id, 'server_name': server_id}
|
||||||
@@ -72,7 +76,9 @@ class ActiveReadEngine(SientiaMonitoring):
|
|||||||
source_str = self._format_ts(source_ts)
|
source_str = self._format_ts(source_ts)
|
||||||
persisted = bool(names)
|
persisted = bool(names)
|
||||||
if persisted:
|
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 {
|
return {
|
||||||
'value': value,
|
'value': value,
|
||||||
'opc_status_code': status_name,
|
'opc_status_code': status_name,
|
||||||
@@ -96,7 +102,9 @@ class ActiveReadEngine(SientiaMonitoring):
|
|||||||
if source_ts is None:
|
if source_ts is None:
|
||||||
return
|
return
|
||||||
lag = MonotonicClock.now_wall() - source_ts.replace(tzinfo=OPC_TIMEZONE).timestamp()
|
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:
|
async def scheduled_tick(self, slot: dict) -> None:
|
||||||
"""Sweep every server/tag in a parsed slot, reading tags due for an active read."""
|
"""Sweep every server/tag in a parsed slot, reading tags due for an active read."""
|
||||||
|
|||||||
@@ -37,13 +37,24 @@ _VARIANT_TYPES = {
|
|||||||
# ponytail: same reconnect-worthy Bad*-status subset as laborious's
|
# ponytail: same reconnect-worthy Bad*-status subset as laborious's
|
||||||
# opc_repository.py (different repo, not importable here); keep in sync by
|
# opc_repository.py (different repo, not importable here); keep in sync by
|
||||||
# hand or promote to a shared lib if it drifts.
|
# hand or promote to a shared lib if it drifts.
|
||||||
_SESSION_BAD_NAMES = frozenset({
|
_SESSION_BAD_NAMES = frozenset(
|
||||||
'BadSessionIdInvalid', 'BadSessionClosed', 'BadSessionNotActivated',
|
{
|
||||||
'BadSecureChannelIdInvalid', 'BadSecureChannelClosed', 'BadSecureChannelTokenUnknown',
|
'BadSessionIdInvalid',
|
||||||
'BadTcpSecureChannelUnknown', 'BadServerNotConnected', 'BadConnectionClosed',
|
'BadSessionClosed',
|
||||||
'BadConnectionRejected', 'BadNotConnected', 'BadTimeout', 'BadRequestTimeout',
|
'BadSessionNotActivated',
|
||||||
'BadTooManySessions',
|
'BadSecureChannelIdInvalid',
|
||||||
})
|
'BadSecureChannelClosed',
|
||||||
|
'BadSecureChannelTokenUnknown',
|
||||||
|
'BadTcpSecureChannelUnknown',
|
||||||
|
'BadServerNotConnected',
|
||||||
|
'BadConnectionClosed',
|
||||||
|
'BadConnectionRejected',
|
||||||
|
'BadNotConnected',
|
||||||
|
'BadTimeout',
|
||||||
|
'BadRequestTimeout',
|
||||||
|
'BadTooManySessions',
|
||||||
|
}
|
||||||
|
)
|
||||||
|
|
||||||
|
|
||||||
@dataclass
|
@dataclass
|
||||||
@@ -174,7 +185,11 @@ class WriteEngine(SientiaMonitoring):
|
|||||||
metric_object=metrics.OPC_WRITE_STATUS_BY_MODEL,
|
metric_object=metrics.OPC_WRITE_STATUS_BY_MODEL,
|
||||||
method='set',
|
method='set',
|
||||||
value=1,
|
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='')
|
result = WriteResult(ok=True, status='ok', opc_status_code='Good', message='')
|
||||||
await self.sink.update_write_status(cmd, result)
|
await self.sink.update_write_status(cmd, result)
|
||||||
|
|||||||
@@ -57,7 +57,11 @@ class DataSink:
|
|||||||
f'UPDATE {schema}.{table} SET write_status = :status ' # noqa: S608
|
f'UPDATE {schema}.{table} SET write_status = :status ' # noqa: S608
|
||||||
'WHERE timestamp = :timestamp AND model_id = :model_id'
|
'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
|
# 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.
|
# Postgres activities use (e.g. export_data_to_postgres); no async pg driver here.
|
||||||
with self.postgres.engine.begin() as connection:
|
with self.postgres.engine.begin() as connection:
|
||||||
|
|||||||
@@ -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['method'] == 'observe'
|
||||||
assert latency_call.kwargs['value'] >= 0
|
assert latency_call.kwargs['value'] >= 0
|
||||||
lag_call = next(
|
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'
|
assert lag_call.kwargs['method'] == 'set'
|
||||||
|
|
||||||
|
|||||||
Reference in New Issue
Block a user