Compressor/backend/indicator_features/make_table.py

#!/usr/bin/env python3
"""生成规范的多值总表与事件前后对照表（中文表头）。

- 特征总表.csv：机组,缸,腔,角度箱,时间,压力幅度,极差,粗糙度,水平,波动次数,极值数
  （每个 测点×角度箱×日 取当日中位数）
- 事件前后对照表.csv：机组,事件,故障类型,缸,腔,角度箱,指标,停机前,开机后,变化
  （5 个指标各一行，方便按"波动次数"筛角度）

只用现成的 `angle_bins.py` 输出，不重算数据库。
"""
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"),
]
META = ["机组", "缸", "腔", "角度箱", "时间", "压力幅度"]


def write_total(unit: str, rows: list[dict], out: Path) -> None:
    buckets = defaultdict(lambda: defaultdict(list))
    for r in rows:
        key = (r["cylinder"], r["side"], r["bin_start"], r["time"].date())
        buckets[key]["压力幅度"].append(r["p_span"])
        for cn, en in METRICS:
            if r[en] is not None:
                buckets[key][cn].append(r[en])
    with out.open("w", encoding="utf-8-sig", newline="") as fh:
        w = csv.writer(fh)
        w.writerow(META + [cn for cn, _ in METRICS])
        for (cyl, side, binstart, day) in sorted(buckets, key=lambda k: (k[0], k[1], k[2], k[3])):
            b = buckets[(cyl, side, binstart, day)]
            med = lambda k: (f"{np.median(b[k]):.4g}" if b.get(k) else "")
            w.writerow([unit, cyl, side, f"{binstart:g}", day.isoformat(), med("压力幅度")]
                       + [med(cn) for cn, _ in METRICS])


def write_event_table(unit: str, rows: list[dict], out: Path) -> None:
    segments = unit_segments(unit, 2)
    out_rows = []
    for eu, date, kind in EVENTS:
        if eu != unit:
            continue
        t0 = datetime.fromisoformat(date)
        for cyl, side in sorted({(r["cylinder"], r["side"]) for r in rows}):
            point_rows = [r for r in rows if r["cylinder"] == cyl and r["side"] == side]
            pre, _ = collect_before(point_rows, t0, segments, 6)
            post, _ = collect_after(point_rows, t0, segments, 30, 45)
            if cycle_count(pre) < 3 or cycle_count(post) < 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]
                if len(pr) < 3 or len(po) < 3:
                    continue
                for cn, en in METRICS:
                    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]
                    if len(a) < 3 or len(c) < 3:
                        continue
                    am, cm = float(np.median(a)), float(np.median(c))
                    out_rows.append({
                        "机组": unit, "事件": date, "故障类型": kind, "缸": cyl, "腔": side,
                        "角度箱": f"{b * 10}-{b * 10 + 10}", "指标": cn,
                        "停机前": f"{am:.4g}", "开机后": f"{cm:.4g}", "变化": f"{cm - am:+.4g}",
                    })
    with out.open("w", encoding="utf-8-sig", newline="") as fh:
        w = csv.DictWriter(fh, fieldnames=["机组", "事件", "故障类型", "缸", "腔", "角度箱", "指标",
                                           "停机前", "开机后", "变化"])
        w.writeheader()
        w.writerows(out_rows)


def main() -> None:
    parser = argparse.ArgumentParser(description="生成规范多值总表")
    parser.add_argument("--csv", nargs="+", required=True, help="一个或多个 angle_bins CSV")
    parser.add_argument("--unit", nargs="+", required=True, help="与 --csv 对应的机组号")
    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):
        print(f"读 {path} ...", flush=True)
        rows = load(path)
        write_total(unit, rows, outdir / f"特征总表_{unit}号.csv")
        write_event_table(unit, rows, outdir / f"事件前后对照_{unit}号.csv")
        print(f"  {unit}号：特征总表 + 事件前后对照 已生成")


if __name__ == "__main__":
    main()