Compressor/backend/indicator_features/event_trend_summary.py

#!/usr/bin/env python3
"""Six-window event trend summary for sampled cycle profiles and pressure levels."""
from __future__ import annotations

import argparse
import csv
import sys
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

BAD = {("7", "四", "盖侧"), ("7", "六", "盖侧"), ("8", "六", "盖侧")}


def load(path):
    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"])
        for key in ("p_suction", "p_discharge", "p_span"):
            row[key] = float(row[key]) if row.get(key) else np.nan
        for key in row:
            if key.startswith("prof_"):
                row[key] = float(row[key]) if row[key] else np.nan
    return rows


def med(rows, key):
    values = [r[key] for r in rows if np.isfinite(r[key])]
    return float(np.median(values)) if values else np.nan


def main():
    parser = argparse.ArgumentParser(description="事件前后六段趋势")
    parser.add_argument("--csv", nargs="+", required=True)
    parser.add_argument("--unit", nargs="+", required=True)
    parser.add_argument("--out", required=True)
    parser.add_argument("--event", nargs=3, action="append", metavar=("UNIT", "DATE", "KIND"),
                        help="只分析指定事件；可重复传入")
    args = parser.parse_args()
    data = {u: load(p) for u, p in zip(args.unit, args.csv)}
    rows_out = []
    events = [tuple(item) for item in args.event] if args.event else EVENTS
    for unit, date_text, kind in events:
        source = [r for r in data.get(unit, []) if (unit, r["cylinder"], r["side"]) not in BAD and r["p_span"] >= 0.3]
        t0 = datetime.fromisoformat(date_text)
        profile_names = sorted(k for k in source[0] if k.startswith("prof_")) if source else []
        for cylinder, side in sorted({(r["cylinder"], r["side"]) for r in source}):
            point = [r for r in source if r["cylinder"] == cylinder and r["side"] == side]
            base = None
            for index, (lo, hi) in enumerate([(-30, -20), (-20, -10), (-10, 0), (0, 10), (10, 20), (20, 30)]):
                a, b = t0 + timedelta(days=lo), t0 + timedelta(days=hi)
                selected = [r for r in point if a <= r["time"] < b]
                profile = np.asarray([[r[k] for k in profile_names] for r in selected], dtype=float)
                profile = np.median(profile, axis=0) if len(profile) and np.all(np.isfinite(profile), axis=1).any() else None
                if profile is not None and base is None:
                    base = profile
                distance = float(np.linalg.norm(profile - base)) if profile is not None and base is not None else np.nan
                rows_out.append({
                    "机组": unit, "事件": date_text, "故障类型": kind,
                    "测点": f"{cylinder}缸{side}", "时间段": f"{lo}~{hi}天",
                    "周期数": len(selected),
                    "吸气压力": med(selected, "p_suction"),
                    "排气压力": med(selected, "p_discharge"),
                    "压力幅度": med(selected, "p_span"),
                    "相对首段形状距离": "" if not np.isfinite(distance) else round(distance, 5),
                })
    out = Path(args.out)
    out.parent.mkdir(parents=True, exist_ok=True)
    fields = list(rows_out[0])
    with out.open("w", encoding="utf-8-sig", newline="") as handle:
        writer = csv.DictWriter(handle, fieldnames=fields)
        writer.writeheader()
        writer.writerows(rows_out)
    print(f"事件趋势 -> {out}（{len(rows_out)} 行）")


if __name__ == "__main__":
    main()