"""把振动小时标签映射为与压力/温度可共用的统一字段。""" 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)