Compressor/backend/indicator_features/make_report.py

#!/usr/bin/env python3
"""把英文中间结果整理成中文交付物（CSV + 汇总 md）。"""
from __future__ import annotations

import argparse
import csv
from collections import defaultdict
from datetime import datetime
from pathlib import Path

import cn_names as CN
from bins_common import BIN_METRICS, CYCLE_METRICS

BIN_DESC = {
    "均值": "箱内平均压力（水平）", "极差": "箱内 max-min（幅度）", "标准差": "箱内波动",
    "中位绝对偏差": "稳健波动", "斜率": "箱内压力对角度的斜率",
    "粗糙度": "去线性趋势后的残差 RMS（平滑/毛糙）", "10分位": "箱内 10% 分位",
    "50分位": "箱内中位数", "90分位": "箱内 90% 分位", "四分位距": "90 分位-10 分位",
    "偏度": "分布不对称度", "峰度": "尖峰程度（4 阶矩）",
    "平均|导数|": "相邻点压力差的平均绝对值（局部抖动）", "最大|导数|": "最大单步压力跳变",
    "导数RMS": "一阶导数均方根（抖动强度）", "二阶导RMS": "二阶导均方根（拐角尖锐）",
    "导数过零数": "一阶导数变号次数（振荡次数）", "极值数": "箱内局部极值个数",
    "二阶导最大": "最大二阶导（最尖的拐角）", "总变差": "Σ|相邻差|（弧长/毛糙）",
    "过均次数": "穿过箱均值的次数", "移动度": "Hjorth 移动度（频率感）",
    "复杂度": "Hjorth 复杂度（形状复杂度）",
}
CYCLE_DESC = {
    "吸气压力": "Otsu 低压平台", "排气压力": "Otsu 高压平台", "压比": "排气/吸气",
    "吸气占比": "低压样本占比", "排气占比": "高压样本占比",
    "最小压力角": "压力最小处的角度", "最大压力角": "压力最大处的角度",
    "上升穿越角": "压力上穿中间阈值角", "下降穿越角": "压力下穿中间阈值角",
    "最大上升斜率": "整周期最大正斜率", "最大下降斜率": "整周期最大负斜率",
    "最大上升角": "最大正斜率所在角", "最大下降角": "最大负斜率所在角",
}


def translate_csv(src: Path, dst: Path, metric_map: dict, family_map: dict | None = None) -> list[dict]:
    if not src.is_file():
        return []
    raw_headers = next(csv.reader(open(src, encoding="utf-8-sig")), [])
    rows = list(csv.DictReader(open(src, encoding="utf-8-sig")))
    headers = [CN.col(name) for name in raw_headers]
    out = []
    for row in rows:
        new = {}
        for key, value in row.items():
            if key == "metric":
                value = metric_map.get(value, value)
            elif key == "family" and family_map:
                value = family_map.get(value, value)
            elif key == "metrics":
                value = CN.translate_metrics_field(value)
            new[CN.col(key)] = value
        out.append(new)
    dst.parent.mkdir(parents=True, exist_ok=True)
    with dst.open("w", encoding="utf-8-sig", newline="") as handle:
        writer = csv.DictWriter(handle, fieldnames=headers)
        writer.writeheader()
        writer.writerows(out)
    return out


def num(value):
    try:
        return float(value)
    except (TypeError, ValueError):
        return None


def write_summary_md(summary_rows: list[dict], bins_dir: Path, out: Path) -> None:
    lines = []
    lines.append("# 功图分角度特征 汇总报告（中文）\n")
    lines.append(f"> 生成时间：{datetime.now():%Y-%m-%d %H:%M}。数据：7/8/9 号机，"
                 "每个周期按键相 0-360° 切 10° 箱，逐箱算 23 个形状/波动/分布指标，"
                 "再加 13 个整周期阀时序指标；对比在**同一测点**上做停机前 vs 开机后。\n")

    lines.append("## 一、怎么读结果\n")
    lines.append("- **特征候选汇总.csv**：每个候选 = 一个事件 × 一个测点 × 一个角度箱，把该箱里同时变化的指标合并在一行。")
    lines.append("  - `总分` = 该组所有指标 |MAD倍数| 之和；`指标数` = 变化 ≥3MAD 的指标个数。")
    lines.append("  - `相隔天数` = 后段起点 − 前段终点；`前跨度天数` = 前段首末之差。**这两个越小越可信**（避免把跨年/跨季工况当成故障）。")
    lines.append("- **事件明细_X号.csv**：每个角度箱/整周期/指标一行的前后值，供细查。")
    lines.append("- **图片**：`plots/` 下三联图——左=平均压力随角度、中=波动随角度、右=关注段随时间（红=停机前，蓝=开机后，虚线=事件日）。\n")

    lines.append("## 二、特征字典\n")
    lines.append("| 中文名 | 英文键 | 含义 |")
    lines.append("|---|---|---|")
    for name, en in [(CN.BIN_CN[m], m) for m in BIN_METRICS]:
        lines.append(f"| {name} | {en} | {BIN_DESC.get(name, '')} |")
    for name, en in [(CN.CYCLE_CN[m], m) for m in CYCLE_METRICS]:
        lines.append(f"| {name} | {en} | {CYCLE_DESC.get(name, '')} |")
    lines.append("")

    # 按事件分组
    groups = defaultdict(list)
    for row in summary_rows:
        groups[(row["机组"], row["事件日期"], row["故障类型"])].append(row)

    lines.append("## 三、各事件候选\n")
    for key in sorted(groups, key=lambda k: (k[0], k[1])):
        unit, date, kind = key
        items = sorted(groups[key], key=lambda r: -num(r["总分"]) if num(r["总分"]) else 0)
        lines.append(f"### {unit}号机 {date} {kind}\n")
        top = items[0]
        lines.append(f"- 前段 {top['前段止']} ← 后段 {top['后段起']}（相隔 {top['相隔天数']} 天，前跨度 {top['前跨度天数']} 天）")
        lines.append(f"- 候选组 {len(items)} 个，Top5：\n")
        lines.append("| 测点 | 角度箱 | 范围 | 指标数 | 总分 | 指标明细 |")
        lines.append("|---|---|---|---|---|---|")
        for r in items[:5]:
            bs = r["角箱起点"]
            if bs in ("", None):
                binst = "整周期"
            elif float(bs) < 0:
                binst = "全箱(整体)"
            else:
                binst = f"{float(bs):g}-{float(bs) + 10:g}°"
            lines.append(f"| {r['测点']} | {binst} | {r['范围']} | {r['指标数']} | {r['总分']} | {r['指标明细']} |")
        if any(str(r.get("范围", "")).startswith("整体") for r in items):
            lines.append("")
            lines.append("> 注：该事件存在**整体水平变化**（全角度一起动，可能负荷/工况），已单列。")
        lines.append("")

    clean = [r for r in summary_rows
             if num(r.get("前跨度天数")) is not None and num(r["前跨度天数"]) <= 20
             and num(r.get("相隔天数")) is not None and num(r["相隔天数"]) <= 40]
    clean.sort(key=lambda r: -num(r["总分"]))
    lines.append("## 四、干净短名单（前跨度≤20天 且 相隔≤40天）\n")
    lines.append("| # | 机组 | 事件 | 测点 | 角度箱 | 范围 | 指标数 | 总分 | 指标明细 |")
    lines.append("|---|---|---|---|---|---|---|---|---|")
    for i, r in enumerate(clean[:15], 1):
        _bs = r["角箱起点"]
        if _bs in ("", None):
            binst = "整周期"
        elif float(_bs) < 0:
            binst = "全箱(整体)"
        else:
            binst = f"{float(_bs):g}-{float(_bs) + 10:g}°"
        lines.append(f"| {i} | {r['机组']}号 | {r['事件日期']} {r['故障类型']} | {r['测点']} | {binst} | {r['范围']} | "
                     f"{r['指标数']} | {r['总分']} | {r['指标明细']} |")
    lines.append("")

    lines.append("## 五、图片索引\n")
    plots = sorted((bins_dir / "plots").glob("*.png"))
    for p in plots:
        lines.append(f"- `plots/{p.name}`")
    lines.append("")

    out.write_text("\n".join(lines), encoding="utf-8")


def main() -> None:
    parser = argparse.ArgumentParser(description="生成中文交付物")
    parser.add_argument("--dir", default=str(Path(__file__).resolve().parents[1] / "cache" / "indicator" / "bins"))
    parser.add_argument("--outdir", default="")
    args = parser.parse_args()
    bins_dir = Path(args.dir)
    outdir = Path(args.outdir) if args.outdir else bins_dir / "results_cn"
    outdir.mkdir(parents=True, exist_ok=True)

    all_metrics = {**CN.BIN_CN, **CN.CYCLE_CN}
    # 汇总
    summary_rows = translate_csv(bins_dir / "summary_final.csv", outdir / "特征候选汇总.csv",
                                 all_metrics, CN.FAMILY_CN)
    # 事件明细
    for unit in ("7", "8", "9"):
        src = bins_dir / f"event_final_{unit}.csv"
        if src.is_file():
            translate_csv(src, outdir / f"事件明细_{unit}号.csv", all_metrics, CN.FAMILY_CN)
        ssrc = bins_dir / f"scan_final_{unit}.csv"
        if ssrc.is_file():
            translate_csv(ssrc, outdir / f"全时间轴扫描_{unit}号.csv", all_metrics, CN.FAMILY_CN)

    write_summary_md(summary_rows, bins_dir, outdir / "特征汇总.md")
    print("中文交付 ->", outdir)
    for p in sorted(outdir.iterdir()):
        print("  ", p.name)


if __name__ == "__main__":
    main()