"""Causal batch quality checks and grouped anomaly episodes, no database IO.""" from copy import deepcopy from datetime import datetime from math import isfinite from statistics import median from uuid import NAMESPACE_URL, uuid5 import re from app.domain.schemas.scada_replay import ScadaReplayCreate from app.services.scada_normal_profile import ( DIAGNOSTIC_POLICY, current_calibration, normal_profile_bounds, weighted_diagnosis, ) def prepare_cleaning(rows: list[dict], config: ScadaReplayCreate) -> list[dict]: result = [] for row in rows: value = row["monitored_value"] quality, reason = "valid", "原始值通过数值与已配置量程校验" if value is None: quality, reason = "missing", "原始读数缺失" elif not isfinite(float(value)): quality, reason = "invalid", "原始读数不是有限数值" else: bounds = config.quality_ranges.get(row["device_id"]) if bounds and ((bounds.lower is not None and value < bounds.lower) or (bounds.upper is not None and value > bounds.upper)): quality, reason = "invalid", "超出本次明确配置的仪表量程" result.append({ "time": row["time"].isoformat() if isinstance(row["time"], datetime) else row["time"], "device_id": row["device_id"], "source_hash": row["source_hash"], "monitored_value": value if value is not None and isfinite(float(value)) else None, "cleaned_value": value if quality == "valid" else None, "quality": quality, "reason": reason, }) return result def detect_frame(document: dict, rows: list[dict], config: ScadaReplayCreate, at: datetime) -> tuple[dict, list[dict]]: """Consume a complete timestamp; never treat missing data as recovery.""" result = deepcopy(document) histories = result.setdefault("histories", {}) states = result.setdefault("rule_states", {}) latest = result.setdefault("latest", {}) for row in rows: latest[row["device_id"]] = row by_id = {row["device_id"]: row for row in rows} changes = [] devices = {device["device_id"]: device for device in result.get("devices", [])} weighted = weighted_diagnosis(result) calibration = current_calibration(result, config.start_time) if weighted else None for device_id in config.device_ids: row = by_id.get(device_id) invalid = row is not None and row["quality"] != "valid" previous = latest.get(device_id) frequency = str(devices.get(device_id, {}).get("transmission_frequency", "")) match = re.fullmatch(r"(\d+):(\d+):(\d+)", frequency) if match: seconds = int(match[1]) * 3600 + int(match[2]) * 60 + int(match[3]) else: minutes = re.fullmatch(r"(\d+)\s*(?:min|分钟)", frequency) seconds = int(minutes[1]) * 60 if minutes else 0 prior_time = datetime.fromisoformat(previous["time"]) if previous else config.start_time missing = not row and seconds > 0 and (at - prior_time).total_seconds() >= seconds * 2 key = "quality-" + device_id state = states.setdefault(key, {"event": None, "recoveries": 0}) if invalid or missing: state["recoveries"] = 0 if state["event"] is None: event = {"event_id": str(uuid5(NAMESPACE_URL, f"{result['replay_id']}:{key}:{at.isoformat()}")), "rule_id": key, "device_ids": [device_id], "occurred_at": at.isoformat(), "last_observed_at": at.isoformat(), "status": "active", "category": "data_quality", "detail": row["reason"] if invalid else "连续两个设备周期未读取到观测", "task_id": None} state["event"] = event changes.append({**event, "change": "raised"}) result["anomaly_count"] = result.get("anomaly_count", 0) + 1 elif row and row["quality"] == "valid" and state["event"]: state["recoveries"] += 1 if state["recoveries"] >= 2: changes.append({**state["event"], "status": "resolved", "resolved_at": at.isoformat(), "change": "resolved"}) state["event"] = None for rule in config.rules: state_key = f"{rule.rule_id}:{DIAGNOSTIC_POLICY}" if weighted and rule.kind == "relative_change" else rule.rule_id state = states.setdefault(state_key, {"hits": 0, "recoveries": 0, "event": None}) hits, usable, reasons = [], [], [] for device_id in rule.device_ids: row = by_id.get(device_id) if not row or row["quality"] != "valid": continue value = row["cleaned_value"] if rule.kind == "bounds": triggered = ((rule.lower is not None and value < rule.lower) or (rule.upper is not None and value > rule.upper)) reason = f"{device_id}={value} {rule.unit}，超出配置范围" elif weighted: # The percentage in the compatibility rule is not a diagnosis # threshold. Learn each device's normal same-phase range first. profile = (calibration or {}).get("diagnostic", {}).get(device_id) bounds = normal_profile_bounds(profile, at) if bounds is None: continue low, high = bounds triggered = (value < low or value > high if rule.direction == "both" else value > high if rule.direction == "increase" else value < low) reason = f"{device_id}={value} {rule.unit}，超出冻结正常同时段范围 [{low:.4f}, {high:.4f}] {rule.unit}" else: previous = histories.get(device_id, [])[-rule.window_samples:] if len(previous) < rule.minimum_samples: continue baseline = median(previous) if baseline == 0: continue change = (value - baseline) / abs(baseline) * 100 triggered = (abs(change) >= rule.threshold if rule.direction == "both" else change >= rule.threshold if rule.direction == "increase" else change <= -rule.threshold) reason = f"{device_id} 较历史中位值变化 {change:.1f}%，阈值 {rule.threshold}%" usable.append(device_id) if triggered: hits.append(device_id) reasons.append(reason) if len(hits) >= rule.minimum_devices: state["hits"] += 1 state["recoveries"] = 0 if state["event"] is None and state["hits"] >= rule.consecutive_samples: event_id = str(uuid5(NAMESPACE_URL, f"{result['replay_id']}:{state_key}:{at.isoformat()}")) event = {"event_id": event_id, "rule_id": rule.rule_id, "device_ids": hits, "occurred_at": at.isoformat(), "last_observed_at": at.isoformat(), "status": "active", "detail": "；".join(reasons)[:2000], "task_id": None} if weighted and rule.kind == "relative_change": event.update(diagnostic_policy=DIAGNOSTIC_POLICY, state_key=state_key) state["event"] = event changes.append({**event, "change": "raised"}) result["anomaly_count"] = result.get("anomaly_count", 0) + 1 elif state["event"]: state["event"]["last_observed_at"] = at.isoformat() elif len(usable) == len(rule.device_ids): state["hits"] = 0 if state["event"]: state["recoveries"] += 1 if state["recoveries"] >= rule.recovery_samples: event = {**state["event"], "status": "resolved", "resolved_at": at.isoformat(), "change": "resolved"} changes.append(event) state["event"] = None state["recoveries"] = 0 # Incomplete frames cannot count as a consecutive hit/recovery. elif not usable and any(device in by_id for device in rule.device_ids): state["hits"] = state["recoveries"] = 0 frozen = {device for rule in config.rules if states[ f"{rule.rule_id}:{DIAGNOSTIC_POLICY}" if weighted and rule.kind == "relative_change" else rule.rule_id ]["event"] for device in rule.device_ids} for row in rows: if row["quality"] == "valid" and row["device_id"] not in frozen: history = histories.setdefault(row["device_id"], []) history.append(row["cleaned_value"]) histories[row["device_id"]] = history[-96:] result["processed_count"] = result.get("processed_count", 0) + len(rows) return result, changes