Compressor/_tmp_cylinder_vibration/review.py

"""输出小时标签的排查指标，便于人工判断标签是否合理。"""
from __future__ import annotations

import numpy as np
import pandas as pd

from . import config


def _slope(values: pd.Series, times: pd.Series) -> float:
    if len(values) < 2:
        return 0.0
    x = (pd.to_datetime(times) - pd.to_datetime(times).iloc[0]).dt.total_seconds() / 3600.0
    x = np.asarray(x, dtype=float)
    y = np.asarray(values, dtype=float)
    if np.ptp(x) <= 0 or not np.isfinite(y).any():
        return 0.0
    return float(np.polyfit(x, y, 1)[0])


def build_review(hourly_labels: pd.DataFrame, frame: pd.DataFrame) -> pd.DataFrame:
    if hourly_labels.empty:
        return pd.DataFrame()
    frame = frame.copy()
    frame["sample_time"] = pd.to_datetime(frame["sample_time"])
    rows = []
    for item in hourly_labels.itertuples(index=False):
        group = frame[frame["cylinder"] == item.cylinder]
        part = group[(group["sample_time"] >= item.start_hour) & (group["sample_time"] < item.end_hour)]
        pre = group[(group["sample_time"] >= item.start_hour - pd.Timedelta(hours=6)) & (group["sample_time"] < item.start_hour)]
        if part.empty:
            continue
        score_max = float(part["score"].max())
        score_mean = float(part["score"].mean())
        rms_max = float(part["rms_ratio"].max())
        rms_mean = float(part["rms_ratio"].mean())
        kurt_max = float(part["kurt"].max())
        kurt_ratio_max = float(part["kurt_ratio"].max())
        kurt_ratio_mean = float(part["kurt_ratio"].mean())
        active_rms = float((part["rms_ratio"] >= 1.10).mean())
        active_kurt = float((part["kurt_ratio"] >= 1.50).mean())
        transition_ratio = float(part["transition"].mean()) if "transition" in part else 0.0
        baseline_ok_ratio = float(part["baseline_ok"].mean()) if "baseline_ok" in part else 1.0
        flags = []
        if not bool(item.confirmed):
            flags.append("未确认")
        if rms_max < 1.10 and kurt_ratio_max >= 1.50:
            flags.append("低幅高冲击")
        if score_max >= 4.50 and rms_max < 1.10 and kurt_ratio_max < 1.50:
            flags.append("分数驱动不明")
        if active_kurt < 0.30 and active_rms < 0.30:
            flags.append("持续证据弱")
        if transition_ratio >= 0.25:
            flags.append("启动/过渡")
        if baseline_ok_ratio < 0.50:
            flags.append("局部基线不足")
        rows.append(
            {
                "cylinder": item.cylinder,
                "start_hour": item.start_hour,
                "end_hour": item.end_hour,
                "hour_count": item.hour_count,
                "score_mean": round(score_mean, 3),
                "score_max": round(score_max, 3),
                "rms_ratio_mean": round(rms_mean, 3),
                "rms_ratio_max": round(rms_max, 3),
                "kurt_ratio_mean": round(kurt_ratio_mean, 3),
                "kurt_ratio_max": round(kurt_ratio_max, 3),
                "kurt_max": round(kurt_max, 3),
                "active_rms": round(active_rms, 2),
                "active_kurt": round(active_kurt, 2),
                "transition_ratio": round(transition_ratio, 2),
                "baseline_ok_ratio": round(baseline_ok_ratio, 2),
                "pre6h_score_slope": round(_slope(pre["score"], pre["sample_time"]), 4),
                "pre6h_kurt_slope": round(_slope(pre["kurt_ratio"], pre["sample_time"]), 4),
                "direction": item.direction,
                "level": item.level,
                "confirmed": bool(item.confirmed),
                "review_flags": " / ".join(flags),
            }
        )
    return pd.DataFrame(rows)