vent_agent_2.0/extensions/vent_airflow/monitoring.py

"""安全监控风量数据源适配器（工具 2 的数据侧）。

本轮没有真实监控系统对接，因此做成**可插拔适配器**：默认实现读本地文件
（`<AIRFLOW_MONITORING_DIR>/<mine_id>.json`），接真实系统时只需替换
`get_source()` 返回的实现，工具与 Skill 都不用改。

无数据源时返回 `data_insufficient` 并转人工复核——绝不用测风报表顶替监控数据
（需求 §2.3 工具 2 要求 4）。
"""
from __future__ import annotations

import json
from bisect import bisect_left
from pathlib import Path
from typing import Any, Iterable, Mapping, Protocol

import config

from algorithms.airflow_review.contracts import parse_time

# 采样与异常判定阈值：恒值、突跳、量程，都可配而不写死在逻辑里
_STUCK_REPEATS = 6
_JUMP_PERCENT = 30.0
_JUMP_ABSOLUTE = 100.0
_RANGE_MIN = 0.0
_RANGE_MAX = 20000.0
_GAP_FACTOR = 3.0


class MonitoringSource(Protocol):
    """监控数据源协议：接真实系统时实现 query 即可。"""

    def query(
        self,
        *,
        mine_id: str,
        target_ids: Iterable[str],
        sensor_ids: Iterable[str],
        start_time: str,
        end_time: str,
        aggregation: str,
    ) -> dict[str, Any]: ...


class LocalFileMonitoringSource:
    """本地文件数据源：每个矿一个 JSON，字段见 data/monitoring/README.md。"""

    def __init__(self, directory: str | Path | None = None) -> None:
        self._directory = Path(directory) if directory else config.resolve_path(config.AIRFLOW_MONITORING_DIR)

    def path_for(self, mine_id: str) -> Path:
        return self._directory / f"{mine_id}.json"

    def query(
        self,
        *,
        mine_id: str,
        target_ids: Iterable[str] = (),
        sensor_ids: Iterable[str] = (),
        start_time: str = "",
        end_time: str = "",
        aggregation: str = "day",
    ) -> dict[str, Any]:
        path = self.path_for(mine_id)
        if not path.is_file():
            return {
                "status": "data_insufficient",
                "reason": f"未接入 {mine_id} 的安全监控数据源（缺少 {path.name}）",
                "series": [],
                "data_quality": {"missing_rate": None, "offline_periods": []},
                "manual_review_required": True,
                "note": "安全监控数据不可用；连续监测与交叉校核结论必须转人工复核，"
                        "不得用测风报表代替监控数据得出结论",
            }
        try:
            payload = json.loads(path.read_text(encoding="utf-8"))
        except json.JSONDecodeError as exc:
            return {"status": "error", "error": f"监控数据源不是合法 JSON：{exc.msg}", "series": []}

        wanted_targets = {str(item) for item in target_ids if item}
        wanted_sensors = {str(item) for item in sensor_ids if item}
        start, end = parse_time(start_time), parse_time(end_time)
        series: list[dict[str, Any]] = []
        missing_total = 0
        online_total = 0
        offline: list[dict[str, Any]] = []
        for raw in payload.get("series") or []:
            if wanted_targets and str(raw.get("target_id")) not in wanted_targets:
                continue
            if wanted_sensors and str(raw.get("sensor_id")) not in wanted_sensors:
                continue
            samples = [item for item in (raw.get("samples") or []) if _in_range(item, start, end)]
            stats, quality = _aggregate(samples, aggregation, float(raw.get("interval_minutes") or 1.0))
            missing_total += quality["missing_slots"]
            online_total += quality["expected_slots"]
            offline.extend(quality["offline_periods"])
            series.append({
                "sensor_id": raw.get("sensor_id"),
                "target_id": raw.get("target_id"),
                "target_type": raw.get("target_type", ""),
                "aggregation": aggregation,
                "statistics": stats,
                "sensor_status": quality["sensor_status"],
                "anomalies": quality["anomalies"],
                "evidence_ids": list(raw.get("evidence_ids") or []),
            })
        data_quality = {
            "series_count": len(series),
            "missing_rate": round(missing_total / online_total, 4) if online_total else None,
            "offline_periods": offline,
            "source_path": str(path),
        }
        status = "ok" if series else "data_insufficient"
        result: dict[str, Any] = {
            "status": status,
            "mine_id": mine_id,
            "window": {"start": start_time, "end": end_time, "aggregation": aggregation},
            "series": series,
            "data_quality": data_quality,
        }
        if status == "data_insufficient":
            result["reason"] = "查询窗口内没有匹配的监控序列"
            result["manual_review_required"] = True
        return result


def get_source() -> MonitoringSource:
    """返回当前配置的监控数据源；接真实系统时在这里换实现。"""
    return LocalFileMonitoringSource()


def _in_range(sample: Mapping[str, Any], start: Any, end: Any) -> bool:
    moment = parse_time(sample.get("at"))
    if moment is None:
        return False
    if start and moment < start:
        return False
    if end and moment > end:
        return False
    return True


def _aggregate(
    samples: list[Mapping[str, Any]],
    aggregation: str,
    interval_minutes: float,
) -> tuple[dict[str, Any], dict[str, Any]]:
    """按 point/hour/shift/day/month 聚合，并标记断线、恒值、突跳与量程异常。"""
    points: list[tuple[Any, float]] = []
    out_of_range: list[dict[str, Any]] = []
    for sample in samples:
        moment = parse_time(sample.get("at"))
        value = sample.get("airflow_m3_min")
        if moment is None or not isinstance(value, (int, float)):
            continue
        numeric = float(value)
        if numeric < _RANGE_MIN or numeric > _RANGE_MAX:
            out_of_range.append({"at": sample.get("at"), "value": numeric})
            continue
        points.append((moment, numeric))
    points.sort(key=lambda item: item[0])

    buckets = _bucketize(points, aggregation)
    statistics: dict[str, Any] = {"sample_count": len(points), "bucket_count": len(buckets)}
    # 统计量按**全体样本**给出（最小/最大是全局极值），分桶明细另列，避免把
    # "各桶最小值的平均"当成最小值——这种口径会让断线低谷被平均掉。
    if points:
        values = sorted(value for _, value in points)
        statistics.update({
            "mean_m3_min": round(sum(values) / len(values), 4),
            "min_m3_min": round(values[0], 4),
            "max_m3_min": round(values[-1], 4),
            "p50_m3_min": round(_percentile(values, 0.5), 4),
            "p95_m3_min": round(_percentile(values, 0.95), 4),
        })
    if buckets:
        statistics["buckets"] = buckets
        statistics["first_bucket"] = {key: buckets[0][key] for key in ("start", "end", "mean_m3_min")}
        statistics["last_bucket"] = {key: buckets[-1][key] for key in ("start", "end", "mean_m3_min")}

    if points:
        span_minutes = (points[-1][0] - points[0][0]).total_seconds() / 60.0
        expected_slots = int(span_minutes / interval_minutes) + 1 if interval_minutes > 0 else len(points)
    else:
        expected_slots = 0
    missing_slots = max(expected_slots - len(points), 0)
    statistics["valid_sample_rate"] = round(len(points) / expected_slots, 4) if expected_slots else 0.0

    anomalies: list[dict[str, Any]] = []
    offline = _offline_periods(points, interval_minutes)
    for period in offline:
        anomalies.append({"kind": "断线", **period})
    anomalies.extend(_stuck_periods(points))
    anomalies.extend(_jumps(points))
    for sample in out_of_range:
        anomalies.append({"kind": "量程异常", **sample})

    kinds = {anomaly["kind"] for anomaly in anomalies}
    if points and "断线" in kinds and len(offline) >= 2:
        sensor_status = "offline"
    elif kinds & {"恒值", "突跳", "量程异常"}:
        sensor_status = "abnormal"
    elif not points:
        sensor_status = "no_data"
    else:
        sensor_status = "normal"
    return statistics, {
        "expected_slots": expected_slots,
        "missing_slots": missing_slots,
        "offline_periods": offline,
        "anomalies": anomalies,
        "sensor_status": sensor_status,
    }


def _bucketize(points: list[tuple[Any, float]], aggregation: str) -> list[dict[str, Any]]:
    if not points:
        return []
    key_func = {
        "point": lambda moment: moment.isoformat(),
        "hour": lambda moment: moment.strftime("%Y-%m-%dT%H"),
        "shift": lambda moment: moment.strftime("%Y-%m-%d-") + _shift_name(moment.hour),
        "day": lambda moment: moment.strftime("%Y-%m-%d"),
        "month": lambda moment: moment.strftime("%Y-%m"),
    }.get(aggregation, lambda moment: moment.strftime("%Y-%m-%d"))
    buckets: dict[str, list[tuple[Any, float]]] = {}
    for moment, value in points:
        buckets.setdefault(key_func(moment), []).append((moment, value))
    result: list[dict[str, Any]] = []
    for key in sorted(buckets):
        group = sorted(buckets[key], key=lambda item: item[0])
        values = sorted(value for _, value in group)
        result.append({
            "key": key,
            "start": group[0][0].isoformat(),
            "end": group[-1][0].isoformat(),
            "mean_m3_min": round(sum(values) / len(values), 4),
            "min_m3_min": round(values[0], 4),
            "max_m3_min": round(values[-1], 4),
            "p50_m3_min": round(_percentile(values, 0.5), 4),
            "p95_m3_min": round(_percentile(values, 0.95), 4),
            "sample_count": len(values),
        })
    return result


def _shift_name(hour: int) -> str:
    if 0 <= hour < 8:
        return "夜班"
    if 8 <= hour < 16:
        return "早班"
    return "中班"


def _percentile(values: list[float], fraction: float) -> float:
    if not values:
        return 0.0
    index = bisect_left(values, values[min(int(len(values) * fraction), len(values) - 1)])
    return values[min(index, len(values) - 1)]


def _offline_periods(points: list[tuple[Any, float]], interval_minutes: float) -> list[dict[str, Any]]:
    limit = interval_minutes * _GAP_FACTOR
    periods: list[dict[str, Any]] = []
    for (previous, _), (current, _) in zip(points, points[1:]):
        gap = (current - previous).total_seconds() / 60.0
        if gap > limit:
            periods.append({"start": previous.isoformat(), "end": current.isoformat(),
                            "gap_minutes": round(gap, 2)})
    return periods


def _stuck_periods(points: list[tuple[Any, float]]) -> list[dict[str, Any]]:
    anomalies: list[dict[str, Any]] = []
    run_start = 0
    for index in range(1, len(points) + 1):
        same = index < len(points) and points[index][1] == points[run_start][1]
        if same:
            continue
        if index - run_start >= _STUCK_REPEATS:
            anomalies.append({"kind": "恒值", "start": points[run_start][0].isoformat(),
                              "end": points[index - 1][0].isoformat(),
                              "value": points[run_start][1], "repeat": index - run_start})
        run_start = index
    return anomalies


def _jumps(points: list[tuple[Any, float]]) -> list[dict[str, Any]]:
    anomalies: list[dict[str, Any]] = []
    for (previous_at, previous_value), (current_at, current_value) in zip(points, points[1:]):
        delta = abs(current_value - previous_value)
        baseline = max(abs(previous_value), 1.0)
        if delta > _JUMP_ABSOLUTE and delta / baseline * 100.0 > _JUMP_PERCENT:
            anomalies.append({"kind": "突跳", "at": current_at.isoformat(),
                              "from": previous_value, "to": current_value,
                              "change_percent": round(delta / baseline * 100.0, 2)})
    return anomalies