Compressor/_tmp_cylinder_vibration/run_machine_events.py

"""生成整机多缸振动事件标签、验证表和摘要。"""
from __future__ import annotations

import json
import sys
from pathlib import Path

import pandas as pd

ROOT = Path(__file__).resolve().parents[1]
if str(ROOT) not in sys.path:
    sys.path.insert(0, str(ROOT))

from _tmp_cylinder_vibration import config, machine_event_html, machine_events, unified_labels


def main() -> None:
    frame = pd.read_csv(config.OUTPUT_DIR / "vibration_features_full_local.csv")
    frame["sample_time"] = pd.to_datetime(frame["sample_time"])
    labels = pd.read_csv(config.OUTPUT_DIR / "cylinder_vibration_label_unified.csv")
    hourly = pd.read_csv(config.OUTPUT_DIR / "vibration_hourly_labels.csv")
    hourly["start_hour"] = pd.to_datetime(hourly["start_hour"])

    runs = machine_events.build_runs(frame)
    events, candidates, windows = machine_events.build_machine_events(labels, frame, hourly, runs)
    validation = machine_events.validate_events(events)
    summary = machine_events.build_summary(events, candidates, validation, runs)

    events.to_csv(
        config.OUTPUT_DIR / "machine_vibration_events.csv",
        index=False,
        date_format="%Y-%m-%d %H:%M:%S",
    )
    candidates.to_csv(
        config.OUTPUT_DIR / "machine_event_candidates.csv",
        index=False,
        date_format="%Y-%m-%d %H:%M:%S",
    )
    validation.to_csv(
        config.OUTPUT_DIR / "machine_event_validation.csv",
        index=False,
        date_format="%Y-%m-%d %H:%M:%S",
    )
    (config.OUTPUT_DIR / "machine_event_summary.json").write_text(
        json.dumps(summary, ensure_ascii=False, indent=2, default=str),
        encoding="utf-8",
    )

    temperature = unified_labels._map_external(
        Path("/Users/zhouhao/Code/Python/Compressor/_tmp_temperature/out/温度异常段.csv"),
        "温度异常",
        "T",
        unit="5号机",
    )
    machine = events.reindex(columns=unified_labels.CORE_FIELDS)
    combined = pd.concat([temperature, machine], ignore_index=True)
    combined.to_csv(
        config.OUTPUT_DIR / "fault_label_unified_machine_preview.csv",
        index=False,
        date_format="%Y-%m-%d %H:%M:%S",
    )


    lines = [
        "# 5 号机缸体振动整机事件标签",
        "",
        "## 规则",
        "",
        "- 稳定运行段：开机后 6 小时、停机前 6 小时不生成最终事件；",
        "- 不排除首次基线预热段；4 月以前只是未取得数据；",
        "- 24 小时窗口内至少 4 个不同缸出现候选标签；",
        "- 其中至少 2 条达到“异常”或“严重异常”；",
        "- 相邻候选窗口合并为一个整机事件；",
        "- 事件不指定故障缸，故障缸只用于事后验证。",
        "",
        "## 结果",
        "",
        f"- 候选标签：{len(candidates)} 条；",
        f"- 整机事件：{len(events)} 个；",
        f"- 运行段：{len(runs)} 个；",
        f"- 有事件运行段：{summary['event_runs']} 个；",
        f"- 无事件运行段：{summary['normal_runs']} 个；",
        f"- 验证事件：{summary['validation']['hit']}/{summary['validation']['total']} 命中。",
        "",
        "## 事件列表",
        "",
    ]
    lines.append("| 标签ID | 开始 | 结束 | 等级 | 参与缸 | 异常缸 | 严重异常缸 | 验证提前量 |")
    lines.append("|---|---|---|---|---|---:|---:|---:|")
    lead_map = {
        row.machine_label_id: row.lead_days
        for row in validation.itertuples(index=False)
        if row.hit
    }
    for row in events.itertuples(index=False):
        lead = lead_map.get(row.标签ID)
        lines.append(
            f"| {row.标签ID} | {row.开始小时} | {row.结束小时} | {row.阶段} | "
            f"{row.参与缸} | {row.异常缸数} | {row.严重异常缸数} | "
            f"{'—' if lead is None else f'{lead:.2f} 天'} |"
        )
    lines.extend([
        "",
        "## 验证",
        "",
        "| 事件 | 故障缸 | 命中 | 整机事件 | 开始 | 结束 | 提前量 |",
        "|---|---|---|---|---|---|---:|",
    ])
    for row in validation.itertuples(index=False):
        lines.append(
            f"| {row.event_id} | {row.fault_cylinder} | {'是' if row.hit else '否'} | "
            f"{row.machine_label_id} | {row.start_time} | {row.end_time} | "
            f"{'—' if not row.hit else f'{row.lead_days:.2f} 天'} |"
        )
    (config.OUTPUT_DIR / "machine_event_labels.md").write_text("\n".join(lines) + "\n", encoding="utf-8")
    machine_event_html.export_machine_events(frame, events, candidates, validation)

    print(f"[machine] candidates={len(candidates)}, events={len(events)}, runs={len(runs)}, event_runs={summary['event_runs']}, preview={len(combined)}")
    print(f"[machine] validation={summary['validation']['hit']}/{summary['validation']['total']}")
    print(events[["标签ID", "开始小时", "结束小时", "阶段", "参与缸", "异常缸数", "严重异常缸数"]].to_string(index=False))


if __name__ == "__main__":
    main()