"""Deterministic monitoring summaries built only from frozen replay outputs.""" from __future__ import annotations from datetime import datetime, timedelta from statistics import median from typing import Any from uuid import NAMESPACE_URL, uuid5 import pandas as pd from app.domain.schemas.scada_replay import ScadaReplayCreate from app.domain.schemas.project_settings import ScadaMonitoringSettings from app.services.scada_normal_profile import current_calibration, normal_profile_bounds from app.services.burst_detection import ( TARGET_MU, TARGET_N_ESTIMATORS, TARGET_RANDOM_STATE, TARGET_SCORE_THRESHOLD, run_burst_detection, ) def _time(value: str | datetime) -> datetime: return value if isinstance(value, datetime) else datetime.fromisoformat(value) def _valid_rows(rows: list[dict], device_id: str) -> list[dict]: return sorted( ( row for row in rows if row.get("device_id") == device_id and row.get("quality") in {"valid", "corrected"} and row.get("cleaned_value") is not None ), key=lambda row: _time(row["time"]), ) def _relative_indicator( rows: list[dict], config: ScadaReplayCreate, device_ids: list[str], period_start: datetime, period_end: datetime, *, direction: str, calibration: dict | None = None, ) -> dict[str, Any]: details: list[dict[str, Any]] = [] gaps: list[dict[str, str]] = [] for device_id in device_ids: rule = next(( item for item in config.rules if item.kind == "relative_change" and device_id in item.device_ids and item.threshold is not None ), None) if rule is None: gaps.append({"device_id": device_id, "reason": "relative_change_rule_missing"}) continue profile = calibration.get(device_id) if calibration is not None else None if calibration is not None and profile is None: gaps.append({"device_id": device_id, "reason": "adaptive_history_not_ready"}) continue threshold = float(profile["threshold_percent"]) if profile else rule.threshold valid = _valid_rows(rows, device_id) current = [row for row in valid if period_start < _time(row["time"]) <= period_end] latest = current[-1] if current else None if not current: gaps.append({"device_id": device_id, "reason": "period_reading_missing"}) continue baseline = ([row for row in valid if _time(row["time"]) < _time(latest["time"])] [-rule.window_samples:]) if latest else [] if len(baseline) < rule.minimum_samples: gaps.append({"device_id": device_id, "reason": "baseline_not_ready"}) continue reference = median(float(row["cleaned_value"]) for row in baseline) if reference == 0: gaps.append({"device_id": device_id, "reason": "zero_baseline"}) continue value = float(latest["cleaned_value"]) change = (value - reference) / abs(reference) * 100 hit = change <= -threshold if direction == "decrease" else change >= threshold bounds = normal_profile_bounds(profile.get("normal_profile"), _time(latest["time"])) if profile else None if profile and bounds is None: gaps.append({"device_id": device_id, "reason": "normal_phase_history_not_ready"}) continue if bounds is not None: hit = hit and not bounds[0] <= value <= bounds[1] details.append({ "device_id": device_id, "observed_at": _time(latest["time"]).isoformat(), "baseline": reference, "baseline_method": "recent_samples_median", "baseline_start": _time(baseline[0]["time"]).isoformat(), "baseline_end": _time(baseline[-1]["time"]).isoformat(), "baseline_sample_count": len(baseline), "value": value, "value_change": value - reference, "unit": rule.unit, "change_percent": change, "threshold_percent": threshold, **({"normal_phase_lower": bounds[0], "normal_phase_upper": bounds[1]} if bounds else {}), **({"threshold_method": profile["method"], "configured_threshold_percent": rule.threshold} if profile else {}), "hit": hit, }) return { "status": "ready" if not gaps else "insufficient_data", "hit": any(item["hit"] for item in details), "hit_device_ids": [item["device_id"] for item in details if item["hit"]], "devices": details, "gaps": gaps, } def _burst_detection( rows: list[dict], pressure_device_ids: list[str], interval_seconds: int, period_end: datetime, *, score_threshold: float = TARGET_SCORE_THRESHOLD, ) -> dict[str, Any]: if len(pressure_device_ids) < 5: return {"status": "sensor_coverage_insufficient", "hit": False, "required_sensors": 5, "available_sensors": len(pressure_device_ids)} if interval_seconds % 60 or 86400 % interval_seconds: return {"status": "unsupported_sampling_interval", "hit": False, "interval_seconds": interval_seconds} points_per_day = 86400 // interval_seconds count = 15 * points_per_day series_by_device: dict[str, dict[datetime, float]] = {} common_times: set[datetime] | None = None for device_id in pressure_device_ids: values = {_time(row["time"]): float(row["cleaned_value"]) for row in _valid_rows(rows, device_id) if _time(row["time"]) <= period_end} series_by_device[device_id] = values common_times = set(values) if common_times is None else common_times & set(values) if not common_times: return {"status": "baseline_not_ready", "hit": False, "required_samples": count} target = period_end if target not in common_times: return {"status": "current_reading_missing", "hit": False, "target_time": target.isoformat(), "latest_common_time": max(common_times).isoformat()} expected = [target - timedelta(seconds=interval_seconds * offset) for offset in range(count - 1, -1, -1)] incomplete = [device_id for device_id, values in series_by_device.items() if any(at not in values for at in expected)] if incomplete: return {"status": "baseline_not_ready", "hit": False, "required_samples": count, "incomplete_device_ids": incomplete} frame = pd.DataFrame({device_id: [series_by_device[device_id][at] for at in expected] for device_id in pressure_device_ids}, index=expected) try: result = run_burst_detection( network="scada-replay", username="scada-replay", observed_pressure_data=frame, points_per_day=points_per_day, mu=TARGET_MU, iforest_params={ "n_estimators": TARGET_N_ESTIMATORS, "random_state": TARGET_RANDOM_STATE, "contamination": "auto", }, sensor_nodes=pressure_device_ids, ) except (TypeError, ValueError) as exc: return {"status": "analysis_failed", "hit": False, "reason": str(exc)[:300]} latest = result["rows"][-1] score = float(latest["Score"]) return { "status": "ready", "hit": score <= score_threshold, "target_time": target.isoformat(), "target_score": score, "score_interpretation": "lower_is_more_anomalous_not_probability", "score_threshold": score_threshold, "sensor_count": len(pressure_device_ids), "reference_day_count": 14, "latest_sensor_rankings": result["summary"].get("latest_sensor_rankings", []), } def evaluate_monitoring_period( document: dict, rows: list[dict], *, period_start: datetime, period_end: datetime, batch_sequence: int, ) -> dict[str, Any]: config = ScadaReplayCreate.model_validate(document["config"]) monitoring = document.get("monitoring_config") rule_version = (monitoring or {}).get("rule_version", "unconfigured") monitoring_id = str(uuid5( NAMESPACE_URL, f"{document['replay_id']}:{period_end.isoformat()}:{rule_version}", )) base = { "monitoring_id": monitoring_id, "period_start": period_start.isoformat(), "period_end": period_end.isoformat(), "knowledge_cutoff": period_end.isoformat(), "batch_sequence": batch_sequence, "rule_version": rule_version, "task_id": None, } if not monitoring: return {**base, "status": "insufficient_data", "detail": "项目未配置 SCADA 监测设备角色", "pressure": {"status": "unconfigured", "hit": False}, "outlet_flow": {"status": "unconfigured", "hit": False}, "burst_detection": {"status": "not_run", "hit": False}} pressure_ids = monitoring.get("pressure_device_ids", []) flow_ids = monitoring.get("outlet_flow_device_ids", []) selected = set(config.device_ids) role_gaps = sorted((set(pressure_ids) | set(flow_ids)) - selected) weighted_policy = monitoring.get("decision_policy") == "weighted" if not weighted_policy and (not pressure_ids or not flow_ids or role_gaps): return {**base, "status": "insufficient_data", "detail": "监测角色为空或不属于本轮回放设备", "role_gaps": role_gaps, "pressure": {"status": "unconfigured", "hit": False}, "outlet_flow": {"status": "unconfigured", "hit": False}, "burst_detection": {"status": "not_run", "hit": False}} calibration = current_calibration(document, config.start_time) if weighted_policy else None if config.start_time > period_end: calibration = None base["role_gaps"] = role_gaps pressure = _relative_indicator( rows, config, pressure_ids, period_start, period_end, direction="decrease", calibration=(calibration or {}).get("pressure", {}) if weighted_policy else None, ) outlet_flow = _relative_indicator( rows, config, flow_ids, period_start, period_end, direction="increase", calibration=(calibration or {}).get("outlet_flow", {}) if weighted_policy else None, ) if flow_ids else {"status": "unconfigured", "hit": False, "hit_device_ids": [], "devices": [], "gaps": []} detector = {"status": "baseline_not_ready", "hit": False} if weighted_policy and ( not calibration or calibration.get("status") != "ready" ) else _burst_detection( rows, pressure_ids, int(document["monitoring_interval_seconds"]), period_end, score_threshold=calibration["burst_detection"]["threshold"] if calibration else TARGET_SCORE_THRESHOLD, ) if weighted_policy: settings = ScadaMonitoringSettings.model_validate(monitoring) weights = settings.weights.model_dump() indicators = {"burst_detection": detector, "pressure": pressure, "outlet_flow": outlet_flow} contributions = {key: weights[key] if indicator["status"] == "ready" and indicator["hit"] else 0.0 for key, indicator in indicators.items()} score = round(sum(contributions.values()), 10) complete = pressure["status"] == detector["status"] == "ready" diagnostic_missing = sorted({device for rule in config.rules if rule.kind == "relative_change" for device in rule.device_ids} - set((calibration or {}).get("diagnostic", {}))) if not complete: status, detail = "insufficient_data", ( "旧回放缺少新版历史诊断校准，请停止后新建演示" if calibration is None else "压力数据或爆管侦测基线不足，尚不能作出组合判定" ) elif score >= settings.alert_threshold: status, detail = "suspected_burst", "爆管相关指标加权达到疑似爆管门槛" elif any(indicator["hit"] for indicator in indicators.values()): status, detail = "attention", "部分指标异常，进入调查；综合结果尚未达到爆管嫌疑门槛" elif diagnostic_missing: status, detail = "insufficient_data", "部分测点缺少完整历史同期基线，不能判定全部指标正常" else: status, detail = "normal", "本周期爆管相关指标均未命中" detail += f"（综合 {score:.2f} / 门槛 {settings.alert_threshold:.2f}，流量权重 {weights['outlet_flow']:.0%}）" return {**base, "status": status, "detail": detail, "pressure": pressure, "outlet_flow": outlet_flow, "burst_detection": detector, "weighted": {"score": score, "threshold": settings.alert_threshold, "weights": weights, "contributions": contributions, "unavailable_indicators": [key for key, value in indicators.items() if value["status"] != "ready"], "calibration_cutoff": calibration["knowledge_cutoff"] if calibration else None, "diagnostic_unavailable_device_ids": diagnostic_missing, "missing_weight_policy": "zero_contribution_without_renormalization"}} complete = pressure["status"] == outlet_flow["status"] == detector["status"] == "ready" hits = (bool(detector["hit"]), bool(pressure["hit"]), bool(outlet_flow["hit"])) if not complete: status, detail = "insufficient_data", "监测数据、设备角色或爆管侦测基线不足" elif all(hits): status, detail = "suspected_burst", "爆管侦测、压力下降与出厂边界流量增加相互支持" elif any(hits): status, detail = "attention", "仅部分爆管相关指标命中，需要继续观察或人工核查" else: status, detail = "normal", "本周期爆管相关指标均未命中" return {**base, "status": status, "detail": detail, "pressure": pressure, "outlet_flow": outlet_flow, "burst_detection": detector}