"""把 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