#!/usr/bin/env python3 """Plot the per-angular-bin statistics before/after a chosen event (read-only). Uses the same running-segment selection as ``event_bins.py`` (rpm>0 segments, at least ``--pre-days`` / ``--post-days`` running days). Two figures per event: * profile: for every measurement point, the pre-event median and post-event median of a chosen metric across the bins (whole profile shift vs shape); * series : for every point, the time trace of the bin whose local (shape) change is the largest for a chosen metric, pre and post coloured differently. """ from __future__ import annotations import argparse import sys from collections import defaultdict from datetime import datetime from pathlib import Path import numpy as np import matplotlib matplotlib.use("Agg") import matplotlib.pyplot as plt # noqa: E402 matplotlib.rcParams["font.sans-serif"] = ["Arial Unicode MS", "Heiti TC", "Hiragino Sans GB", "DejaVu Sans"] matplotlib.rcParams["axes.unicode_minus"] = False BACKEND = Path(__file__).resolve().parents[1] sys.path.insert(0, str(BACKEND)) from event_bins import load as load_rows, collect_before, collect_after, robust_scale # noqa: E402 from run_segments import unit_segments # noqa: E402 CYLINDERS = ("一", "二", "三", "四", "五", "六") SIDES = ("盖侧", "轴侧") VOLATILE = ("std", "mad", "range", "roughness") def median_at(rows, metric, bin_index): values = [r[metric] for r in rows if r["bin"] == bin_index and r[metric] is not None] return float(np.median(values)) if values else None def main() -> None: parser = argparse.ArgumentParser(description="每角度箱 前后剖面 / 时间序列图（运行段口径）") parser.add_argument("--csv", required=True, help="angle_bins.py 长表 CSV") parser.add_argument("--unit", required=True, choices=("7", "8", "9")) parser.add_argument("--date", required=True, help="事件日期 YYYY-MM-DD") parser.add_argument("--pre-days", type=int, default=15) parser.add_argument("--post-days", type=int, default=15) parser.add_argument("--merge-gap", type=int, default=2) parser.add_argument("--max-gap", type=int, default=30) parser.add_argument("--profile-metric", default="mean") parser.add_argument("--series-metric", default="std") parser.add_argument("--outdir", default="cache/indicator/bins/plots") args = parser.parse_args() rows = load_rows(args.csv) t0 = datetime.fromisoformat(args.date) segments = unit_segments(args.unit, args.merge_gap) bins = sorted({r["bin"] for r in rows}) bin_start = {r["bin"]: r["bin_start"] for r in rows} bin_deg = round(bin_start[bins[1]] - bin_start[bins[0]], 6) if len(bins) > 1 else 10.0 points = [(c, s) for c in CYLINDERS for s in SIDES if any(r["cylinder"] == c and r["side"] == s for r in rows)] outdir = Path(args.outdir) outdir.mkdir(parents=True, exist_ok=True) # pre/post rows per point via running segments selected = {} for cylinder, side in points: point_rows = [r for r in rows if r["cylinder"] == cylinder and r["side"] == side] pre, _ = collect_before(point_rows, t0, segments, args.pre_days) post, _ = collect_after(point_rows, t0, segments, args.post_days, args.max_gap) selected[(cylinder, side)] = (pre, post) cols = 3 rows_n = (len(points) + cols - 1) // cols # ---- profile figure ---- fig, axes = plt.subplots(rows_n, cols, figsize=(4.2 * cols, 2.6 * rows_n), squeeze=False) for ax, key in zip(axes.ravel(), points): pre, post = selected[key] xs = [bin_start[b] + bin_deg / 2 for b in bins] prof_pre = [median_at(pre, args.profile_metric, b) for b in bins] prof_post = [median_at(post, args.profile_metric, b) for b in bins] ax.plot(xs, prof_pre, color="crimson", lw=1.4, label="pre") ax.plot(xs, prof_post, color="royalblue", lw=1.4, label="post") ax.set_title(f"{key[0]}缸{key[1]}", fontsize=10) ax.set_xlim(0, 360) ax.set_xticks([0, 90, 180, 270, 360]) ax.tick_params(labelsize=8) ax.grid(alpha=0.25) for ax in axes.ravel()[len(points):]: ax.axis("off") axes.ravel()[0].legend(fontsize=7) fig.suptitle(f"{args.unit}号机 {args.date} 每角度箱 {args.profile_metric}（pre红 post蓝）") fig.tight_layout() p1 = outdir / f"{args.unit}_{args.date}_profile_{args.profile_metric}.png" fig.savefig(p1, dpi=110) plt.close(fig) print("图:", p1) # ---- series figure: strongest local shape change per point ---- fig, axes = plt.subplots(rows_n, cols, figsize=(4.2 * cols, 2.6 * rows_n), squeeze=False) for ax, key in zip(axes.ravel(), points): pre, post = selected[key] best_bin, best_score = None, 0.0 for b in bins: a = [r[args.series_metric] for r in pre if r["bin"] == b and r[args.series_metric] is not None] c = [r[args.series_metric] for r in post if r["bin"] == b and r[args.series_metric] is not None] if len(a) < 3 or len(c) < 3: continue scale = robust_scale(np.asarray(a)) if scale <= 1e-12: continue score = abs(float(np.median(c)) - float(np.median(a))) / scale if score > best_score: best_bin, best_score = b, score if best_bin is None: ax.set_title(f"{key[0]}缸{key[1]} 无数据", fontsize=9) ax.axis("off") continue trace = sorted([r for r in rows if r["cylinder"] == key[0] and r["side"] == key[1] and r["bin"] == best_bin and r[args.series_metric] is not None], key=lambda r: r["time"]) for r in trace: color = "royalblue" if r["time"] >= t0 else "crimson" ax.plot(r["time"], r[args.series_metric], ".", color=color, markersize=3) ax.axvline(t0, color="k", lw=0.8, ls="--") start = bin_start[best_bin] ax.set_title(f"{key[0]}缸{key[1]} 箱{start:g}-{start + bin_deg:g}° ({best_score:.1f}MAD)", fontsize=9) ax.tick_params(labelsize=7, rotation=30) ax.grid(alpha=0.25) for ax in axes.ravel()[len(points):]: ax.axis("off") fig.suptitle(f"{args.unit}号机 {args.date} 各点局部变化最大的箱 {args.series_metric} 时间序列") fig.tight_layout() p2 = outdir / f"{args.unit}_{args.date}_series_{args.series_metric}.png" fig.savefig(p2, dpi=110) plt.close(fig) print("图:", p2) if __name__ == "__main__": main()