#!/usr/bin/env python3 """Build an interactive ECharts dashboard for the high-density one-cylinder events.""" from __future__ import annotations import argparse import json import sys from collections import defaultdict from datetime import datetime, timedelta from pathlib import Path import numpy as np BACKEND = Path(__file__).resolve().parents[1] sys.path.insert(0, str(BACKEND)) from extract_from_db import fetch_cycles, fetch_file_rows, point_name, sample_per_day # noqa: E402 from indicator_features.indicator_features import smooth, _plateau_levels, volume_from_angle # noqa: E402 from app.config import settings # noqa: E402 from extract_from_db import connect # noqa: E402 from new_unit_events import collect_events, top_cylinder_event # noqa: E402 EVENTS = { "7号机": {"unit": "7", "date": "2026-05-21", "label": "7号机 2026-05-21 气阀事件", "pre_days": 30, "post_days": 30}, "8号机（2026）": {"unit": "8", "date": "2026-03-30", "label": "8号机 2026-03-29/04-01 合并事件", "pre_days": 30, "post_days": 30}, "8号机（2025同期）": {"unit": "8", "date": "2025-03-30", "label": "8号机 2025年3月底/4月初同期工况", "pre_days": 30, "post_days": 60}, } SIDES = ("盖侧", "轴侧") COLORS = {"pre": "#4f7cac", "post": "#d95d39", "early": "#91a7c2", "late": "#e69f72"} def collect(connection, unit: str, side: str, event_date: str, per_day: int, bins: int, pre_days: int, post_days: int): t0 = datetime.fromisoformat(event_date) start = (t0 - timedelta(days=pre_days)).date().isoformat() end = (t0 + timedelta(days=post_days + 1)).date().isoformat() files = fetch_file_rows(connection, point_name(unit[0], "一", side), start, end) picked = sample_per_day(files, per_day) cycles = fetch_cycles(connection, [int(r["id"]) for r in picked], 200) result = [] for row in picked: signal = cycles.get(int(row["id"])) if signal is None: continue if len(signal) < 60 or not np.all(np.isfinite(signal)): continue pressure = smooth(signal, 21) suction, discharge, _ = _plateau_levels(pressure) if discharge - suction < 0.3: continue profile = np.interp(np.arange(bins) * len(pressure) / bins, np.arange(len(pressure)), pressure) profile = (profile - suction) / (discharge - suction) dt = row["sample_time"] if not isinstance(dt, datetime): dt = datetime.fromisoformat(str(dt)) result.append({ "date": dt.strftime("%Y-%m-%d %H:%M:%S"), "day": (dt.date() - t0.date()).days, "phase": "修前" if dt < t0 else "修后", "profile": [round(float(x), 6) for x in profile], "absolute": [round(float(x), 6) for x in np.interp(np.arange(bins) * len(pressure) / bins, np.arange(len(pressure)), pressure)], "volume": [round(float(x), 6) for x in np.interp( np.arange(bins) * len(pressure) / bins, np.arange(len(pressure)), volume_from_angle(len(pressure), 360.0), )], "p_suction": round(suction, 6), "p_discharge": round(discharge, 6), "p_span": round(discharge - suction, 6), }) return result def build_data(per_day: int, bins: int): data = {"angle": [round(i * 360 / bins, 2) for i in range(bins)], "machines": {}} connection = connect() try: for machine, config in EVENTS.items(): unit = config["unit"] data["machines"][machine] = {"label": config["label"], "sides": {}} for side in SIDES: samples = collect(connection, unit, side, config["date"], per_day, bins, config["pre_days"], config["post_days"]) pre = [r for r in samples if r["phase"] == "修前"] post = [r for r in samples if r["phase"] == "修后"] pre_template = np.median([r["profile"] for r in pre], axis=0) if pre else None if pre_template is None: pre_template = np.zeros(bins) for r in samples: r["distance"] = round(float(np.linalg.norm(np.asarray(r["profile"]) - pre_template)), 5) def median_profile(items): return [round(float(x), 6) for x in np.median([r["profile"] for r in items], axis=0)] if items else [] def median_absolute(items): return [round(float(x), 6) for x in np.median([r["absolute"] for r in items], axis=0)] if items else [] trend = [{"date": r["date"], "day": r["day"], "distance": r["distance"]} for r in samples] data["machines"][machine]["sides"][f"一缸{side}"] = { "samples": samples, "preMedian": median_profile(pre), "postMedian": median_profile(post), "preAbsoluteMedian": median_absolute(pre), "postAbsoluteMedian": median_absolute(post), "trend": trend, "counts": {"pre": len(pre), "post": len(post)}, "markerDays": [0], } for key, event in collect_events(connection, per_day=per_day, bins=bins).items(): data["machines"][key] = top_cylinder_event(event) finally: connection.close() return data def html(data: dict) -> str: payload = json.dumps(data, ensure_ascii=False, separators=(",", ":")) return f''' 一缸功图事件前后对比

一缸 P-V 功图事件前后对比

每台机组每个腔体显示：四阶段 P-V 功图、压力-角度曲线、逐周期形状距离。图例在图外，可逐时间点勾选；页面按宽度自适应，每行最多3张图。
''' def main(): parser = argparse.ArgumentParser(description="生成一缸功图ECharts网页") parser.add_argument("--per-day", type=int, default=6) parser.add_argument("--bins", type=int, default=90) parser.add_argument("--out", required=True) args = parser.parse_args() data = build_data(args.per_day, args.bins) out = Path(args.out) out.parent.mkdir(parents=True, exist_ok=True) out.write_text(html(data), encoding="utf-8") print(f"HTML -> {out}") for machine, item in data["machines"].items(): for side, value in item["sides"].items(): print(machine, side, value["counts"]) if __name__ == "__main__": main()