Compressor/_tmp_cylinder_vibration/machine_events.py

"""整机多缸振动事件标签。

最终标签不再按单缸输出，而是按稳定运行段内的整机多缸聚集事件输出：

- 排除开机后 6 小时、停机前 6 小时；
- 基线至少成熟 7 天；
- 24 小时窗口内至少 4 个不同缸有候选标签；
- 其中至少 2 条达到“异常”或“严重异常”；
- 相邻候选窗口合并为同一个整机事件；
- 事件只描述“整机可能异常”，不指定故障缸。
"""
from __future__ import annotations

from datetime import timedelta
from pathlib import Path

import numpy as np
import pandas as pd

from . import config, unified_labels

SEVERITY = {"轻微": 1, "异常": 2, "严重异常": 3}
LEVELS = ("轻微", "异常", "严重异常")
BAD_LEVELS = {"异常", "严重异常"}

MACHINE_FIELDS = unified_labels.CORE_FIELDS + [
    "参与缸数", "参与缸", "异常缸数", "严重异常缸数",
    "异常标签数", "严重异常标签数", "候选标签数",
    "最大异常分数", "最大RMS", "最大峭度",
    "转速范围", "运行段开始", "运行段结束", "整机聚集窗口小时",
]


def build_runs(frame: pd.DataFrame, gap_hours: float = 2.0) -> pd.DataFrame:
    frame = frame.copy()
    frame["sample_time"] = pd.to_datetime(frame["sample_time"])
    running = frame.loc[frame["rpm"].astype(float) > 0, ["sample_time"]].drop_duplicates()
    running = running.sort_values("sample_time").reset_index(drop=True)
    if running.empty:
        return pd.DataFrame(columns=["run_id", "start", "end", "hours"])
    running["gap_hours"] = running["sample_time"].diff().dt.total_seconds() / 3600.0
    running["run_id"] = (running["gap_hours"] > gap_hours).cumsum().astype(int)
    rows = []
    for run_id, group in running.groupby("run_id", sort=True):
        start = pd.Timestamp(group["sample_time"].min())
        end = pd.Timestamp(group["sample_time"].max())
        rows.append({
            "run_id": int(run_id),
            "start": start,
            "end": end,
            "hours": round((end - start).total_seconds() / 3600.0, 2),
        })
    return pd.DataFrame(rows)


def _run_index(timestamp: pd.Timestamp, runs: pd.DataFrame, tolerance_hours: float = 1.0) -> int | None:
    tolerance = pd.Timedelta(hours=tolerance_hours)
    for row in runs.itertuples(index=False):
        if row.start - tolerance <= timestamp <= row.end + tolerance:
            return int(row.run_id)
    return None


def prepare_candidates(
    labels: pd.DataFrame,
    frame: pd.DataFrame,
    runs: pd.DataFrame,
    guard_start_hours: int = 6,
    guard_end_hours: int = 6,
) -> pd.DataFrame:
    labels = labels.copy()
    labels["开始小时"] = pd.to_datetime(labels["开始小时"])
    labels["结束小时"] = pd.to_datetime(labels["结束小时"])
    labels["cylinder"] = labels["阶段部位"].astype(str).str.replace("5号机", "", regex=False)

    run_by_id = runs.set_index("run_id")

    rows = []
    for row in labels.itertuples(index=False):
        run_id = _run_index(pd.Timestamp(row.开始小时), runs)
        if run_id is None:
            continue
        run = run_by_id.loc[run_id]
        start = pd.Timestamp(row.开始小时)
        end = pd.Timestamp(row.结束小时)
        stable = bool(start >= run["start"] + pd.Timedelta(hours=guard_start_hours)
                      and end <= run["end"] - pd.Timedelta(hours=guard_end_hours))
        rows.append({
            **row._asdict(),
            "run_id": run_id,
            "run_start": run["start"],
            "run_end": run["end"],
            "stable": stable,
        })
    result = pd.DataFrame(rows)
    if result.empty:
        return result
    return result


def _merge_candidate_windows(candidates: list[dict], merge_gap_hours: int = 6) -> list[dict]:
    if not candidates:
        return []
    candidates = sorted(candidates, key=lambda item: item["window_start"])
    merged = [candidates[0].copy()]
    for candidate in candidates[1:]:
        current = merged[-1]
        if candidate["window_start"] <= current["window_end"] + pd.Timedelta(hours=merge_gap_hours):
            current["window_end"] = max(current["window_end"], candidate["window_end"])
            current["indices"] = sorted(set(current["indices"]) | set(candidate["indices"]))
        else:
            merged.append(candidate.copy())
    return merged


def _ordered_cylinders(values: list[str]) -> list[str]:
    present = set(values)
    return [name for name in config.CYLINDERS if name in present]


def build_machine_events(
    labels: pd.DataFrame,
    frame: pd.DataFrame,
    hourly: pd.DataFrame,
    runs: pd.DataFrame,
    guard_start_hours: int = 6,
    guard_end_hours: int = 6,
    window_hours: int = 24,
    min_cylinders: int = 4,
    min_bad_labels: int = 2,
    merge_gap_hours: int = 6,
) -> tuple[pd.DataFrame, pd.DataFrame, pd.DataFrame]:
    candidates = prepare_candidates(
        labels, frame, runs,
        guard_start_hours=guard_start_hours,
        guard_end_hours=guard_end_hours,
    )
    if candidates.empty:
        return pd.DataFrame(columns=MACHINE_FIELDS), candidates, pd.DataFrame()
    candidates = candidates[candidates["stable"]].copy()
    candidates = candidates.sort_values(["开始小时", "cylinder"]).reset_index(drop=True)
    candidates["severity_num"] = candidates["阶段"].map(SEVERITY).fillna(0).astype(int)

    score_map = {
        (row.cylinder, pd.Timestamp(row.start_hour)): float(row.peak_score)
        for row in hourly.itertuples(index=False)
    }

    windows = []
    for index, row in candidates.iterrows():
        start = pd.Timestamp(row["开始小时"])
        end = start + pd.Timedelta(hours=window_hours)
        part = candidates[(candidates["结束小时"] >= start) & (candidates["开始小时"] <= end)]
        if part["cylinder"].nunique() < min_cylinders:
            continue
        if int(part["阶段"].isin(BAD_LEVELS).sum()) < min_bad_labels:
            continue
        windows.append({
            "window_start": start,
            "window_end": end,
            "indices": part.index.tolist(),
        })
    windows = _merge_candidate_windows(windows, merge_gap_hours=merge_gap_hours)

    frame = frame.copy()
    frame["sample_time"] = pd.to_datetime(frame["sample_time"])
    rows = []
    for index, window in enumerate(windows, start=1):
        part = candidates.loc[window["indices"]].copy()
        start = pd.Timestamp(part["开始小时"].min())
        end = pd.Timestamp(part["结束小时"].max())
        cylinders = _ordered_cylinders(part["cylinder"].tolist())
        bad = part[part["阶段"].isin(BAD_LEVELS)]
        severe = part[part["阶段"] == "严重异常"]
        bad_cylinders = _ordered_cylinders(bad["cylinder"].tolist())
        severe_cylinders = _ordered_cylinders(severe["cylinder"].tolist())
        max_level = max(part["阶段"], key=lambda value: SEVERITY.get(value, 0))
        peak_score = max(
            (score_map.get((row.cylinder, pd.Timestamp(row.开始小时)), np.nan) for row in part.itertuples(index=False)),
            default=np.nan,
        )
        rpm_part = frame[
            (frame["sample_time"] >= start)
            & (frame["sample_time"] <= end)
            & (frame["cylinder"].isin(cylinders))
        ]
        rpm_min = float(rpm_part["rpm"].min()) if not rpm_part.empty else np.nan
        rpm_max = float(rpm_part["rpm"].max()) if not rpm_part.empty else np.nan
        rms_col = pd.to_numeric(part.get("阶段最大RMS"), errors="coerce")
        kurt_col = pd.to_numeric(part.get("阶段最大峭度"), errors="coerce")
        rms_max = float(rms_col.max()) if rms_col.notna().any() else np.nan
        kurt_max = float(kurt_col.max()) if kurt_col.notna().any() else np.nan
        run_start = pd.Timestamp(part["run_start"].min())
        run_end = pd.Timestamp(part["run_end"].max())
        hours = round((end - start).total_seconds() / 3600.0, 2)
        bad_hours = int(bad["持续自然小时数"].sum()) if not bad.empty else 0
        severe_hours = int(severe["持续自然小时数"].sum()) if not severe.empty else 0
        trigger = (
            f"5号机整机稳定运行段：{start:%Y-%m-%d %H:%M}~{end:%Y-%m-%d %H:%M}，"
            f"24小时窗口内{len(cylinders)}个缸出现候选标签（{'、'.join(cylinders)}），"
            f"其中{len(bad)}条达到异常/严重异常（{'、'.join(bad_cylinders)}）；"
            f"最大等级{max_level}，最大异常分数{peak_score:.2f}；"
            f"参与缸数={len(cylinders)}，异常缸数={len(bad_cylinders)}"
        )
        rows.append({
            "标签ID": f"5号机-MACH-{index:04d}",
            "标签类型": "振动",
            "机组": "5号机",
            "批次": "全量",
            "周期编号": f"5号机-MACH-{index:04d}",
            "阶段": max_level,
            "阶段部位": "5号机整机",
            "阶段点位": "5号机一缸~六缸缸体振动",
            "开始小时": start,
            "结束小时": end,
            "持续自然小时数": hours,
            "有效运行小时数": int(part["持续自然小时数"].sum()),
            "基线值": np.nan,
            "正常波动带": "",
            "阶段开始值": np.nan,
            "阶段结束值": np.nan,
            "阶段最大值": rms_max,
            "相对基线变化": np.nan,
            "阶段最大连续异常有效小时数": bad_hours,
            "阶段高报警小时数": bad_hours,
            "阶段高高报警小时数": severe_hours,
            "方向": "整机多缸",
            "特征类型": "整机振动",
            "置信度": "高" if severe_cylinders else "中",
            "是否形成确认等级": "是",
            "阶段触发依据": trigger,
            "事件关联": "",
            "启动过渡比例": 0.0,
            "局部基线可用比例": 1.0,
            "数据来源": "wave_file 首周期波形",
            "算法版本": "cylinder_vibration_machine_v2",
            "参与缸数": len(cylinders),
            "参与缸": " / ".join(cylinders),
            "异常缸数": len(bad_cylinders),
            "严重异常缸数": len(severe_cylinders),
            "异常标签数": int(len(bad)),
            "严重异常标签数": int(len(severe)),
            "候选标签数": int(len(part)),
            "最大异常分数": round(peak_score, 3) if np.isfinite(peak_score) else np.nan,
            "最大RMS": round(rms_max, 4) if np.isfinite(rms_max) else np.nan,
            "最大峭度": round(kurt_max, 3) if np.isfinite(kurt_max) else np.nan,
            "转速范围": f"{rpm_min:.1f}~{rpm_max:.1f}" if np.isfinite(rpm_min) and np.isfinite(rpm_max) else "",
            "运行段开始": run_start,
            "运行段结束": run_end,
            "整机聚集窗口小时": window_hours,
        })
    result = pd.DataFrame(rows, columns=MACHINE_FIELDS)
    return result, candidates, pd.DataFrame(windows)


def validate_events(events: pd.DataFrame, events_config: list[dict] | None = None) -> pd.DataFrame:
    rows = []
    for event in events_config or config.EVENTS:
        event_id = event["event_id"]
        event_date = pd.Timestamp(event["date"])
        window_start = event_date - pd.Timedelta(days=7)
        window_end = event_date + pd.Timedelta(days=1)
        matched = events[
            (events["开始小时"] <= window_end)
            & (events["结束小时"] >= window_start)
        ].sort_values("开始小时") if not events.empty else events
        if matched.empty:
            rows.append({
                "event_id": event_id,
                "event_date": event["date"],
                "fault_cylinder": event["cylinder"],
                "hit": False,
                "machine_label_id": "",
                "start_time": pd.NaT,
                "end_time": pd.NaT,
                "duration_hours": np.nan,
                "lead_days": np.nan,
                "max_level": "",
                "participating_cylinders": "",
                "abnormal_cylinders": "",
            })
            continue
        matched_event = matched.iloc[0]
        start = pd.Timestamp(matched_event["开始小时"])
        rows.append({
            "event_id": event_id,
            "event_date": event["date"],
            "fault_cylinder": event["cylinder"],
            "hit": True,
            "machine_label_id": matched_event["标签ID"],
            "start_time": start,
            "end_time": pd.Timestamp(matched_event["结束小时"]),
            "duration_hours": float(matched_event["持续自然小时数"]),
            "lead_days": round((event_date - start).total_seconds() / 86400.0, 2),
            "max_level": matched_event["阶段"],
            "participating_cylinders": matched_event["参与缸"],
            "abnormal_cylinders": matched_event["异常缸数"],
        })
    return pd.DataFrame(rows)


def build_summary(events: pd.DataFrame, candidates: pd.DataFrame, validation: pd.DataFrame, runs: pd.DataFrame) -> dict:
    event_run_ids = sorted(set(events["运行段开始"].dropna().tolist()))
    all_run_ids = set(runs["run_id"].tolist())
    return {
        "rule": {
            "stable_start_hours": 6,
            "stable_end_hours": 6,
            "window_hours": 24,
            "min_cylinders": 4,
            "min_bad_labels": 2,
            "merge_gap_hours": 6,
        },
        "candidate_labels": int(len(candidates)),
        "events": int(len(events)),
        "runs": int(len(runs)),
        "event_run_ids": event_run_ids,
        "event_runs": int(len(event_run_ids)),
        "normal_runs": int(len(all_run_ids) - len(event_run_ids)),
        "validation": {
            "total": int(len(validation)),
            "hit": int(validation["hit"].sum()) if not validation.empty else 0,
        },
    }