Compressor/_tmp_temperature/make_transition_labels.py

#!/usr/bin/env python3
"""Build ``温度过渡段.csv`` from the legacy temperature events.

The new run-state logic removes start/stop transition hours.  The events below
are the ones that disappeared for that reason; they are kept here as a separate
"transition" review set, not as confirmed faults.
"""
from __future__ import annotations

import csv
from pathlib import Path

ROOT = Path(__file__).resolve().parent
OUT = ROOT / "out"
LEGACY = OUT / "legacy_temperature_events.csv"

TRANSITIONS = [
    ("4号机", "2026-02-15 00:00:00", "持续偏高", "停机", "旧事件落在停机降温段"),
    ("4号机", "2026-03-24 10:00:00", "混合", "启动", "旧事件落在开机升温段"),
    ("4号机", "2026-04-04 17:00:00", "持续偏低", "启动", "旧事件落在开机升温段"),
    ("4号机", "2026-04-25 16:00:00", "持续偏高", "停机", "旧事件落在停机降温段"),
    ("4号机", "2026-04-26 17:00:00", "短时突升", "运行中", "运行中短时变化，非启停过渡"),
    ("4号机", "2026-05-30 16:00:00", "持续偏高", "停机", "旧事件落在停机降温段"),
    ("5号机", "2025-05-09 20:00:00", "混合", "停机", "旧事件落在停机降温段"),
    ("7号机", "2025-02-17 17:00:00", "持续偏高", "启动", "旧事件落在开机升温段"),
    ("8号机", "2025-01-18 10:00:00", "混合", "启动", "旧事件落在开机升温段"),
]

FIELDS = [
    "key", "title", "机组", "批次", "周期编号", "级", "过渡类型", "温度阶段",
    "特征类型", "方向", "置信度", "开始小时", "结束小时", "持续自然小时数",
    "有效运行小时数", "峰值缸", "最高温度", "最大级差", "最大级差变化",
    "最大同级偏离", "最大同级偏低", "关联通道异常", "阶段触发依据", "关联旧事件", "过渡说明",
]


def main() -> None:
    if not LEGACY.exists():
        raise SystemExit(f"缺少旧事件文件：{LEGACY}")
    with LEGACY.open(encoding="utf-8-sig", newline="") as handle:
        rows = [row for row in csv.DictReader(handle) if row.get("是否事件") == "是"]

    output = []
    for index, (unit, start, kind, transition, note) in enumerate(TRANSITIONS):
        match = next(
            (row for row in rows
             if row.get("机组") == unit and row.get("开始小时") == start and row.get("特征类型") == kind),
            None,
        )
        if match is None:
            print(f"未找到旧事件：{unit} {start} {kind}")
            continue
        level = match.get("温度阶段") or ""
        output.append({
            "key": f"transition:{index}",
            "title": f"{unit}-{match.get('级')}-{transition}过渡-{start[:16]}",
            "机组": unit,
            "批次": match.get("批次") or "",
            "周期编号": match.get("周期编号") or "",
            "级": match.get("级") or "",
            "过渡类型": transition,
            "温度阶段": level,
            "特征类型": kind,
            "方向": match.get("方向") or "",
            "置信度": match.get("置信度") or "",
            "开始小时": start,
            "结束小时": match.get("结束小时") or "",
            "持续自然小时数": match.get("持续自然小时数") or "",
            "有效运行小时数": match.get("有效运行小时数") or "",
            "峰值缸": match.get("峰值缸") or "",
            "最高温度": match.get("最高温度") or "",
            "最大级差": match.get("最大级差") or "",
            "最大级差变化": match.get("最大级差变化") or "",
            "最大同级偏离": match.get("最大同级偏离") or "",
            "最大同级偏低": match.get("最大同级偏低") or "",
            "关联通道异常": match.get("关联通道异常") or "",
            "阶段触发依据": "；".join(
                list(dict.fromkeys(
                    item for item in (match.get("阶段触发依据") or "").split("；") if item
                ))
            ),
            "关联旧事件": f"{level}-{kind}",
            "过渡说明": note,
        })

    path = OUT / "温度过渡段.csv"
    with path.open("w", encoding="utf-8-sig", newline="") as handle:
        writer = csv.DictWriter(handle, fieldnames=FIELDS)
        writer.writeheader()
        writer.writerows(output)
    print(f"written {len(output)} rows -> {path}")


if __name__ == "__main__":
    main()