Compressor/backend/indicator_features/segment_compare.py

#!/usr/bin/env python3
"""按运行段逐段对比：每一段开机 vs 上一段，列出最明显的变化（一页纸/段）。"""
from __future__ import annotations

import argparse
import csv
import sys
from datetime import datetime
from pathlib import Path

import numpy as np

BACKEND = Path(__file__).resolve().parents[1]
sys.path.insert(0, str(BACKEND))

from event_bins import load  # noqa: E402
from run_segments import unit_segments  # noqa: E402

METRICS = {"极差": "range", "粗糙度": "roughness", "水平": "mean",
           "波动次数": "zero_cross", "极值数": "n_extrema"}
UP = {"极差": "波动变大", "粗糙度": "变毛糙", "水平": "抬高",
      "波动次数": "起伏次数变多", "极值数": "转折变多"}
DOWN = {"极差": "波动变小", "粗糙度": "变平滑", "水平": "降低",
        "波动次数": "起伏次数变少", "极值数": "转折变少"}
BAD = {("7", "四", "盖侧"), ("7", "六", "盖侧"), ("8", "六", "盖侧")}


def med_mad(values):
    if not values:
        return None, None
    arr = np.asarray(values, dtype=float)
    med = float(np.median(arr))
    mad = float(np.median(np.abs(arr - med))) * 1.4826
    if mad <= 1e-9:
        mad = float(np.std(arr))
    return med, mad


def main() -> None:
    parser = argparse.ArgumentParser(description="运行段逐段对比")
    parser.add_argument("--csv", nargs="+", required=True)
    parser.add_argument("--unit", nargs="+", required=True)
    parser.add_argument("--top", type=int, default=5, help="每段保留几条")
    parser.add_argument("--z", type=float, default=6.0)
    parser.add_argument("--outdir", required=True)
    args = parser.parse_args()

    outdir = Path(args.outdir)
    outdir.mkdir(parents=True, exist_ok=True)
    for unit, path in zip(args.unit, args.csv):
        rows = load(path)
        segments = unit_segments(unit, 2)
        print(f"\n===== {unit}号机 运行段（{len(segments)} 段）=====")
        records = []
        for i in range(1, len(segments)):
            prev, cur = segments[i - 1], segments[i]
            pdate, cdate = f"{prev[0]}~{prev[1]}", f"{cur[0]}~{cur[1]}"
            cands = []
            for cyl, side in sorted({(r["cylinder"], r["side"]) for r in rows}):
                if (unit, cyl, side) in BAD:
                    continue
                p_rows = [r for r in rows if r["cylinder"] == cyl and r["side"] == side
                          and prev[0] <= r["time"].date() <= prev[1]]
                c_rows = [r for r in rows if r["cylinder"] == cyl and r["side"] == side
                          and cur[0] <= r["time"].date() <= cur[1]]
                if len({r["wave_file_id"] for r in p_rows}) < 3 or len({r["wave_file_id"] for r in c_rows}) < 3:
                    continue
                if np.median([r["p_span"] for r in p_rows if r["p_span"]] or [0]) < 0.3:
                    continue
                for b in range(36):
                    pb = [r for r in p_rows if r["bin"] == b]
                    cb = [r for r in c_rows if r["bin"] == b]
                    for cn, en in METRICS.items():
                        pm, mad = med_mad([r[en] for r in pb if r[en] is not None])
                        cm, _ = med_mad([r[en] for r in cb if r[en] is not None])
                        if pm is None or cm is None or mad is None or mad <= 1e-9:
                            continue
                        z = (cm - pm) / mad
                        if abs(z) < args.z:
                            continue
                        # 还要有"实际幅度"：幅度类≥3%，次数类≥2 次
                        if cn in ("波动次数", "极值数"):
                            if abs(cm - pm) < 2:
                                continue
                        elif abs(pm) > 1e-9 and abs(cm - pm) < 0.03 * abs(pm):
                            continue
                        z = max(-999.0, min(z, 999.0))
                        cands.append({
                            "机组": unit, "本段": cdate, "上段": pdate,
                            "测点": f"{cyl}缸{side}", "角度段": f"{b*10}-{b*10+10}°",
                            "指标": cn, "上段值": round(pm, 3), "本段值": round(cm, 3),
                            "强度": round(abs(z), 1),
                            "结论": (UP if z > 0 else DOWN)[cn],
                        })
            cands.sort(key=lambda r: -r["强度"])
            seen = set()
            picked = []
            for row in cands:
                key = (row["测点"], row["角度段"])
                if key in seen:
                    continue
                seen.add(key)
                picked.append(row)
                if len(picked) >= args.top:
                    break
            records.extend(picked)
            print(f"  {cdate} 相对于 上段 {pdate}：")
            if not picked:
                print("     无明显变化")
            for r in picked:
                print(f"     {r['测点']:<9}{r['角度段']:<10}{r['指标']:<6}"
                      f"{r['上段值']:>8}->{r['本段值']:<8} 强度{r['强度']:<5}{r['结论']}")
        if records:
            out = outdir / f"分段对比_{unit}号.csv"
            with out.open("w", encoding="utf-8-sig", newline="") as fh:
                w = csv.DictWriter(fh, fieldnames=list(records[0].keys()))
                w.writeheader()
                w.writerows(records)
            print(f"  -> {out}")


if __name__ == "__main__":
    main()