"""整机多缸振动事件标签。 最终标签不再按单缸输出，而是按稳定运行段内的整机多缸聚集事件输出： - 排除开机后 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, }, }