Compressor/_tmp_cylinder_vibration/hourly.py

"""把 15 分钟自适应标签汇总成小时标签，并合并相邻小时。"""
from __future__ import annotations

from datetime import timedelta

import pandas as pd

from . import config


def build_hourly(intervals: pd.DataFrame, frame: pd.DataFrame | None = None) -> pd.DataFrame:
    if intervals.empty:
        return pd.DataFrame()
    rows = intervals.copy()
    rows["start_time"] = pd.to_datetime(rows["start_time"])
    rows["end_time"] = pd.to_datetime(rows["end_time"])
    frame_by_cylinder = {}
    if frame is not None:
        frame = frame.copy()
        frame["sample_time"] = pd.to_datetime(frame["sample_time"])
        frame_by_cylinder = {name: group.sort_values("sample_time") for name, group in frame.groupby("cylinder")}

    buckets: dict[tuple[str, pd.Timestamp], dict] = {}
    for row in rows.itertuples(index=False):
        hour = pd.Timestamp(row.start_time).floor("h")
        end = pd.Timestamp(row.end_time)
        hours = []
        if row.cylinder in frame_by_cylinder:
            part = frame_by_cylinder[row.cylinder]
            part = part[(part["sample_time"] >= row.start_time) & (part["sample_time"] <= end)]
            counts = part.groupby(part["sample_time"].dt.floor("h")).size()
            hours = [(pd.Timestamp(hour_value), int(count)) for hour_value, count in counts.items() if count >= 2]
        if not hours and frame is None:
            # 没有明细时退回“任意重叠”，兼容旧调用。
            cursor = hour
            while cursor < end:
                hours.append((cursor, 0))
                cursor += timedelta(hours=1)
        for hour, point_count in hours:
            key = (row.cylinder, hour)
            current = buckets.get(key)
            candidate = {
                "cylinder": row.cylinder,
                "stage": row.stage,
                "hour": hour,
                "max_score": float(row.max_score),
                "direction": row.direction,
                "level": row.level,
                "confirmed": bool(row.multi_feature_confirmed),
                "start_time": pd.Timestamp(row.start_time),
                "end_time": pd.Timestamp(row.end_time),
            }
            candidate["point_count"] = point_count
            if current is None or candidate["max_score"] > current["max_score"]:
                buckets[key] = candidate
            else:
                current["confirmed"] = current["confirmed"] or candidate["confirmed"]
                current["point_count"] = max(current.get("point_count", 0), point_count)

    ordered = sorted(buckets.values(), key=lambda item: (item["cylinder"], item["hour"]))
    merged: list[dict] = []
    for item in ordered:
        if (
            merged
            and merged[-1]["cylinder"] == item["cylinder"]
            and merged[-1]["end_hour"] == item["hour"]
        ):
            merged[-1]["end_hour"] = item["hour"] + timedelta(hours=1)
            merged[-1]["max_score"] = max(merged[-1]["max_score"], item["max_score"])
            merged[-1]["confirmed"] = merged[-1]["confirmed"] or item["confirmed"]
            if item["max_score"] > merged[-1]["peak_score"]:
                merged[-1]["peak_score"] = item["max_score"]
                merged[-1]["direction"] = item["direction"]
                merged[-1]["level"] = item["level"]
        else:
            merged.append(
                {
                    "cylinder": item["cylinder"],
                    "stage": item["stage"],
                    "start_hour": item["hour"],
                    "end_hour": item["hour"] + timedelta(hours=1),
                    "max_score": item["max_score"],
                    "peak_score": item["max_score"],
                    "direction": item["direction"],
                    "level": item["level"],
                    "confirmed": item["confirmed"],
                }
            )

    for item in merged:
        item["hour_count"] = int((item["end_hour"] - item["start_hour"]).total_seconds() / 3600)
        group = frame_by_cylinder.get(item["cylinder"])
        if group is not None:
            part = group[(group["sample_time"] >= item["start_hour"]) & (group["sample_time"] < item["end_hour"])]
            previous = group[group["sample_time"] < item["start_hour"]].tail(1)
            item["gap_before_hours"] = (
                round((item["start_hour"] - pd.Timestamp(previous["sample_time"].iloc[0])).total_seconds() / 3600, 2)
                if not previous.empty else None
            )
            item["transition_ratio"] = round(float(part["transition"].mean()), 2) if "transition" in part else 0.0
        else:
            item["gap_before_hours"] = None
            item["transition_ratio"] = 0.0
    result = pd.DataFrame(merged)
    return result.sort_values(["start_hour", "cylinder"]).reset_index(drop=True)


def clean_observations(labels: pd.DataFrame) -> tuple[pd.DataFrame, pd.DataFrame]:
    """只清理弱观察标签；确认标签全部保留。"""
    if labels.empty:
        return labels, labels
    kept, removed = [], []
    for row in labels.itertuples(index=False):
        reasons = []
        if not bool(row.confirmed):
            if row.gap_before_hours is not None and row.gap_before_hours >= 2:
                reasons.append("前面断档")
            if row.direction == "未确认":
                reasons.append("未确认")
            if row.direction == "高幅":
                reasons.append("仅高幅")
            if float(row.transition_ratio or 0) >= 0.5:
                reasons.append("启动/过渡")
        if reasons:
            item = row._asdict()
            item["remove_reason"] = " / ".join(dict.fromkeys(reasons))
            removed.append(item)
        else:
            kept.append(row._asdict())
    kept_frame = pd.DataFrame(kept).sort_values(["start_hour", "cylinder"]).reset_index(drop=True) if kept else pd.DataFrame(columns=labels.columns)
    removed_frame = pd.DataFrame(removed).sort_values(["start_hour", "cylinder"]).reset_index(drop=True) if removed else pd.DataFrame(columns=list(labels.columns) + ["remove_reason"])
    return kept_frame, removed_frame