"""全时间段标签的 HTML：选中一个标签，看该时刻的首周期多缸归一化波形。"""
from __future__ import annotations
import json
from pathlib import Path
import numpy as np
import pandas as pd
from . import config, data
COLORS = {
"一缸": "#c0392b",
"二缸": "#e67e22",
"三缸": "#16a085",
"四缸": "#2980b9",
"五缸": "#8e44ad",
"六缸": "#7f8c8d",
}
MAX_POINTS = 520
def _nearest_row(frame: pd.DataFrame, cylinder: str, target: pd.Timestamp, tolerance=pd.Timedelta(hours=2)):
group = frame[frame["cylinder"] == cylinder]
if group.empty:
return None
distance = (group["sample_time"] - target).abs()
index = distance.idxmin()
if distance.loc[index] > tolerance:
return None
return group.loc[index]
def _baseline_rms(meta: dict) -> float:
median = float(meta.get("rms_log", [0.0, 1.0])[0])
return float(10 ** median)
def export_waveforms(frame: pd.DataFrame, intervals: pd.DataFrame, baseline: dict) -> None:
if frame.empty or intervals.empty:
return
out_dir = config.OUTPUT_DIR / "html"
out_dir.mkdir(parents=True, exist_ok=True)
records: list[dict] = []
id_to_row: dict[int, pd.Series] = {}
for row in intervals.itertuples(index=False):
seed_time = pd.Timestamp(row.max_time)
rows: dict[str, pd.Series] = {}
for name in config.CYLINDERS:
item = _nearest_row(frame, name, seed_time)
if item is not None:
rows[name] = item
id_to_row[int(item["id"])] = item
if not rows:
continue
records.append(
{
"id": len(records),
"cylinder": row.cylinder,
"start": str(row.start_time),
"end": str(row.end_time),
"duration": float(row.duration_hours),
"max_score": float(row.max_score),
"direction": str(row.direction),
"level": str(row.level),
"confirmed": bool(row.multi_feature_confirmed),
"seed_time": str(seed_time),
"rows": {name: int(item["id"]) for name, item in rows.items()},
}
)
samples = data.fetch_samples(list(id_to_row), chunk_size=200)
for record in records:
series = []
for name, file_id in record.pop("rows").items():
signal = samples.get(int(file_id))
if signal is None or signal.size == 0:
continue
item = id_to_row[int(file_id)]
scale = _baseline_rms(baseline.get(name, {}))
if not np.isfinite(scale) or scale <= 0:
scale = float(np.sqrt(np.mean(signal ** 2))) or 1.0
fs = float(item["sample_frequency_hz"] or 0) or 25600.0
step = max(1, int(np.ceil(signal.size / MAX_POINTS)))
chosen = np.arange(0, signal.size, step)
x = chosen / fs * 1000.0
y = signal[chosen] / scale
series.append(
{
"name": f"cyl{config.CYLINDER_TO_INDEX[name]}",
"cylinder": name,
"color": "#c0392b" if name == record["cylinder"] else COLORS[name],
"width": 2.2 if name == record["cylinder"] else 1.0,
"data": [[round(float(x[i]), 3), round(float(y[i]), 4)] for i in range(len(chosen))],
}
)
record["series"] = series
html = """
5号机全时段缸体振动标签
5号机全时段缸体振动标签 · 首周期多缸归一化波形
下拉选择标签；横轴是首周期时间（ms），纵轴是“波形 / 该缸正常基线 RMS”。事件缸加粗红色。缺数据的缸不画。
"""
html = html.replace("__RECORDS__", json.dumps(records, ensure_ascii=False))
(out_dir / "waveforms.html").write_text(html, encoding="utf-8")
(out_dir / "index.html").write_text(
"""5号机缸体振动
5号机缸体振动
打开全时段标签波形
""",
encoding="utf-8",
)
print(f"[html] {out_dir / 'waveforms.html'} ({len(records)} labels)")