Compressor/backend/indicator_features/one_page.py

#!/usr/bin/env python3
"""一页纸候选：每个事件只留最强的 2 条（机组/测点/角度段/指标/前后值/一句话）。"""
from __future__ import annotations

import argparse
import csv
import sys
from collections import defaultdict
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 EVENTS, collect_after, collect_before, cycle_count, load  # noqa: E402
from run_segments import unit_segments  # noqa: E402

METRICS = {"极差": "range", "粗糙度": "roughness", "水平": "mean",
           "波动次数": "zero_cross", "极值数": "n_extrema"}
# 已知坏道/传感器通道（观察记录里确认过）
BAD = {("7", "四", "盖侧"), ("7", "六", "盖侧"), ("8", "六", "盖侧")}
UP = {"极差": "波动变大", "粗糙度": "变毛糙", "水平": "抬高",
      "波动次数": "起伏次数变多", "极值数": "转折变多"}
DOWN = {"极差": "波动变小", "粗糙度": "变平滑", "水平": "降低",
        "波动次数": "起伏次数变少", "极值数": "转折变少"}


def robust_z(pre, post):
    if len(pre) < 3 or len(post) < 3:
        return None
    med = float(np.median(pre))
    mad = float(np.median(np.abs(np.asarray(pre) - med))) * 1.4826
    if mad <= 1e-9:
        mad = float(np.std(pre))
    if mad <= 1e-9:
        return None
    return (float(np.median(post)) - 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=2, help="每个事件保留几条")
    parser.add_argument("--z", type=float, default=4.0, help="强度阈值")
    parser.add_argument("--out", required=True)
    args = parser.parse_args()

    picked = []
    for unit, path in zip(args.unit, args.csv):
        rows = load(path)
        segments = unit_segments(unit, 2)
        for eu, date, kind in EVENTS:
            if eu != unit:
                continue
            t0 = datetime.fromisoformat(date)
            cands = []
            for cyl, side in sorted({(r["cylinder"], r["side"]) for r in rows}):
                if (unit, cyl, side) in BAD:
                    continue
                point = [r for r in rows if r["cylinder"] == cyl and r["side"] == side]
                pre, _ = collect_before(point, t0, segments, 6)
                post, _ = collect_after(point, t0, segments, 30, 45)
                if cycle_count(pre) < 3 or cycle_count(post) < 3:
                    continue
                span = np.median([r["p_span"] for r in pre if r["p_span"]]) if pre else 0
                if span < 0.3:  # 坏道/卡死
                    continue
                for b in range(36):
                    pr = [r for r in pre if r["bin"] == b]
                    po = [r for r in post if r["bin"] == b]
                    for cn, en in METRICS.items():
                        a = [r[en] for r in pr if r[en] is not None]
                        c = [r[en] for r in po if r[en] is not None]
                        z = robust_z(a, c)
                        if z is None or abs(z) < args.z:
                            continue
                        am, cm = float(np.median(a)), float(np.median(c))
                        cands.append({
                            "机组": unit, "事件": date, "故障类型": kind,
                            "测点": f"{cyl}缸{side}", "角度段": f"{b*10}-{b*10+10}°",
                            "指标": cn, "停机前": round(am, 3), "开机后": round(cm, 3),
                            "强度": round(abs(z), 1),
                            "结论": (UP if z > 0 else DOWN)[cn],
                        })
            cands.sort(key=lambda r: -r["强度"])
            seen = set()
            for row in cands:
                key = (row["测点"], row["角度段"])
                if key in seen:
                    continue
                seen.add(key)
                picked.append(row)
                if len(seen) >= args.top:
                    break

    out = Path(args.out)
    out.parent.mkdir(parents=True, exist_ok=True)
    fields = ["机组", "事件", "故障类型", "测点", "角度段", "指标", "停机前", "开机后", "强度", "结论"]
    with out.open("w", encoding="utf-8-sig", newline="") as fh:
        w = csv.DictWriter(fh, fieldnames=fields)
        w.writeheader()
        w.writerows(picked)
    print(f"一页纸候选 -> {out}（{len(picked)} 行）")
    print(f"{'机组':<4}{'事件':<12}{'测点':<9}{'角度段':<10}{'指标':<6}{'前':>8}{'后':>8}{'强度':>6}  结论")
    for r in picked:
        print(f"{r['机组']+'号':<4}{r['事件']:<12}{r['测点']:<9}{r['角度段']:<10}{r['指标']:<6}"
              f"{r['停机前']:>8}{r['开机后']:>8}{r['强度']:>6}  {r['结论']}")


if __name__ == "__main__":
    main()