Compressor/backend/indicator_features/bin_plot.py

#!/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()