Compressor/_tmp_cylinder_vibration/run_intervals.py

"""第二轮：单点候选 -> 连续时间段 + 多缸实际曲线。

用法：
  python3 _tmp_cylinder_vibration/run_intervals.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, data, intervals, labels, plots_actual


def main() -> None:
    features_path = config.OUTPUT_DIR / "vibration_features_refined.csv"
    points_path = config.OUTPUT_DIR / "vibration_labels_refined.csv"
    baseline_path = config.OUTPUT_DIR / "baseline.json"
    if not features_path.exists() or not points_path.exists() or not baseline_path.exists():
        raise SystemExit("请先运行 run.py --stage all")

    frame = labels.prepare_frame(pd.read_csv(features_path).to_dict("records"), 15)
    baseline = json.loads(baseline_path.read_text(encoding="utf-8"))
    frame, _, _ = labels.build_labels(frame, 15, baseline=baseline)

    point_labels = pd.read_csv(points_path)
    for column in ("start_time", "end_time", "max_time"):
        if column in point_labels:
            point_labels[column] = pd.to_datetime(point_labels[column], errors="coerce")

    result = intervals.build_intervals(frame, point_labels)
    result_path = config.OUTPUT_DIR / "vibration_intervals_refined.csv"
    result.to_csv(result_path, index=False, date_format="%Y-%m-%d %H:%M:%S")
    print(f"[intervals] {len(result)} -> {result_path}")

    temperature = data.load_temperature_reference()
    validation = intervals.validate_events(result, temperature)
    validation_path = config.OUTPUT_DIR / "event_validation_intervals.csv"
    validation.to_csv(validation_path, index=False, date_format="%Y-%m-%d %H:%M:%S")

    for event in config.EVENTS:
        plots_actual.plot_feature_grid(frame, result, baseline, event)
        plots_actual.plot_waveform_grid(frame, result, event)

    summary = {
        "point_labels": len(point_labels),
        "intervals": len(result),
        "event_hits": int(validation["vibration_hit"].sum()),
        "false_alarm_intervals": int(validation.attrs.get("false_alarm_intervals", 0)),
        "false_alarm_hours": float(validation.attrs.get("false_alarm_hours", 0.0)),
        "event_validation": validation.to_dict("records"),
    }
    summary_path = config.OUTPUT_DIR / "interval_summary.json"
    summary_path.write_text(json.dumps(summary, ensure_ascii=False, indent=2, default=str), encoding="utf-8")
    print(json.dumps(summary, ensure_ascii=False, indent=2, default=str))
    print(f"output: {config.OUTPUT_DIR}")


if __name__ == "__main__":
    main()