#!/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()