#!/usr/bin/env python3 from __future__ import annotations import argparse import csv import glob import math import os from dataclasses import dataclass import numpy as np def _should_use_agg(save_path: str | None) -> bool: if save_path: return True # Headless environment return not (os.getenv("DISPLAY") or os.getenv("WAYLAND_DISPLAY")) def _parse_float(x: object) -> float: try: if x is None: return float("nan") s = str(x).strip() if not s: return float("nan") return float(s) except Exception: return float("nan") @dataclass class ScanSeries: label: str mm: np.ndarray f0_hz: np.ndarray dphi_rad: np.ndarray dphi_deg: np.ndarray amp1_pp_adc: np.ndarray amp2_pp_adc: np.ndarray def load_scan_csv(path: str, *, label: str | None = None) -> ScanSeries: mm: list[float] = [] f0_hz: list[float] = [] dphi_rad: list[float] = [] dphi_deg: list[float] = [] amp1: list[float] = [] amp2: list[float] = [] with open(path, "r", encoding="utf-8", newline="") as f: r = csv.DictReader(f) for row in r: mm.append(_parse_float(row.get("mm"))) f0_hz.append(_parse_float(row.get("f0_hz"))) dphi_rad.append(_parse_float(row.get("dphi_rad"))) dphi_deg.append(_parse_float(row.get("dphi_deg"))) amp1.append(_parse_float(row.get("amp1_pp_adc"))) amp2.append(_parse_float(row.get("amp2_pp_adc"))) arr_mm = np.asarray(mm, dtype=float) order = np.argsort(arr_mm) series_label = label or os.path.basename(path) return ScanSeries( label=series_label, mm=arr_mm[order], f0_hz=np.asarray(f0_hz, dtype=float)[order], dphi_rad=np.asarray(dphi_rad, dtype=float)[order], dphi_deg=np.asarray(dphi_deg, dtype=float)[order], amp1_pp_adc=np.asarray(amp1, dtype=float)[order], amp2_pp_adc=np.asarray(amp2, dtype=float)[order], ) def nan_moving_average(y: np.ndarray, window: int) -> np.ndarray: if window <= 1: return y y = np.asarray(y, dtype=float) n = y.size out = np.full(n, np.nan, dtype=float) half = window // 2 for i in range(n): lo = max(0, i - half) hi = min(n, i + half + 1) out[i] = np.nanmean(y[lo:hi]) return out def _mask_range(x: np.ndarray, *, x_min: float | None, x_max: float | None) -> np.ndarray: m = np.isfinite(x) if x_min is not None: m &= x >= x_min if x_max is not None: m &= x <= x_max return m def _get_pingfang_font_properties(ttf_path: str): """Return (FontProperties, name) for PingFang SC from a TTF file.""" from matplotlib import font_manager if not os.path.exists(ttf_path): raise FileNotFoundError( f"未找到字体文件：{ttf_path}（请确认 PingFangSC-Regular.ttf 已放在项目目录，或已安装到 ~/.local/share/fonts）" ) # Ensure Matplotlib can see this font even if fontconfig cache is stale. try: font_manager.fontManager.addfont(ttf_path) except Exception: pass fp = font_manager.FontProperties(fname=ttf_path) return fp, fp.get_name() def main() -> None: ap = argparse.ArgumentParser(description="可视化 motor_rigol_scan_csv.py 输出的扫描 CSV") ap.add_argument( "csv", nargs="+", help="CSV 路径（支持通配符），例如 scan_*.csv", ) ap.add_argument("--unwrap", action="store_true", help="对相位差 dphi 做 unwrap（基于 dphi_rad）") ap.add_argument("--smooth", type=int, default=1, help="平滑窗口(点数)，默认1=不平滑") ap.add_argument("--mm-min", type=float, default=None, help="仅显示 mm >= mm-min") ap.add_argument("--mm-max", type=float, default=None, help="仅显示 mm <= mm-max") ap.add_argument("--save", default="", help="保存为图片(PNG/PDF等)，例如 out.png；为空则弹窗显示") ap.add_argument("--title", default="Motor Scan", help="图标题") args = ap.parse_args() paths: list[str] = [] for p in args.csv: expanded = sorted(glob.glob(p)) if expanded: paths.extend(expanded) else: paths.append(p) if not paths: raise SystemExit("未找到任何 CSV") save_path = args.save.strip() or None if _should_use_agg(save_path): import matplotlib matplotlib.use("Agg") import matplotlib pingfang_ttf = os.path.join(os.path.dirname(__file__), "PingFangSC-Regular.ttf") cjk_fp, cjk_name = _get_pingfang_font_properties(pingfang_ttf) matplotlib.rcParams["font.family"] = cjk_name matplotlib.rcParams["axes.unicode_minus"] = False print(f"[+] Font: {cjk_name} ({pingfang_ttf})") import matplotlib.pyplot as plt # type: ignore series_list: list[ScanSeries] = [] for path in paths: if not os.path.exists(path): raise SystemExit(f"CSV 不存在：{path}") series_list.append(load_scan_csv(path)) fig = plt.figure(figsize=(12, 5.5)) gs = fig.add_gridspec(1, 2) ax_phase = fig.add_subplot(gs[0, 0]) ax_amp = fig.add_subplot(gs[0, 1]) ax_phase.set_title("相位差", fontproperties=cjk_fp) ax_phase.set_xlabel("位置 (mm)", fontproperties=cjk_fp) ax_phase.set_ylabel("相位差 (度)", fontproperties=cjk_fp) ax_phase.grid(True, alpha=0.3) ax_amp.set_title("峰峰值", fontproperties=cjk_fp) ax_amp.set_xlabel("位置 (mm)", fontproperties=cjk_fp) ax_amp.set_ylabel("峰峰值 (ADC)", fontproperties=cjk_fp) ax_amp.grid(True, alpha=0.3) for s in series_list: mm_mask = _mask_range(s.mm, x_min=args.mm_min, x_max=args.mm_max) mm = s.mm[mm_mask] amp1 = s.amp1_pp_adc[mm_mask] amp2 = s.amp2_pp_adc[mm_mask] # phase (deg) if args.unwrap: dphi_r = s.dphi_rad[mm_mask] dphi_r_unwrapped = np.unwrap(dphi_r) dphi_deg = dphi_r_unwrapped * 180.0 / math.pi else: dphi_deg = s.dphi_deg[mm_mask] # scatter raw ax_phase.plot(mm, dphi_deg, ".", ms=3, alpha=0.6, label=s.label) ax_amp.plot(mm, amp1, ".", ms=3, alpha=0.35, label=f"{s.label} CH1") ax_amp.plot(mm, amp2, ".", ms=3, alpha=0.35, label=f"{s.label} CH2") # smoothed overlay if args.smooth and args.smooth > 1: ax_phase.plot(mm, nan_moving_average(dphi_deg, args.smooth), "-", lw=1.2, alpha=0.9) ax_amp.plot(mm, nan_moving_average(amp1, args.smooth), "-", lw=1.2, alpha=0.9) ax_amp.plot(mm, nan_moving_average(amp2, args.smooth), "-", lw=1.2, alpha=0.9) # legends (avoid huge) ax_phase.legend(loc="best", fontsize=8) ax_amp.legend(loc="best", fontsize=7, ncols=2) subtitle = " | ".join([s.label for s in series_list]) # 中文标题使用 CJK 字体；subtitle 一般是文件名/ASCII，用默认字体。 fig.suptitle(f"{args.title}", fontproperties=cjk_fp) fig.text(0.5, 0.965, subtitle, ha="center", va="top", fontsize=9) fig.tight_layout(rect=(0, 0, 1, 0.95)) if save_path: fig.savefig(save_path, dpi=160) print(f"[+] Saved: {save_path}") else: plt.show() if __name__ == "__main__": main()