Compressor/_tmp_cylinder_vibration/unified_labels.py

"""把振动小时标签映射为与压力/温度可共用的统一字段。"""
from __future__ import annotations

from pathlib import Path

import numpy as np
import pandas as pd

from . import config

CORE_FIELDS = [
    "标签ID", "标签类型", "机组", "批次", "周期编号", "阶段",
    "阶段部位", "阶段点位", "开始小时", "结束小时",
    "持续自然小时数", "有效运行小时数",
    "基线值", "正常波动带", "阶段开始值", "阶段结束值", "阶段最大值",
    "相对基线变化", "阶段最大连续异常有效小时数",
    "阶段高报警小时数", "阶段高高报警小时数",
    "方向", "特征类型", "置信度", "是否形成确认等级",
    "阶段触发依据", "事件关联", "启动过渡比例", "局部基线可用比例",
    "数据来源", "算法版本",
]
VIBRATION_FIELDS = CORE_FIELDS + [
    "近期基线RMS", "RMS正常波动带", "阶段开始RMS", "阶段结束RMS",
    "阶段最大RMS", "阶段最大峭度", "阶段最大峭度/基线",
    "阶段最大残差峭度z", "触发特征", "多特征确认",
]


def _event_hits(cylinder: str, start: pd.Timestamp, end: pd.Timestamp) -> str:
    hits = []
    for event in config.EVENTS:
        if event["cylinder"] != cylinder:
            continue
        center = pd.Timestamp(event["date"])
        if start <= center + pd.Timedelta(days=1) and end >= center - pd.Timedelta(days=7):
            hits.append(event["event_id"])
    return " / ".join(hits)


def _max_consecutive_hours(frame: pd.DataFrame) -> int:
    if frame.empty:
        return 0
    hours = frame["sample_time"].dt.floor("h").drop_duplicates().sort_values()
    if hours.empty:
        return 0
    gaps = hours.diff().dt.total_seconds().fillna(3600)
    best = current = 1
    for gap in gaps.iloc[1:]:
        if gap <= 3600:
            current += 1
            best = max(best, current)
        else:
            current = 1
    return int(best)


def build_vibration_unified(hourly: pd.DataFrame, frame: pd.DataFrame, review: pd.DataFrame) -> pd.DataFrame:
    frame = frame.copy()
    frame["sample_time"] = pd.to_datetime(frame["sample_time"])
    review_index = review.set_index(["cylinder", "start_hour"])
    rows = []
    for index, item in enumerate(hourly.itertuples(index=False), start=1):
        part = frame[
            (frame["cylinder"] == item.cylinder)
            & (frame["sample_time"] >= item.start_hour)
            & (frame["sample_time"] < item.end_hour)
        ].sort_values("sample_time")
        if part.empty:
            continue
        peak = part.loc[part["score"].idxmax()]
        base_log = float(part["base_rms_log"].median())
        scale_log = float(part["scale_rms_log"].median())
        base_rms = float(10 ** base_log) if np.isfinite(base_log) else float(part["rms"].median())
        low = float(10 ** (base_log - 0.6745 * scale_log)) if np.isfinite(scale_log) else base_rms * 0.9
        high = float(10 ** (base_log + 0.6745 * scale_log)) if np.isfinite(scale_log) else base_rms * 1.1
        rms_max = float(part["rms"].max())
        kurt_max = float(part["kurt"].max())
        kurt_ratio_max = float(part["kurt_ratio"].max())
        z_kurt = float(part["z_resid_kurt"].abs().max())
        review_row = review_index.loc[(item.cylinder, item.start_hour)].to_dict() if (item.cylinder, item.start_hour) in review_index.index else {}
        transition_ratio = review_row.get("transition_ratio", item.transition_ratio)
        baseline_ok_ratio = review_row.get("baseline_ok_ratio", 1.0)
        confirmed = bool(item.confirmed)
        event_hits = _event_hits(item.cylinder, item.start_hour, item.end_hour)
        relative = (rms_max / base_rms - 1.0) if base_rms else 0.0
        trigger = (
            f"5号机{item.cylinder} {item.start_hour:%Y-%m-%d %H:%M}~{item.end_hour:%Y-%m-%d %H:%M}："
            f"{item.direction}；RMS峰值{rms_max:.2f}（近期基线{base_rms:.2f}，{relative:+.1%}），"
            f"实际峭度最大{kurt_max:.2f}（局部基线{float(part['base_kurt'].median()):.2f}，"
            f"{kurt_ratio_max:.2f}倍），残差峭度z={z_kurt:.2f}，持续{item.hour_count}小时；"
            f"多特征确认={'是' if confirmed else '否'}"
            + (f"；事件关联={event_hits}" if event_hits else "")
        )
        high_hours = item.hour_count if item.level in ("异常", "严重异常") else 0
        high_high_hours = item.hour_count if item.level == "严重异常" else 0
        rows.append(
            {
                "标签ID": f"5号机-VIB-{index:04d}",
                "标签类型": "振动",
                "机组": "5号机",
                "批次": "全量",
                "周期编号": f"5号机-VIB-{index:04d}",
                "阶段": item.level,
                "阶段部位": f"5号机{item.cylinder}",
                "阶段点位": str(peak["point_name"]),
                "开始小时": item.start_hour,
                "结束小时": item.end_hour,
                "持续自然小时数": item.hour_count,
                "有效运行小时数": int(part["sample_time"].dt.floor("h").nunique()),
                "基线值": round(base_rms, 4),
                "正常波动带": f"{low:.4f}~{high:.4f}",
                "阶段开始值": round(float(part["rms"].iloc[0]), 4),
                "阶段结束值": round(float(part["rms"].iloc[-1]), 4),
                "阶段最大值": round(rms_max, 4),
                "相对基线变化": round(relative, 4),
                "阶段最大连续异常有效小时数": _max_consecutive_hours(part[part["hit"]]),
                "阶段高报警小时数": high_hours,
                "阶段高高报警小时数": high_high_hours,
                "方向": item.direction,
                "特征类型": "振动",
                "置信度": "高" if item.level == "严重异常" else ("中" if item.level == "异常" else "低"),
                "是否形成确认等级": "是" if confirmed else "否",
                "阶段触发依据": trigger,
                "事件关联": event_hits,
                "启动过渡比例": round(float(transition_ratio or 0), 2),
                "局部基线可用比例": round(float(baseline_ok_ratio or 1), 2),
                "数据来源": "wave_file 首周期波形",
                "算法版本": "cylinder_vibration_v1",
                "近期基线RMS": round(base_rms, 4),
                "RMS正常波动带": f"{low:.4f}~{high:.4f}",
                "阶段开始RMS": round(float(part["rms"].iloc[0]), 4),
                "阶段结束RMS": round(float(part["rms"].iloc[-1]), 4),
                "阶段最大RMS": round(rms_max, 4),
                "阶段最大峭度": round(kurt_max, 3),
                "阶段最大峭度/基线": round(kurt_ratio_max, 3),
                "阶段最大残差峭度z": round(z_kurt, 3),
                "触发特征": str(peak["driver"]),
                "多特征确认": "是" if confirmed else "否",
            }
        )
    return pd.DataFrame(rows, columns=VIBRATION_FIELDS)


def _map_external(path: Path, label_type: str, prefix: str, unit: str | None = None) -> pd.DataFrame:
    if not path.exists():
        return pd.DataFrame(columns=CORE_FIELDS)
    frame = pd.read_csv(path, encoding="utf-8-sig")
    if unit and "机组" in frame.columns:
        frame = frame[frame["机组"].astype(str).str.strip() == unit].copy().reset_index(drop=True)
    if frame.empty:
        return pd.DataFrame(columns=CORE_FIELDS)
    stage_column = "压力阶段" if label_type == "润滑油压力" else "温度阶段"
    if stage_column not in frame:
        stage_column = "振动阶段"
    result = pd.DataFrame()
    result["标签ID"] = [f"{prefix}-{i:04d}" for i in range(1, len(frame) + 1)]
    result["标签类型"] = label_type
    result["机组"] = frame.get("机组", "")
    result["批次"] = frame.get("批次", "")
    result["周期编号"] = frame.get("周期编号", "")
    result["阶段"] = frame.get(stage_column, "")
    result["阶段部位"] = frame.get("阶段部位", frame.get("级", ""))
    result["阶段点位"] = frame.get("阶段点位", "")
    result["开始小时"] = frame.get("开始小时", "")
    result["结束小时"] = frame.get("结束小时", "")
    result["持续自然小时数"] = frame.get("持续自然小时数", "")
    result["有效运行小时数"] = frame.get("有效运行小时数", "")
    result["方向"] = frame.get("方向", "")
    result["特征类型"] = frame.get("特征类型", "")
    result["置信度"] = frame.get("置信度", "")
    result["是否形成确认等级"] = frame.get("是否形成确认等级", "")
    result["阶段触发依据"] = frame.get("阶段触发依据", "")
    result["数据来源"] = str(path.name)
    for column in CORE_FIELDS:
        if column not in result:
            result[column] = ""
    return result[CORE_FIELDS]


def build_combined_preview(out_dir: Path) -> pd.DataFrame:
    pressure = _map_external(
        Path("/Users/zhouhao/Code/Python/Compressor/models/results/pressure/润滑油压力小时阶段段.csv"),
        "润滑油压力",
        "P",
    )
    temperature = _map_external(
        Path("/Users/zhouhao/Code/Python/Compressor/_tmp_temperature/out/温度异常段.csv"),
        "温度异常",
        "T",
    )
    vibration = pd.read_csv(out_dir / "cylinder_vibration_label_unified.csv", parse_dates=["开始小时", "结束小时"])
    vibration_core = vibration.reindex(columns=CORE_FIELDS)
    return pd.concat([pressure, temperature, vibration_core], ignore_index=True)