Compressor/_tmp_cylinder_vibration/run_adaptive.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 adaptive, config, html_export, labels


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"
    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)
    frame = adaptive.add_ratios(frame, baseline)
    point_labels = pd.read_csv(points_path)
    point_labels["max_time"] = pd.to_datetime(point_labels["max_time"], errors="coerce")

    result = adaptive.build_intervals(frame, point_labels, baseline)
    result_path = config.OUTPUT_DIR / "vibration_intervals_adaptive.csv"
    result.to_csv(result_path, index=False, date_format="%Y-%m-%d %H:%M:%S")

    validation = adaptive.validate_events(result)
    validation_path = config.OUTPUT_DIR / "event_validation_adaptive.csv"
    validation.to_csv(validation_path, index=False, date_format="%Y-%m-%d %H:%M:%S")
    html_export.export_html(frame, result, validation)

    event_windows = []
    for event in config.EVENTS:
        center = pd.Timestamp(event["date"])
        event_windows.append((center - pd.Timedelta(days=7), center + pd.Timedelta(days=1)))
    if result.empty:
        false_all = false_conf = result
    else:
        related = result.apply(
            lambda row: any(row["start_time"] <= end and row["end_time"] >= start for start, end in event_windows),
            axis=1,
        )
        false_all = result[~related]
        false_conf = false_all[false_all["multi_feature_confirmed"]]
    summary = {
        "false_alarm_intervals": int(len(false_all)),
        "false_alarm_hours": round(float(false_all["duration_hours"].sum()), 2) if not false_all.empty else 0.0,
        "confirmed_false_alarm_intervals": int(len(false_conf)),
        "confirmed_false_alarm_hours": round(float(false_conf["duration_hours"].sum()), 2) if not false_conf.empty else 0.0,
        "intervals": len(result),
        "confirmed_intervals": int(result["multi_feature_confirmed"].sum()) if not result.empty else 0,
        "event_hits": int(validation["hit"].sum()) if not validation.empty else 0,
        "confirmed_event_hits": int(validation["confirmed"].sum()) if not validation.empty else 0,
        "validation": validation.to_dict("records"),
    }
    summary_path = config.OUTPUT_DIR / "adaptive_summary.json"
    summary_path.write_text(json.dumps(summary, ensure_ascii=False, indent=2, default=str), encoding="utf-8")
    print(result.to_string())
    print(json.dumps(summary, ensure_ascii=False, indent=2, default=str))


if __name__ == "__main__":
    main()