"""整机多缸振动事件 HTML：振动 1×6 子图 + 异常缸温度子图。""" from __future__ import annotations import json import re from datetime import timedelta from pathlib import Path import numpy as np import pandas as pd from . import config, data try: from backend.app.services import temperature_transition_service as temp_service except Exception: temp_service = None SLOT_MINUTES = 15 WAVEFORM_POINTS = 16 CONTEXT_HOURS = 12 TEMP_SEARCH_HOURS = 24 TEMP_MAX_LABELS = 2 COLORS = ["#2980b9", "#e67e22", "#16a085", "#8e44ad", "#7f8c8d", "#d35400"] def _finite(value): return None if pd.isna(value) else round(float(value), 4) def _slot_of(moment: pd.Timestamp, start: pd.Timestamp) -> int: return int((pd.Timestamp(moment) - pd.Timestamp(start)).total_seconds() // (SLOT_MINUTES * 60)) def _baseline_rms(meta: dict) -> float: return float(10 ** float(meta.get("rms_log", [0.0, 1.0])[0])) def _temp_cylinder(label: str) -> str | None: match = re.match(r"^(一级|二级)(\d)#缸$", str(label or "").strip()) if not match: return None return config.STAGE_CYLINDER_TO_GLOBAL.get((match.group(1), f"{match.group(2)}#")) def _load_temperature_labels() -> pd.DataFrame: path = config.BASE_DIR.parent / "_tmp_temperature" / "out" / "温度异常段.csv" if not path.exists(): return pd.DataFrame() frame = pd.read_csv(path, encoding="utf-8-sig") frame["_csv_index"] = frame.index frame = frame[frame["机组"].astype(str).str.strip() == "5号机"].copy() frame["开始小时"] = pd.to_datetime(frame["开始小时"], errors="coerce") frame["结束小时"] = pd.to_datetime(frame["结束小时"], errors="coerce") frame["cylinder"] = frame["阶段部位"].map(_temp_cylinder) return frame.dropna(subset=["开始小时", "结束小时", "cylinder"]).reset_index(drop=True) def _select_temperatures(temp: pd.DataFrame, event_start: pd.Timestamp, event_end: pd.Timestamp) -> pd.DataFrame: if temp.empty: return temp window_start = event_start - pd.Timedelta(hours=TEMP_SEARCH_HOURS) window_end = event_end + pd.Timedelta(hours=TEMP_SEARCH_HOURS) part = temp[(temp["结束小时"] >= window_start) & (temp["开始小时"] <= window_end)].copy() if part.empty: return part part["distance_hours"] = part.apply( lambda row: min( abs((row["开始小时"] - event_end).total_seconds() / 3600.0), abs((event_start - row["结束小时"]).total_seconds() / 3600.0), 0.0, ), axis=1, ) return part.sort_values("distance_hours").head(TEMP_MAX_LABELS) def _temperature_subplot(row: pd.Series, context_start: pd.Timestamp, context_end: pd.Timestamp) -> dict | None: if temp_service is None: return None stage_name = re.match(r"^(一级|二级)", str(row["阶段部位"] or "")) if not stage_name: return None stage = stage_name.group(1) columns = temp_service.CYLINDERS.get(5, {}).get(stage) if not columns: return None batch = int(float(row.get("批次") or 0)) rows = temp_service._query(5, batch, context_start, context_end) if not rows: return None target = str(row["阶段部位"]).replace("缸", "") peer_colors = ["#1f6feb", "#2e9e5b", "#8e44ad"] target_color = "#c0392b" stride = max(1, len(rows) // 20000) series_out = [] all_values = [] peer_index = 0 for column, name in columns: is_target = name.startswith(target) color = target_color if is_target else peer_colors[peer_index % len(peer_colors)] if not is_target: peer_index += 1 points = [] for row_index in range(0, len(rows), stride): item = rows[row_index] value = temp_service._temp_value(item.get(column)) if value is None: continue moment = pd.Timestamp(item["t"]) local_slot = (moment - context_start).total_seconds() / (SLOT_MINUTES * 60) points.append([round(local_slot, 4), round(float(value), 2)]) all_values.append(float(value)) if not points: continue series_out.append({ "name": name, "color": color, "is_target": is_target, "data": points, }) if not series_out: return None limits = temp_service._limits(5, [column for column, _ in columns]) highs = [limits[column]["high"] for column, _ in columns if limits[column]["high"] is not None] high_highs = [limits[column]["high_high"] for column, _ in columns if limits[column]["high_high"] is not None] refs = [] if highs: refs.append({"value": min(highs), "label": "高报"}) if high_highs: refs.append({"value": min(high_highs), "label": "高高报"}) y_min, y_max = temp_service._padded(all_values) band_start = max(0.0, (row["开始小时"] - context_start).total_seconds() / (SLOT_MINUTES * 60)) band_end = min( (context_end - context_start).total_seconds() / (SLOT_MINUTES * 60), (row["结束小时"] - context_start).total_seconds() / (SLOT_MINUTES * 60), ) band = [round(band_start, 3), round(band_end, 3)] if band_end > band_start else None return { "cylinder": row["cylinder"], "stage_part": str(row["阶段部位"]), "level": str(row["温度阶段"]), "direction": str(row["方向"] or ""), "start": str(row["开始小时"]), "end": str(row["结束小时"]), "title": f"{stage}缸排气温度（异常缸：{target}）", "unit": "℃", "ref_lines": refs, "y_min": round(y_min, 4), "y_max": round(y_max, 4), "series": series_out, "band": band, } def export_machine_events( frame: pd.DataFrame, events: pd.DataFrame, candidates: pd.DataFrame, validation: pd.DataFrame, ) -> Path: frame = frame.copy() frame["sample_time"] = pd.to_datetime(frame["sample_time"]) frame = frame.dropna(subset=["cylinder"]).sort_values(["cylinder", "sample_time"]) if frame.empty or events.empty: raise RuntimeError("缺少特征或整机事件，无法生成 HTML") candidates = candidates.copy() candidates["开始小时"] = pd.to_datetime(candidates["开始小时"]) candidates["结束小时"] = pd.to_datetime(candidates["结束小时"]) candidates["cylinder"] = candidates["阶段部位"].astype(str).str.replace("5号机", "", regex=False) temperatures = _load_temperature_labels() baseline = json.loads((config.OUTPUT_DIR / "baseline_full.json").read_text(encoding="utf-8")) validation_map = {} for row in validation.itertuples(index=False): if row.hit: validation_map[row.machine_label_id] = { "event_id": row.event_id, "lead_days": row.lead_days, } segments = [] needed_ids: set[int] = set() for item in events.itertuples(index=False): event_start = pd.Timestamp(item.开始小时) event_end = pd.Timestamp(item.结束小时) selected_temps = _select_temperatures(temperatures, event_start, event_end) context_start = event_start - timedelta(hours=CONTEXT_HOURS) context_end = event_end + timedelta(hours=CONTEXT_HOURS) for _, temp_row in selected_temps.iterrows(): context_start = min(context_start, pd.Timestamp(temp_row["开始小时"]) - timedelta(hours=2)) context_end = max(context_end, pd.Timestamp(temp_row["结束小时"]) + timedelta(hours=2)) slot_count = max(1, int((context_end - context_start).total_seconds() // (SLOT_MINUTES * 60)) + 1) part = frame[(frame["sample_time"] >= context_start) & (frame["sample_time"] <= context_end)] part = part.drop_duplicates(["cylinder", "sample_time"], keep="first") blocks: dict[str, list[list]] = {} for name in config.CYLINDERS: pieces = [] for row in part[part["cylinder"] == name].itertuples(index=False): local_slot = _slot_of(pd.Timestamp(row.sample_time), context_start) file_id = int(row.id) pieces.append([ local_slot, file_id, _finite(row.score), _finite(row.kurt), _finite(row.kurt_ratio), _finite(row.rms_ratio), ]) needed_ids.add(file_id) blocks[name] = pieces context_labels = {} for name in config.CYLINDERS: areas = [] for row in candidates[candidates["cylinder"] == name].itertuples(index=False): start = pd.Timestamp(row.开始小时) end = pd.Timestamp(row.结束小时) if end <= event_start or start >= event_end: if start < context_end and end > context_start: areas.append([ max(0, _slot_of(start, context_start)), min(slot_count, _slot_of(end, context_start)), row.阶段, ]) context_labels[name] = areas temp_subplots = [] temp_labels = [] for _, temp_row in selected_temps.iterrows(): temp_labels.append({ "cylinder": temp_row["cylinder"], "stage_part": temp_row["阶段部位"], "level": temp_row["温度阶段"], "direction": temp_row["方向"] or "", "start": str(temp_row["开始小时"]), "end": str(temp_row["结束小时"]), }) subplot = _temperature_subplot(temp_row, context_start, context_end) if subplot: temp_subplots.append(subplot) event_start_slot = max(0, _slot_of(event_start, context_start)) event_end_slot = min(slot_count, _slot_of(event_end, context_start)) validation_info = validation_map.get(item.标签ID) segments.append({ "id": item.标签ID, "level": item.阶段, "start": str(event_start), "end": str(event_end), "start_slot": event_start_slot, "end_slot": event_end_slot, "context_start": str(context_start), "slot_count": slot_count, "participating": item.参与缸, "cylinders": int(item.参与缸数), "abnormal_cylinders": int(item.异常缸数), "severe_cylinders": int(item.严重异常缸数), "bad_labels": int(item.异常标签数), "severe_labels": int(item.严重异常标签数), "candidate_labels": int(item.候选标签数), "max_score": item.最大异常分数, "rpm": item.转速范围, "run_start": str(item.运行段开始), "run_end": str(item.运行段结束), "reason": item.阶段触发依据, "validation": validation_info, "temperature_labels": temp_labels, "temperature_subplots": temp_subplots, "blocks": blocks, "context_labels": context_labels, }) samples = data.fetch_samples(sorted(needed_ids), chunk_size=200) waveforms: dict[int, list[list[float]]] = {} for file_id, signal in samples.items(): if signal is None or signal.size == 0: continue row = frame[frame["id"] == file_id] if row.empty: continue cylinder = row.iloc[0]["cylinder"] scale = _baseline_rms(baseline.get(cylinder, {})) if not np.isfinite(scale) or scale <= 0: scale = float(np.sqrt(np.mean(signal ** 2))) or 1.0 chosen = np.linspace(0, signal.size - 1, WAVEFORM_POINTS).astype(int) waveforms[file_id] = [ [round(i / (WAVEFORM_POINTS - 1), 4), round(float(signal[idx]) / scale, 4)] for i, idx in enumerate(chosen) ] payload = {"segments": segments, "waveforms": waveforms} out_dir = config.OUTPUT_DIR / "html" out_dir.mkdir(parents=True, exist_ok=True) html = """ 5号机整机振动事件

5号机整机振动事件

振动波形（左轴） 异常分数（右轴） 实际峭度（右轴，虚线） 温度曲线（异常缸同级3缸，左轴） 参与整机事件缸 事件时间但未参与缸 上下文候选 温度标签 15 分钟槽位交替
""" html = html.replace("__PAYLOAD__", json.dumps(payload, ensure_ascii=False)) target = out_dir / "machine_events.html" target.write_text(html, encoding="utf-8") index = out_dir / "index.html" index.write_text( """5号机缸体振动

5号机缸体振动

打开整机多缸事件标签

打开单缸候选波形（历史）

""", encoding="utf-8", ) print(f"[machine-html] {target}") return target