Compressor/backend/indicator_features/event_shape_audit.py

#!/usr/bin/env python3
"""Audit event coverage and normalized whole-cycle shape changes from sampled features."""
from __future__ import annotations

import argparse
import csv
import sys
from collections import defaultdict
from datetime import datetime, timedelta
from pathlib import Path

import numpy as np

BACKEND = Path(__file__).resolve().parents[1]
sys.path.insert(0, str(BACKEND))
from event_bins import EVENTS  # noqa: E402


def load(path: str) -> list[dict]:
    with open(path, encoding="utf-8-sig", newline="") as handle:
        rows = list(csv.DictReader(handle))
    for row in rows:
        row["time"] = datetime.fromisoformat(row["sample_time"])
        row["span"] = float(row.get("p_span") or 0)
        for key in row:
            if key.startswith("prof_"):
                row[key] = float(row[key]) if row[key] else np.nan
    return rows


def median_profile(rows: list[dict], names: list[str]) -> np.ndarray | None:
    values = [[r[n] for n in names] for r in rows]
    if not values:
        return None
    result = np.nanmedian(np.asarray(values, dtype=float), axis=0)
    return result if np.all(np.isfinite(result)) else None


def main() -> None:
    parser = argparse.ArgumentParser(description="事件样本覆盖和整条功图形状审计")
    parser.add_argument("--csv", nargs="+", required=True)
    parser.add_argument("--unit", nargs="+", required=True)
    parser.add_argument("--pre-days", type=int, default=30)
    parser.add_argument("--post-days", type=int, default=30)
    parser.add_argument("--out", required=True)
    args = parser.parse_args()

    by_unit = {unit: load(path) for unit, path in zip(args.unit, args.csv)}
    output = []
    for unit, date_text, kind in EVENTS:
        rows = [r for r in by_unit.get(unit, []) if r["span"] >= 0.3]
        t0 = datetime.fromisoformat(date_text)
        pre_start = t0 - timedelta(days=args.pre_days)
        post_end = t0 + timedelta(days=args.post_days)
        pre = [r for r in rows if pre_start <= r["time"] < t0]
        post = [r for r in rows if t0 <= r["time"] <= post_end]
        points = sorted({(r["cylinder"], r["side"]) for r in rows})
        profile_names = sorted(k for k in rows[0] if k.startswith("prof_")) if rows else []
        for cylinder, side in points:
            a = [r for r in pre if r["cylinder"] == cylinder and r["side"] == side]
            b = [r for r in post if r["cylinder"] == cylinder and r["side"] == side]
            pa = median_profile(a, profile_names)
            pb = median_profile(b, profile_names)
            distance = float(np.linalg.norm(pa - pb)) if pa is not None and pb is not None else None
            output.append({
                "机组": unit,
                "事件": date_text,
                "故障类型": kind,
                "测点": f"{cylinder}缸{side}",
                "修前有效周期": len(a),
                "修前日期数": len({r["time"].date() for r in a}),
                "修后有效周期": len(b),
                "修后日期数": len({r["time"].date() for r in b}),
                "形状距离": "" if distance is None else round(distance, 4),
            })

    fields = list(output[0]) if output else []
    out = Path(args.out)
    out.parent.mkdir(parents=True, exist_ok=True)
    with out.open("w", encoding="utf-8-sig", newline="") as handle:
        writer = csv.DictWriter(handle, fieldnames=fields)
        writer.writeheader()
        writer.writerows(output)

    print(f"事件审计 -> {out}")
    for event in EVENTS:
        unit, date_text, kind = event
        selected = [r for r in output if r["机组"] == unit and r["事件"] == date_text]
        comparable = [r for r in selected if r["修前有效周期"] and r["修后有效周期"]]
        print(f"{unit}号 {date_text} {kind}: 可比较测点 {len(comparable)}/{len(selected)}")
        for row in sorted(comparable, key=lambda r: float(r["形状距离"] or 0), reverse=True)[:3]:
            print(f"  {row['测点']}: {row['修前有效周期']}->{row['修后有效周期']} 个周期，形状距离 {row['形状距离']}")


if __name__ == "__main__":
    main()