266 lines
15 KiB
Python
266 lines
15 KiB
Python
"""Verify monitoring charts against observations and require run-local device coverage."""
|
||
from __future__ import annotations
|
||
|
||
from copy import deepcopy
|
||
from datetime import datetime, timedelta, timezone
|
||
from math import isclose
|
||
import re
|
||
|
||
from app.domain.domain_errors import require
|
||
|
||
|
||
DISPLAY_TIMEZONE = timezone(timedelta(hours=8))
|
||
METRICS = {"pressure": "pressure", "pipe_flow": "flow", "flow": "flow"}
|
||
|
||
|
||
def applies(task: dict) -> bool:
|
||
if task.get("workflow") != "scada_anomaly" and not task.get("source", {}).get("replay_context"):
|
||
return False
|
||
# Recovery has its own Server-verified post-repair evidence and finalization gate.
|
||
return (task.get("handling_context") or {}).get("phase") not in {"verification_ready", "completed"}
|
||
|
||
|
||
def cutoff(task: dict) -> datetime:
|
||
replay = task.get("source", {}).get("replay_context") or {}
|
||
return datetime.fromisoformat(replay.get("knowledge_cutoff") or task["trigger"]["occurred_at"])
|
||
|
||
|
||
def _unit(value: str) -> str:
|
||
return value.replace("³", "3").replace(" ", "").lower()
|
||
|
||
|
||
class ScadaObservations:
|
||
def __init__(self, task: dict, evidence: list[dict], refs: list[str] | None = None):
|
||
self.devices: dict[str, dict] = {}
|
||
self.values: dict[tuple[str, datetime], dict] = {}
|
||
self.windows: list[tuple[set[str], set[str], datetime, datetime]] = []
|
||
self.frozen_limits: list[tuple[set[str], datetime]] = []
|
||
target = cutoff(task)
|
||
replay = task.get("source", {}).get("replay_context") or {}
|
||
for item in evidence:
|
||
snapshot = item.get("data_snapshot") or {}
|
||
result = snapshot.get("data") or {}
|
||
if snapshot.get("source") != "query_scada_readings" or not isinstance(result, dict):
|
||
continue
|
||
data = result.get("data")
|
||
coverage = data.get("coverage", {}) if isinstance(data, dict) else {}
|
||
if replay and coverage and (coverage.get("replay_id") != replay["replay_id"]
|
||
or coverage.get("batch_sequence") != replay["batch_sequence"]):
|
||
continue
|
||
window = result.get("window") or {}
|
||
if not window.get("start") or not window.get("end"):
|
||
continue
|
||
start, end = (datetime.fromisoformat(window[key]) for key in ("start", "end"))
|
||
if end > target:
|
||
continue
|
||
devices = data.get("devices", []) if isinstance(data, dict) else []
|
||
for device in devices:
|
||
self.devices[str(device["device_id"])] = device
|
||
if refs is not None and item["evidence_id"] not in refs:
|
||
continue
|
||
rows = data.get("readings", []) if isinstance(data, dict) else data
|
||
if not isinstance(rows, list):
|
||
continue
|
||
ids = {str(row["device_id"]) for row in rows if "device_id" in row}
|
||
outside = set(coverage.get("outside_frozen_scope_device_ids", []))
|
||
if coverage.get("query_complete") is True:
|
||
ids.update(coverage.get("queried_device_ids", []))
|
||
ids.update(str(device["device_id"]) for device in devices)
|
||
page = result.get("page") or {}
|
||
complete = not (snapshot.get("truncated") or result.get("truncated")
|
||
or page.get("has_more") or page.get("offset", 0))
|
||
if complete:
|
||
self.windows.append((ids, outside, start, end))
|
||
if replay and coverage.get("query_complete") and coverage.get("frozen_start"):
|
||
self.frozen_limits.append((ids, datetime.fromisoformat(coverage["frozen_start"])))
|
||
for row in rows:
|
||
if "time" in row and "device_id" in row:
|
||
at = datetime.fromisoformat(row["time"])
|
||
if start <= at <= end:
|
||
self.values[(str(row["device_id"]), at)] = row
|
||
|
||
def absent_reason(self, device: str, at: datetime) -> str | None:
|
||
for ids, outside, start, end in self.windows:
|
||
if device in ids and start <= at <= end:
|
||
return "outside_frozen_scope" if device in outside else "no_readings"
|
||
for ids, frozen_start in self.frozen_limits:
|
||
if device in ids and at < frozen_start:
|
||
return "outside_frozen_window"
|
||
return None
|
||
|
||
def window_is_empty(self, device: str, start: datetime, end: datetime) -> bool:
|
||
if any(key == device and start <= at <= end for key, at in self.values):
|
||
return False
|
||
intervals = [(left, right) for ids, _, left, right in self.windows if device in ids]
|
||
intervals.extend((start, frozen_start) for ids, frozen_start in self.frozen_limits if device in ids)
|
||
through = start
|
||
for left, right in sorted(intervals):
|
||
if left > through:
|
||
break
|
||
through = max(through, right)
|
||
if through >= end:
|
||
return True
|
||
return False
|
||
|
||
def period(self, device: str, start: datetime, end: datetime) -> timedelta:
|
||
frequency = str(self.devices[device].get("transmission_frequency") or "")
|
||
match = re.fullmatch(r"(\d+):([0-5]\d):([0-5]\d)", frequency)
|
||
if match:
|
||
hours, minutes, seconds = map(int, match.groups())
|
||
period = timedelta(hours=hours, minutes=minutes, seconds=seconds)
|
||
else:
|
||
minutes = re.fullmatch(r"(\d+)\s*(?:min|分钟)", frequency)
|
||
times = sorted(at for key, at in self.values if key == device and start <= at <= end)
|
||
period = (timedelta(minutes=int(minutes[1])) if minutes else
|
||
min((right - left for left, right in zip(times, times[1:])), default=timedelta(0)))
|
||
require(period > timedelta(0), "SCADA_CHART_EVIDENCE_REQUIRED",
|
||
f"设备 {device} 缺少可核实的采样周期，请读取完整设备属性", refs=[device])
|
||
return period
|
||
|
||
def slots(self, device: str, start: datetime, end: datetime) -> list[datetime]:
|
||
"""Keep the business bounds exact, aligning slots to the observed sampling phase."""
|
||
period = self.period(device, start, end)
|
||
observed = sorted(at for key, at in self.values if key == device and start <= at <= end)
|
||
historical = sorted(at for key, at in self.values if key == device and at < start)
|
||
# Referenced historical observations preserve phase even if current data is absent.
|
||
anchor = (observed[0] if observed else historical[-1] if historical else
|
||
start.astimezone(DISPLAY_TIMEZONE).replace(hour=0, minute=0, second=0, microsecond=0))
|
||
first = start + (anchor - start) % period
|
||
return [first + period * index for index in range(int((end - first) // period) + 1)]
|
||
|
||
|
||
def add_targets(document: dict | None, devices: list[str], observations: ScadaObservations) -> dict:
|
||
result = deepcopy(document) if document else {"version": 1, "targets": [], "charts": {}}
|
||
targets = set(result["targets"])
|
||
for device in devices:
|
||
require(device in observations.devices, "SCADA_CHART_EVIDENCE_REQUIRED",
|
||
f"设备 {device} 未出现在本轮截止时间内的完整设备证据中", refs=[device])
|
||
if observations.devices[device].get("device_type") in METRICS:
|
||
targets.add(device)
|
||
result["targets"] = sorted(targets)
|
||
return result
|
||
|
||
|
||
def verify_chart(task: dict, chart: dict, evidence: list[dict], refs: list[str]) -> dict:
|
||
"""Return verified coverage; mutate only the outgoing chart's Server-derived gaps."""
|
||
context = chart["scada"]
|
||
end = cutoff(task)
|
||
start = end - timedelta(hours=3)
|
||
require(datetime.fromisoformat(context["window_start"]) == start
|
||
and datetime.fromisoformat(context["window_end"]) == end,
|
||
"SCADA_CHART_WINDOW_INVALID", "SCADA图表必须使用本轮截止时刻之前的完整三小时窗口")
|
||
observations = ScadaObservations(task, evidence, refs)
|
||
series = chart["series"]
|
||
ids = {item.get("device_id") for item in series}
|
||
require(None not in ids and all(device in observations.devices for device in ids),
|
||
"SCADA_CHART_EVIDENCE_REQUIRED", "每条SCADA序列须绑定已核实的 device_id", refs=sorted(ids - {None}))
|
||
comparison = context["role"] == "comparison"
|
||
require(chart["chart_type"] == ("time_comparison" if comparison else "line"),
|
||
"SCADA_CHART_INVALID", "SCADA总览须使用line，同期图须使用time_comparison")
|
||
bindings = [(item["device_id"], item.get("day_offset")) for item in series]
|
||
require(len(bindings) == len(set(bindings)), "SCADA_CHART_INVALID", "SCADA序列的设备与日期绑定不能重复")
|
||
require((len(ids) == 1 and {day for _, day in bindings} == set(range(8))) if comparison
|
||
else all(day == 0 for _, day in bindings),
|
||
"SCADA_CHART_INVALID", "总览每个设备绑定day_offset=0；同期图每个设备须有当天及此前七天共8组序列")
|
||
field = context.get("value_field", "monitored_value")
|
||
gaps = []
|
||
coverage = {}
|
||
for item in series:
|
||
device, day = item["device_id"], item["day_offset"]
|
||
metadata = observations.devices[device]
|
||
require(METRICS.get(metadata.get("device_type")) == context["metric"]
|
||
and _unit(metadata.get("measurement_unit") or "") == _unit(context["unit"]),
|
||
"SCADA_CHART_INVALID", f"设备 {device} 的指标或单位与完整设备证据不一致", refs=[device])
|
||
period = observations.period(device, start, end)
|
||
full_slots = observations.slots(device, start, end)
|
||
no_schedule = not full_slots
|
||
if no_schedule:
|
||
require(observations.window_is_empty(device, start - timedelta(days=day), end - timedelta(days=day)),
|
||
"SCADA_CHART_EVIDENCE_REQUIRED", f"设备 {device} 无计划采样槽，须完整读取对应三小时窗口并证明无读数", refs=[device])
|
||
# This is an empty display entry, never an observation or a missed scheduled sample.
|
||
full_slots = [end]
|
||
labels = [at.astimezone(DISPLAY_TIMEZONE).strftime("%H:%M:%S" if period.total_seconds() % 60 else "%H:%M")
|
||
for at in full_slots]
|
||
first = labels.index(chart["x_data"][0]) if chart["x_data"][0] in labels else -1
|
||
count = len(chart["x_data"])
|
||
require(first >= 0 and chart["x_data"] == labels[first:first + count]
|
||
and len(item.get("sample_times") or []) == count,
|
||
"SCADA_CHART_INVALID", f"设备 {device} 须保留连续真实采样槽和sample_times；超限按时间分图", refs=[device])
|
||
slots = full_slots[first:first + count]
|
||
coverage[device] = {"covered": [at.isoformat() for at in slots]}
|
||
require((item.get("type") or "line") == ("scatter" if day else "line"),
|
||
"SCADA_CHART_INVALID", "当前序列须为line，历史序列须为scatter")
|
||
missing = 0
|
||
reason = "no_readings"
|
||
expected = {at - timedelta(days=day) for at in (full_slots[-3:] if day else full_slots)}
|
||
# Off-grid observations cannot be silently discarded by a regular-looking chart.
|
||
observed = {at for key, at in observations.values if key == device
|
||
and min(expected) <= at <= max(expected)}
|
||
require(observed <= expected, "SCADA_CHART_INVALID",
|
||
f"设备 {device} 含非规则采样时刻，须明确处理，不能静默丢点", refs=[device])
|
||
for index, slot in enumerate(slots):
|
||
at = slot - timedelta(days=day)
|
||
value, sample_time = item["data"][index], item["sample_times"][index]
|
||
if at not in expected:
|
||
require(value is None and sample_time is None, "SCADA_CHART_INVALID",
|
||
"历史散点只保留每日末三个对应时刻")
|
||
continue
|
||
record = observations.values.get((device, at))
|
||
actual = record.get(field) if record is not None else None
|
||
if actual is None:
|
||
absent = observations.absent_reason(device, at)
|
||
require(record is not None or absent is not None, "SCADA_CHART_EVIDENCE_REQUIRED",
|
||
f"设备 {device} 的 {at.isoformat()} 未被完整读取，不能将未查询当作缺测", refs=[device])
|
||
require(value is None and sample_time is None, "SCADA_CHART_DATA_MISMATCH",
|
||
f"设备 {device} 缺测槽必须保留null，不能补零或推算", refs=[device])
|
||
missing += 1
|
||
if absent in {"outside_frozen_scope", "outside_frozen_window"}:
|
||
reason = absent
|
||
elif no_schedule:
|
||
reason = "no_scheduled_samples"
|
||
else:
|
||
require(value is not None and sample_time is not None
|
||
and datetime.fromisoformat(sample_time) == at
|
||
and (isclose(value, actual, rel_tol=1e-9, abs_tol=1e-9)
|
||
or any(isclose(value, round(actual, digits), rel_tol=0, abs_tol=1e-9)
|
||
for digits in range(1, 7))),
|
||
"SCADA_CHART_DATA_MISMATCH",
|
||
f"设备 {device} 的 {at.isoformat()} 图表值或采样时间与证据不符", refs=[device])
|
||
if missing:
|
||
gaps.append({"device_id": device, "day_offset": day, "missing_count": missing, "reason": reason})
|
||
context["gaps"] = gaps
|
||
return {"role": context["role"], "metric": context["metric"], "device_ids": sorted(ids),
|
||
"window_start": context["window_start"], "window_end": context["window_end"], "gaps": gaps,
|
||
"coverage": coverage}
|
||
|
||
|
||
def require_complete(task: dict, document: dict | None, observations: ScadaObservations) -> None:
|
||
targets = set((document or {}).get("targets", []))
|
||
supported = {key for key, device in observations.devices.items() if device.get("device_type") in METRICS}
|
||
if not supported and not targets:
|
||
return
|
||
require(bool(targets), "SCADA_PRESENTATION_INCOMPLETE",
|
||
"请声明本轮SCADA目标设备并发布当前总览与七天同期图；旧格式图表不计入设备覆盖", status=409)
|
||
end = cutoff(task)
|
||
start = end - timedelta(hours=3)
|
||
covered = {"current": {}, "comparison": {}}
|
||
for chart in document["charts"].values():
|
||
if (datetime.fromisoformat(chart["window_start"]) == start
|
||
and datetime.fromisoformat(chart["window_end"]) == end):
|
||
for device, slots in chart.get("coverage", {}).items():
|
||
if device not in chart["device_ids"]:
|
||
continue
|
||
covered[chart["role"]].setdefault(device, set()).update(
|
||
datetime.fromisoformat(at) for at in slots["covered"])
|
||
missing = {}
|
||
for role in covered:
|
||
missing[role] = sorted(device for device in targets
|
||
if not covered[role].get(device) or not set(observations.slots(device, start, end) or [end])
|
||
<= covered[role][device])
|
||
missing_current, missing_comparison = missing["current"], missing["comparison"]
|
||
require(not missing_current and not missing_comparison, "SCADA_PRESENTATION_INCOMPLETE",
|
||
f"本轮SCADA目标共{len(targets)}个；缺少当前总览设备：{', '.join(missing_current) or '无'}；"
|
||
f"缺少七天同期对比设备：{', '.join(missing_comparison) or '无'}。请补图后重试；仅作参考不能豁免展示。",
|
||
refs=sorted(set(missing_current + missing_comparison)), status=409,
|
||
recovery="retry_same_request")
|