Merge branch 'main' of https://git.dominatedconvergence.com/louis/flow_control

# Conflicts:
#	__pycache__/config.cpython-313.pyc
#	__pycache__/controllers.cpython-313.pyc

.gitignore

@@ -0,0 +1,6 @@
+__pycache__
+.claude
+.vscode
+解释压差传感器工作原理.md
+气体流量控制方案设计.md
+阻塞流压力判断.md

README.md

@@ -0,0 +1,119 @@
+# 气体流量闭环控制系统
+
+基于 **MT2-AM8 模拟量采集/输出模块** 的单回路气体流量 PID 闭环控制程序。
+程序以 10 Hz 周期读取流量（可选读取压力），经增量式 PID 计算阀门开度，
+映射为电机行程后通过 Modbus TCP 下发给执行机构，并在每次运行结束后自动
+保存逐周期数据、生成三联曲线图与一次性的控制性能报告。
+
+## 硬件与通讯
+
+- 采集/输出模块：艾莫迅 **MT2-AM8**（Modbus TCP，默认 `192.168.1.12:502`，站号 1）
+- 模拟量输入（AI）：地址 `0x00~0x03`，0~4095 对应 0~10V / 0~20mA
+- 模拟量输出（AO）：保持寄存器 `0x00~0x03`
+- 流量计量程默认 300 SLM，压力表量程默认 400 kPa
+- 电机行程与开度**反向作用**：行程越大开度越小，
+  `opening=0% → MOTOR_CLOSED_POSITION`，`opening=100% → MOTOR_OPEN_POSITION`
+
+> 所有地址、通道、量程、行程端值均在 [config.py](config.py) 中集中配置，
+> 首次连接真实设备前必须核对。
+
+## 架构与模块职责
+
+| 文件 | 职责 |
+| --- | --- |
+| [main.py](main.py) | 程序入口：解析参数、装配日志/控制器/记录器/报告器、主循环与安全退出 |
+| [flow_control.py](flow_control.py) | 闭环控制层：`读取传感器 → PID → 开度/行程映射 → 下发电机位置`，含全部安全处理 |
+| [controllers.py](controllers.py) | 通用增量式 PID 控制器 `IncrementalPID`（输出限幅 + 输出变化率限幅） |
+| [config.py](config.py) | 集中配置与 `validate_config()` 启动前校验 |
+| [data_logger.py](data_logger.py) | `FlowRunRecorder`：逐周期写 CSV，运行结束生成三联曲线图 |
+| [performance_reporter.py](performance_reporter.py) | `PerformanceReporter`：滑窗稳态判定，输出一次性性能报告 |
+| [PcControl.py](PcControl.py) | 硬件层：`MT2AM8Client` 等 Modbus 客户端（本项目实际使用 `MT2AM8Client`） |
+
+### 控制回路流程
+
+```
+读取流量/压力 (MT2-AM8)
+      │
+      ▼
+IncrementalPID.update()  ──►  开度百分比 0~100%
+      │
+      ▼
+opening_to_motor_position()  ──►  反向行程 0~1000
+      │
+      ▼
+hardware.set_motor_position()  ──►  Modbus TCP 写 AO
+```
+
+`error = setpoint - measurement`；PID 输出累加后统一映射为阀门开度百分比，
+再做行程换算，因此 PID 与执行机构解耦。
+
+## 快速开始
+
+```bash
+python main.py --target 50
+```
+
+- `--target`：目标流量（SLM），必须在 `[TARGET_FLOW_MIN_SLM, TARGET_FLOW_MAX_SLM]` 内。
+- 启动流程：连接 MT2-AM8 → 确认阀门关闭 → 读取初始流量 → 启动闭环。
+- 停止：`Ctrl+C`；控制故障、未处理异常同样会触发安全关阀并断开设备。
+
+## 运行输出
+
+1. **终端/日志**：周期状态行（状态、目标/实际流量、压力、开度、行程、实际 dt），
+   日志同时写入 [logs/flow_control.log](logs/flow_control.log)。
+2. **逐周期数据**：每次运行生成 `data/flow_run_<时间戳>.csv`。
+3. **曲线图**：与 CSV 同名的 PNG 三联图（流量响应 / 阀门开度 / 压力）。
+4. **性能报告**：首次达到稳态时打印一次，包含：
+
+   - 本次 PID 参数（Kp / Ki / Kd）
+   - 调节时间（至稳态，秒）
+   - 超调量（绝对 SLM + 相对目标百分比）
+   - 本次控制范围
+
+   稳态判定为两个条件**同时**满足：滑窗均值落在目标 ±`STEADY_STATE_BAND_PCT`% 误差带内，
+   且滑窗标准差低于 `STEADY_STATE_STD_PCT`% 目标流量阈值。
+
+## 安全机制
+
+- **启动前安全关阀**：控制开始前先确认阀门已关闭。
+- **流量/压力越限关阀**：读数超出 `FLOW_VALID_*` 或超过 `MAX_PRESSURE_KPA` 立即关阀。
+- **连续读取失败关阀**：短暂读取失败保持上一输出；连续达到阈值后关阀。
+- **控制循环超时保护**：实际周期超过 `MAX_CONTROL_DT_S` 判定超时。
+- **输出变化率限幅**：`MAX_OPENING_RATE_PCT_S` 限制开度变化速度，防止阀门阶跃冲击。
+- **退出安全关阀**：任何退出路径（含异常）都会再次尝试关阀并断开设备。
+
+## 关键配置项
+
+见 [config.py](config.py)，常用项：
+
+| 配置 | 默认值 | 说明 |
+| --- | --- | --- |
+| `MT2AM8_HOST` / `MT2AM8_PORT` / `MT2AM8_SLAVE_ID` | `192.168.1.12` / `502` / `1` | 模块地址 |
+| `FLOW_INPUT_CHANNEL` / `PRESSURE_INPUT_CHANNEL` / `MOTOR_OUTPUT_CHANNEL` | `1` / `0` / `0` | AI/AO 通道（0 起始） |
+| `FLOW_METER_RANGE_SLM` / `PRESSURE_RANGE_KPA` | `300` / `400` | 量程 |
+| `CONTROL_PERIOD_S` | `0.1` | 控制周期（100 ms，10 Hz） |
+| `PID_KP` / `PID_KI` / `PID_KD` | `0.5` / `0.1` / `0.0` | 安全起步 PI 参数，需现场整定 |
+| `MOTOR_CLOSED_POSITION` / `MOTOR_OPEN_POSITION` | `1000` / `0` | 行程端值 |
+| `MAX_PRESSURE_KPA` | `400` | 压力上限，须按管路额定压力确认 |
+| `STEADY_STATE_WINDOW_S` / `STEADY_STATE_BAND_PCT` / `STEADY_STATE_STD_PCT` | `2.0` / `2.0` / `2.0` | 稳态判定参数 |
+
+## 依赖
+
+- Python 3
+- [pymodbus](https://pymodbus.readthedocs.io/)（`PcControl.py` 的 Modbus TCP 通讯）
+- [matplotlib](https://matplotlib.org/)（曲线图生成，使用 `Agg` 后端，无显示器环境也可保存）
+
+## 项目结构
+
+```
+flow_control/
+├── main.py                    # 程序入口
+├── flow_control.py            # 闭环控制层
+├── controllers.py             # 增量式 PID
+├── config.py                  # 集中配置与校验
+├── data_logger.py             # CSV 记录 + 曲线图
+├── performance_reporter.py    # 稳态判定 + 性能报告
+├── PcControl.py               # Modbus 硬件客户端
+├── data/                      # 每次运行的 CSV / PNG
+└── logs/                      # 运行日志
+```

config.py

@@ -62,18 +62,21 @@ STEADY_STATE_STD_PCT = 2.0       # 滑窗标准差阈值（相对目标百分比
 # opening=0%   -> MOTOR_CLOSED_POSITION（完全关闭）
 # opening=100% -> MOTOR_OPEN_POSITION（打开端/机械死区边界）
 MOTOR_CLOSED_POSITION = 1000.0
-MOTOR_OPEN_POSITION = 240.0
+MOTOR_OPEN_POSITION = 0.0
+
+# 控制启动时的阀门初始开度；增量 PID 从该开度开始累加 Δu。
+INITIAL_OPENING_PCT = 100.0
 
 # ---------------------------------------------------------------------------
 # 安全阈值
 # ---------------------------------------------------------------------------
 
-# 物理读数允许范围；超过范围立即关阀。
+# 物理读数允许范围；超过范围立即报故障停机（阀门不被驱动时回到全开）。
 FLOW_VALID_MIN_SLM = 0.0
 FLOW_VALID_MAX_SLM = 300.0
 MAX_PRESSURE_KPA = 400         # 必须按实际管路额定压力确认
 
-# 短暂读取失败时保持上一输出，不下发新位置；连续达到阈值后关阀。
+# 短暂读取失败时保持上一输出，不下发新位置；连续达到阈值后报故障停机。
 MAX_CONSECUTIVE_FLOW_FAILURES = 3
 MAX_CONSECUTIVE_PRESSURE_FAILURES = 3
 

flow_control.py

@@ -166,12 +166,9 @@ class FlowControlLoop:
         self.faulted = False
         self.running = True
 
-    def stop(self, close_valve=True):
-        """停止 PID；默认同时安全关阀。"""
+    def stop(self):
+        """停止 PID；阀门保持当前开度（默认全开）。"""
         self.running = False
-        self.pid.reset(initial_output=0.0)
-        if close_valve:
-            return self.safe_close("STOP_REQUESTED")
         return True
 
     def step(self, now=None):
@@ -290,39 +287,6 @@ class FlowControlLoop:
             actual_dt_s=actual_dt,
         )
 
-    def safe_close(self, reason="UNSPECIFIED"):
-        """尝试把阀门置于关闭端；失败时返回 False 并详细记录。"""
-        try:
-            success = bool(
-                self.hardware.set_motor_position(
-                    self.motor_closed_position,
-                    channel=self.motor_channel,
-                )
-            )
-        except Exception as exc:
-            self._log(
-                "critical",
-                "SAFE_CLOSE_FAILED reason=%s exception=%s context=%s",
-                reason,
-                repr(exc),
-                self._context(),
-            )
-            return False
-
-        if success:
-            self.last_opening_pct = 0.0
-            self.last_motor_position = self.motor_closed_position
-            self._log("warning", "阀门已安全关闭，原因=%s", reason)
-            return True
-
-        self._log(
-            "critical",
-            "SAFE_CLOSE_FAILED reason=%s hardware_return=False context=%s",
-            reason,
-            self._context(),
-        )
-        return False
-
     def _read_flow(self):
         try:
             value = self.hardware.get_flow(self.flow_channel)
@@ -409,8 +373,6 @@ class FlowControlLoop:
         context.update(extra_context or {})
         self.faulted = True
         self.running = False
-        close_ok = self.safe_close(code)
-        context["safe_close_success"] = close_ok
         fault = FlowControlFault(code, stage, message, context)
         self._log("critical", "%s", fault)
         raise fault

logs/open_loop.log

@@ -0,0 +1,5 @@
+2026-08-19 14:22:33.301 CRITICAL 缺少 PcControl.py 所需的 pymodbus；请在运行本工程的 Python 环境中安装与现有硬件代码兼容的 pymodbus 版本
+2026-08-19 14:23:00.260 INFO 开度扫描顺序（已打乱）: 70.0%, 30.0%, 20.0%, 80.0%, 50.0%, 60.0%, 90.0%, 40.0%, 0.0%, 10.0%, 100.0%
+2026-08-19 14:23:00.260 INFO 正在连接 MT2-AM8 192.168.1.12:502
+2026-08-19 14:23:03.274 CRITICAL 无法连接 MT2-AM8，实验中止
+2026-08-19 14:23:03.274 INFO 程序结束，退出码=2

main.py

@@ -3,8 +3,8 @@
 示例：
     python main.py --target 50
 
-程序启动后会先下发关闭位置，再开始闭环。Ctrl+C、控制故障或其他异常都会
-再次尝试关阀并断开 MT2-AM8。
+程序启动后直接从当前阀门开度（默认全开）开始闭环。Ctrl+C、控制故障或
+其他异常会断开 MT2-AM8；阀门不被驱动时默认回到全开。
 """
 
 import argparse
@@ -156,14 +156,6 @@ def run(target_flow_slm):
                 },
             )
 
-        if not controller.safe_close("STARTUP"):
-            raise FlowControlFault(
-                "SAFE_CLOSE_FAILED",
-                "STARTUP",
-                "启动前无法确认阀门关闭命令已成功写入",
-                {"motor_closed_position": config.MOTOR_CLOSED_POSITION},
-            )
-
         # 控制前读取一次当前流量，作为超调量方向的基准。
         try:
             initial_flow_slm = hardware.get_flow(config.FLOW_INPUT_CHANNEL)
@@ -181,7 +173,7 @@ def run(target_flow_slm):
             logger=logger,
         )
 
-        controller.start(initial_opening=0.0)
+        controller.start(initial_opening=config.INITIAL_OPENING_PCT)
         recorder = FlowRunRecorder(
             Path(__file__).resolve().parent / config.DATA_DIRECTORY,
             plot_dpi=config.PLOT_DPI,
@@ -216,13 +208,9 @@ def run(target_flow_slm):
         logger.critical("控制故障：%s", exc)
     except Exception:
         exit_code = 3
-        logger.exception("未处理异常，系统将进入安全关闭")
+        logger.exception("未处理异常，系统将退出")
     finally:
         if connected:
-            close_ok = controller.safe_close("PROGRAM_EXIT")
-            if not close_ok:
-                exit_code = max(exit_code, 4)
-                logger.critical("程序退出时安全关阀失败，请立即人工检查")
             try:
                 hardware.disconnect()
             except Exception:

open_loop.py

@@ -0,0 +1,648 @@
+"""气体流量开环扫点实验（不参与闭环反馈，也不被 main.py 导入）。
+
+本模块可独立运行，用于阀门开度—流量特性的开环标定/扫点。程序会：
+
+1. 生成 0%~100%（默认 10% 间隔）共 11 个开度值并随机打乱；
+2. 逐个开度：写入电机行程 → 以固定周期采集控流阀前后 4 个传感器
+   （前压力、前流量、后压力、后流量）→ 等待“控流阀后流量”进入稳态后
+   切换到下一个开度；
+3. 全程逐周期写 CSV，结束后在单个窗口内输出两张子图（前后压力、前后流量），
+   保存到工程目录下的 open_loop_data/。
+
+示例：
+    python open_loop.py
+    python open_loop.py --seed 42          # 复现同一种打乱顺序
+    python open_loop.py --min 0 --max 100 --step 5
+
+量程、行程端值、稳态窗口等沿用 config.py；本文件顶部的常量可在不修改
+config.py 的前提下覆盖实验相关参数（尤其是四个传感器通道映射）。
+"""
+
+import argparse
+import csv
+import logging
+import math
+import random
+import statistics
+import time
+from collections import deque
+from pathlib import Path
+
+import config
+from flow_control import opening_to_motor_position
+
+
+# ---------------------------------------------------------------------------
+# 实验参数（可在本文件修改，不影响 config.py）
+# ---------------------------------------------------------------------------
+
+# 四个传感器的 AI 通道（0 起始，MT2-AM8 AI0~AI3）。
+# “后”= 控流阀出口侧（靠近流量计/负载一侧），沿用 config 的原始通道；
+# “前”= 控流阀入口侧（靠近减压阀一侧），为本次新增传感器，请按实际接线确认。
+PRESSURE_AFTER_CHANNEL = config.PRESSURE_INPUT_CHANNEL   # 后压力，默认 0
+FLOW_AFTER_CHANNEL = config.FLOW_INPUT_CHANNEL           # 后流量，默认 1
+PRESSURE_BEFORE_CHANNEL = 2                              # 前压力（新增，需确认）
+FLOW_BEFORE_CHANNEL = 3                                  # 前流量（新增，需确认）
+
+MOTOR_CHANNEL = config.MOTOR_OUTPUT_CHANNEL              # AO 通道
+
+# 开度扫描：0%~100%，默认 10% 间隔，共 11 个开度值，随机打乱。
+OPENING_MIN_PCT = 0.0
+OPENING_MAX_PCT = 100.0
+OPENING_STEP_PCT = 10.0
+RANDOM_SEED = 42            # 设为整数可复现同一打乱顺序
+
+# 采样与稳态判定
+SAMPLE_PERIOD_S = 0.1                                  # 采样周期（秒）
+STEADY_STATE_WINDOW_S = config.STEADY_STATE_WINDOW_S   # 滑窗长度，复用 config
+STEADY_STATE_STD_PCT = config.STEADY_STATE_STD_PCT     # 相对标准差阈值（%）
+MAX_WAIT_S = 60.0                                      # 单个开度的最长等待时间
+MIN_ABS_STD_SLM = 0.05                                 # 近零流量时的绝对标准差下限
+MAX_CONSECUTIVE_READ_FAILURES = 5                      # 后流量连续读取失败判定设备断开
+
+# 安全阈值（沿用 config 的闭环安全值；0% 开度时近零噪声若略为负，可放宽流量下限）
+SAFETY_FLOW_MIN_SLM = config.FLOW_VALID_MIN_SLM
+SAFETY_FLOW_MAX_SLM = config.FLOW_VALID_MAX_SLM
+SAFETY_MAX_PRESSURE_KPA = config.MAX_PRESSURE_KPA
+
+# 输出
+OUTPUT_DIRECTORY = "open_loop_data"                    # 相对工程目录
+PLOT_DPI = config.PLOT_DPI
+ANNOTATE_STEPS = True                                  # 曲线上标注开度步
+
+
+CSV_FIELDS = (
+    "time_s",
+    "step_index",
+    "opening_pct",
+    "motor_position",
+    "flow_before_slm",
+    "flow_after_slm",
+    "pressure_before_kpa",
+    "pressure_after_kpa",
+)
+
+
+# ---------------------------------------------------------------------------
+# 稳态判定
+# ---------------------------------------------------------------------------
+
+class SteadyStateDetector:
+    """滑动窗口稳态判定，复用 PerformanceReporter 的滑窗标准差思路。
+
+    闭环有目标流量，所以用“均值落在目标 ±band”+“标准差低于阈值”两个条件；
+    开环没有目标，因此只保留“滑窗相对标准差足够小”这一条，表示信号不再
+    明显变化：
+
+        std <= max(STD_PCT% * |mean|, MIN_ABS_STD_SLM)
+
+    近零流量时以绝对下限兜底，避免除以零。
+    """
+
+    def __init__(self, window_s, std_pct, sample_period_s=0.1, min_abs_std_slm=0.05):
+        self.window_s = float(window_s)
+        self.std_pct = float(std_pct)
+        self.sample_period_s = float(sample_period_s)
+        self.min_abs_std_slm = float(min_abs_std_slm)
+        self._window = deque()  # 元素为 (elapsed_s, flow_slm)
+
+    def reset(self):
+        self._window.clear()
+
+    def observe(self, elapsed_s, flow_slm):
+        """喂入一个样本；达到稳态返回 True，否则返回 False。"""
+        if flow_slm is None or not math.isfinite(float(flow_slm)):
+            return False
+
+        self._window.append((float(elapsed_s), float(flow_slm)))
+        cutoff = float(elapsed_s) - self.window_s
+        while self._window and self._window[0][0] < cutoff:
+            self._window.popleft()
+
+        if len(self._window) < 2:
+            return False
+        # 滑窗覆盖时长需至少达到 window_s。真实时间戳带抖动，若严格比较
+        # span < window_s，几乎总差一个采样间隙而永远无法判定；放宽一个
+        # 采样周期的余量。
+        min_span = max(0.0, self.window_s - self.sample_period_s)
+        if self._window[-1][0] - self._window[0][0] < min_span:
+            return False
+
+        flows = [item[1] for item in self._window]
+        mean = statistics.mean(flows)
+        std = statistics.pstdev(flows)
+        threshold = max(self.std_pct / 100.0 * abs(mean), self.min_abs_std_slm)
+        return std <= threshold
+
+
+# ---------------------------------------------------------------------------
+# 数据记录与绘图
+# ---------------------------------------------------------------------------
+
+class OpenLoopRecorder:
+    """逐周期写 CSV，结束时输出两张子图（前后压力、前后流量）。"""
+
+    def __init__(self, output_directory, *, plot_dpi=160):
+        output_dir = Path(output_directory)
+        output_dir.mkdir(parents=True, exist_ok=True)
+
+        run_id = time.strftime("%Y%m%d_%H%M%S")
+        self.csv_path = output_dir / f"open_loop_{run_id}.csv"
+        self.image_path = output_dir / f"open_loop_{run_id}.png"
+        self.plot_dpi = int(plot_dpi)
+
+        self._file = self.csv_path.open("w", newline="", encoding="utf-8-sig")
+        self._writer = csv.DictWriter(self._file, fieldnames=CSV_FIELDS)
+        self._writer.writeheader()
+        self._file.flush()
+
+        self.sample_count = 0
+        self.steps = []          # 每个开度步的 (time_s, opening_pct)
+        self._last_step_index = None
+        self._closed = False
+
+    def record(
+        self,
+        *,
+        time_s,
+        step_index,
+        opening_pct,
+        motor_position,
+        flow_before_slm,
+        flow_after_slm,
+        pressure_before_kpa,
+        pressure_after_kpa,
+    ):
+        if self._closed:
+            raise RuntimeError("记录器已关闭")
+
+        self._writer.writerow(
+            {
+                "time_s": self._fmt(time_s),
+                "step_index": int(step_index),
+                "opening_pct": self._fmt(opening_pct),
+                "motor_position": self._fmt(motor_position),
+                "flow_before_slm": self._fmt(flow_before_slm),
+                "flow_after_slm": self._fmt(flow_after_slm),
+                "pressure_before_kpa": self._fmt(pressure_before_kpa),
+                "pressure_after_kpa": self._fmt(pressure_after_kpa),
+            }
+        )
+        self._file.flush()
+        self.sample_count += 1
+
+        if step_index != self._last_step_index:
+            self.steps.append((float(time_s), float(opening_pct)))
+            self._last_step_index = step_index
+
+    def finalize(self):
+        """关闭 CSV 并生成 PNG；无采样点时只保留 CSV。"""
+        self.close()
+        if self.sample_count == 0:
+            return None
+        self._create_plot()
+        return self.image_path
+
+    def close(self):
+        if not self._closed:
+            self._file.flush()
+            self._file.close()
+            self._closed = True
+
+    def _create_plot(self):
+        # 无界面后端，保证终端/无显示器环境也能保存图片。
+        import matplotlib
+
+        matplotlib.use("Agg")
+        import matplotlib.pyplot as plt
+
+        time_s = []
+        flow_before = []
+        flow_after = []
+        pressure_before = []
+        pressure_after = []
+
+        with self.csv_path.open("r", newline="", encoding="utf-8-sig") as file:
+            for row in csv.DictReader(file):
+                time_s.append(float(row["time_s"]))
+                flow_before.append(self._float_or_nan(row["flow_before_slm"]))
+                flow_after.append(self._float_or_nan(row["flow_after_slm"]))
+                pressure_before.append(
+                    self._float_or_nan(row["pressure_before_kpa"])
+                )
+                pressure_after.append(
+                    self._float_or_nan(row["pressure_after_kpa"])
+                )
+
+        figure, axes = plt.subplots(
+            2,
+            1,
+            figsize=(12, 9),
+            sharex=True,
+            constrained_layout=True,
+        )
+        figure.suptitle("Open-loop Valve Sweep", fontsize=15)
+
+        self._plot_series(
+            axes[0], time_s, pressure_before,
+            "Pressure before valve", "#1565C0",
+        )
+        self._plot_series(
+            axes[0], time_s, pressure_after,
+            "Pressure after valve", "#D84315",
+        )
+        axes[0].set_ylabel("Pressure (kPa)")
+        axes[0].set_title("Pressure across control valve")
+        axes[0].legend(loc="best")
+
+        self._plot_series(
+            axes[1], time_s, flow_before,
+            "Flow before valve", "#2E7D32",
+        )
+        self._plot_series(
+            axes[1], time_s, flow_after,
+            "Flow after valve", "#6A1B9A",
+        )
+        axes[1].set_ylabel("Flow (SLM)")
+        axes[1].set_xlabel("Time (s)")
+        axes[1].set_title("Flow across control valve")
+        axes[1].legend(loc="best")
+
+        if ANNOTATE_STEPS and self.steps:
+            y_top = axes[1].get_ylim()[1]
+            for step_time, opening in self.steps:
+                for axis in axes:
+                    axis.axvline(
+                        step_time,
+                        color="0.5",
+                        linestyle="--",
+                        linewidth=0.7,
+                        alpha=0.6,
+                    )
+                axes[1].text(
+                    step_time,
+                    y_top,
+                    f"{opening:.0f}%",
+                    rotation=90,
+                    fontsize=7,
+                    ha="left",
+                    va="bottom",
+                    color="0.3",
+                )
+
+        for axis in axes:
+            axis.grid(True, alpha=0.3, linestyle="--")
+            axis.margins(x=0)
+
+        figure.savefig(self.image_path, dpi=self.plot_dpi, bbox_inches="tight")
+        plt.close(figure)
+
+    @staticmethod
+    def _plot_series(axis, time_s, series, label, color):
+        if any(math.isfinite(value) for value in series):
+            axis.plot(time_s, series, color=color, linewidth=1.4, label=label)
+        else:
+            axis.text(
+                0.5,
+                0.5,
+                f"No data: {label}",
+                transform=axis.transAxes,
+                ha="center",
+                va="center",
+                color="gray",
+            )
+
+    @staticmethod
+    def _fmt(value):
+        if value is None:
+            return ""
+        number = float(value)
+        return "" if not math.isfinite(number) else f"{number:.6f}"
+
+    @staticmethod
+    def _float_or_nan(value):
+        if value in (None, ""):
+            return math.nan
+        try:
+            return float(value)
+        except (TypeError, ValueError):
+            return math.nan
+
+
+# ---------------------------------------------------------------------------
+# 硬件读写辅助
+# ---------------------------------------------------------------------------
+
+def _safe_read(channel, read_fn):
+    """读取单路传感器；通道为 None 或读取失败时返回 None，不抛异常。"""
+    if channel is None:
+        return None
+    try:
+        return read_fn(channel)
+    except Exception:
+        return None
+
+
+def read_sensors(hardware):
+    """读取控流阀前后四个传感器，返回 (前压, 前流量, 后压, 后流量)。"""
+    pressure_before = _safe_read(PRESSURE_BEFORE_CHANNEL, hardware.get_pressure)
+    flow_before = _safe_read(FLOW_BEFORE_CHANNEL, hardware.get_flow)
+    pressure_after = _safe_read(PRESSURE_AFTER_CHANNEL, hardware.get_pressure)
+    flow_after = _safe_read(FLOW_AFTER_CHANNEL, hardware.get_flow)
+    return pressure_before, flow_before, pressure_after, flow_after
+
+
+def check_safety(pressure_before, flow_before, pressure_after, flow_after):
+    """越限检查，返回故障描述字符串；正常返回 None。
+
+    沿用闭环 FlowControlLoop 的安全阈值：流量超出
+    [SAFETY_FLOW_MIN_SLM, SAFETY_FLOW_MAX_SLM]，或任一压力超过
+    SAFETY_MAX_PRESSURE_KPA 时判定为故障（顺带识别未接线通道的负读数）。
+    """
+    if (
+        flow_after is not None
+        and not SAFETY_FLOW_MIN_SLM <= flow_after <= SAFETY_FLOW_MAX_SLM
+    ):
+        return f"FLOW_AFTER_OUT_OF_RANGE({flow_after:.3f} SLM)"
+    if (
+        flow_before is not None
+        and not SAFETY_FLOW_MIN_SLM <= flow_before <= SAFETY_FLOW_MAX_SLM
+    ):
+        return f"FLOW_BEFORE_OUT_OF_RANGE({flow_before:.3f} SLM)"
+    if pressure_after is not None and pressure_after > SAFETY_MAX_PRESSURE_KPA:
+        return f"PRESSURE_AFTER_OVER_LIMIT({pressure_after:.3f} kPa)"
+    if pressure_before is not None and pressure_before > SAFETY_MAX_PRESSURE_KPA:
+        return f"PRESSURE_BEFORE_OVER_LIMIT({pressure_before:.3f} kPa)"
+    return None
+
+
+def set_opening(hardware, opening_pct):
+    """把 0~100% 开度换算成反向电机行程并下发给 MT2-AM8。"""
+    position = opening_to_motor_position(
+        opening_pct,
+        config.MOTOR_OPEN_POSITION,
+        config.MOTOR_CLOSED_POSITION,
+    )
+    return bool(
+        hardware.set_motor_position(position, channel=MOTOR_CHANNEL)
+    ), position
+
+
+def generate_openings(min_pct, max_pct, step_pct, seed):
+    """生成 [min, max] 内 step 间隔的开度序列并随机打乱。"""
+    count = int(round((max_pct - min_pct) / step_pct)) + 1
+    openings = [round(min_pct + i * step_pct, 6) for i in range(count)]
+    # step 不能整除区间时，自动补上 max 端点，保证上下限都被覆盖。
+    if not openings or abs(openings[-1] - max_pct) > 1e-6:
+        openings.append(float(max_pct))
+    rng = random.Random(seed)
+    rng.shuffle(openings)
+    return openings
+
+
+# ---------------------------------------------------------------------------
+# 日志
+# ---------------------------------------------------------------------------
+
+def build_logger():
+    log_dir = Path(__file__).resolve().parent / config.LOG_DIRECTORY
+    log_dir.mkdir(parents=True, exist_ok=True)
+    log_path = log_dir / "open_loop.log"
+
+    logger = logging.getLogger("open_loop")
+    logger.setLevel(logging.INFO)
+    logger.handlers.clear()
+
+    formatter = logging.Formatter(
+        "%(asctime)s.%(msecs)03d %(levelname)s %(message)s",
+        datefmt="%Y-%m-%d %H:%M:%S",
+    )
+    file_handler = logging.FileHandler(log_path, encoding="utf-8")
+    file_handler.setFormatter(formatter)
+    logger.addHandler(file_handler)
+
+    console_handler = logging.StreamHandler()
+    console_handler.setFormatter(formatter)
+    logger.addHandler(console_handler)
+    return logger
+
+
+# ---------------------------------------------------------------------------
+# 主流程
+# ---------------------------------------------------------------------------
+
+def parse_args(argv=None):
+    parser = argparse.ArgumentParser(description="MT2-AM8 气体流量开环扫点实验")
+    parser.add_argument("--min", type=float, default=OPENING_MIN_PCT,
+                        help="扫描起始开度（%%），默认 %.0f" % OPENING_MIN_PCT)
+    parser.add_argument("--max", type=float, default=OPENING_MAX_PCT,
+                        help="扫描结束开度（%%），默认 %.0f" % OPENING_MAX_PCT)
+    parser.add_argument("--step", type=float, default=OPENING_STEP_PCT,
+                        help="开度间隔（%%），默认 %.0f" % OPENING_STEP_PCT)
+    parser.add_argument("--seed", type=int, default=RANDOM_SEED,
+                        help="随机种子，默认 None（每次打乱顺序不同）")
+    args = parser.parse_args(argv)
+    if args.step <= 0:
+        parser.error("--step 必须大于 0")
+    if not (0 <= args.min < args.max <= 100):
+        parser.error("必须满足 0 <= --min < --max <= 100")
+    return args
+
+
+def run(args):
+    config.validate_config()
+    logger = build_logger()
+
+    try:
+        from PcControl import MT2AM8Client
+    except ModuleNotFoundError as exc:
+        if exc.name and exc.name.startswith("pymodbus"):
+            logger.critical(
+                "缺少 PcControl.py 所需的 pymodbus；请在运行本工程的 Python "
+                "环境中安装与现有硬件代码兼容的 pymodbus 版本"
+            )
+            return 6
+        raise
+
+    hardware = MT2AM8Client(
+        host=config.MT2AM8_HOST,
+        port=config.MT2AM8_PORT,
+        slave_id=config.MT2AM8_SLAVE_ID,
+        pressure_range=config.PRESSURE_RANGE_KPA,
+        flow_range=config.FLOW_METER_RANGE_SLM,
+    )
+
+    openings = generate_openings(
+        args.min, args.max, args.step, args.seed
+    )
+    logger.info(
+        "开度扫描顺序（已打乱）: %s",
+        ", ".join(f"{o:.1f}%" for o in openings),
+    )
+
+    connected = False
+    recorder = None
+    exit_code = 0
+
+    try:
+        logger.info("正在连接 MT2-AM8 %s:%s", config.MT2AM8_HOST, config.MT2AM8_PORT)
+        connected = bool(hardware.connect())
+        if not connected:
+            logger.critical("无法连接 MT2-AM8，实验中止")
+            exit_code = 2
+            return exit_code
+
+        # 启动前先关阀。
+        ok, _ = set_opening(hardware, 0.0)
+        if not ok:
+            logger.critical("启动前关阀失败，实验中止")
+            exit_code = 2
+            return exit_code
+
+        recorder = OpenLoopRecorder(
+            Path(__file__).resolve().parent / OUTPUT_DIRECTORY,
+            plot_dpi=PLOT_DPI,
+        )
+        logger.info("逐周期数据将保存到 %s", recorder.csv_path)
+
+        detector = SteadyStateDetector(
+            STEADY_STATE_WINDOW_S,
+            STEADY_STATE_STD_PCT,
+            sample_period_s=SAMPLE_PERIOD_S,
+            min_abs_std_slm=MIN_ABS_STD_SLM,
+        )
+
+        run_start = time.perf_counter()
+        read_failures = 0
+
+        for step_index, opening in enumerate(openings):
+            ok, position = set_opening(hardware, opening)
+            if not ok:
+                logger.critical(
+                    "写入开度 %.1f%% 对应行程 %.1f 失败，实验中止", opening, position
+                )
+                exit_code = 2
+                break
+
+            logger.info(
+                "步 %d/%d：设置开度 %.1f%%（行程 %.1f），等待后流量稳定…",
+                step_index + 1, len(openings), opening, position,
+            )
+
+            detector.reset()
+            step_start = time.perf_counter()
+            next_sample = step_start
+            stable = False
+
+            while True:
+                now = time.perf_counter()
+                elapsed_s = now - run_start
+
+                pressure_before, flow_before, pressure_after, flow_after = (
+                    read_sensors(hardware)
+                )
+                recorder.record(
+                    time_s=elapsed_s,
+                    step_index=step_index,
+                    opening_pct=opening,
+                    motor_position=position,
+                    flow_before_slm=flow_before,
+                    flow_after_slm=flow_after,
+                    pressure_before_kpa=pressure_before,
+                    pressure_after_kpa=pressure_after,
+                )
+
+                fault = check_safety(
+                    pressure_before, flow_before, pressure_after, flow_after
+                )
+                if fault is not None:
+                    logger.critical("安全阈值触发：%s，实验中止并关阀", fault)
+                    exit_code = 2
+                    break
+
+                if flow_after is None:
+                    read_failures += 1
+                else:
+                    read_failures = 0
+                if read_failures >= MAX_CONSECUTIVE_READ_FAILURES:
+                    logger.critical(
+                        "控流阀后流量连续 %d 次读取失败，判定设备断开，实验中止",
+                        read_failures,
+                    )
+                    exit_code = 2
+                    break
+
+                if detector.observe(elapsed_s, flow_after):
+                    stable = True
+                    settle_s = now - step_start
+                    flow_text = "--" if flow_after is None else f"{flow_after:.3f}"
+                    logger.info(
+                        "步 %d 完成：开度 %.1f%% 后流量稳定 %s SLM，耗时 %.2f s",
+                        step_index + 1, opening, flow_text, settle_s,
+                    )
+                    break
+
+                if now - step_start >= MAX_WAIT_S:
+                    logger.warning(
+                        "开度 %.1f%% 在 %.1f s 内未稳定，强制进入下一开度",
+                        opening, MAX_WAIT_S,
+                    )
+                    break
+
+                next_sample += SAMPLE_PERIOD_S
+                sleep_s = next_sample - time.perf_counter()
+                if sleep_s > 0:
+                    time.sleep(sleep_s)
+                else:
+                    next_sample = time.perf_counter()
+
+            if exit_code != 0:
+                break
+
+    except KeyboardInterrupt:
+        logger.info("收到 Ctrl+C，正在停止实验")
+    except Exception:
+        exit_code = 3
+        logger.exception("未处理异常，实验进入收尾")
+    finally:
+        if connected:
+            try:
+                close_ok, _ = set_opening(hardware, 0.0)
+                if not close_ok:
+                    exit_code = max(exit_code, 4)
+                    logger.critical("收尾安全关阀失败，请立即人工检查")
+            except Exception:
+                exit_code = max(exit_code, 4)
+                logger.exception("收尾安全关阀时发生异常")
+            try:
+                hardware.disconnect()
+            except Exception:
+                exit_code = max(exit_code, 5)
+                logger.exception("断开 MT2-AM8 时发生异常")
+
+        if recorder is not None:
+            try:
+                image_path = recorder.finalize()
+                logger.info("开环数据已保存：%s", recorder.csv_path)
+                if image_path is not None:
+                    logger.info("曲线图已保存：%s", image_path)
+                else:
+                    logger.warning("本次运行没有采样点，因此未生成曲线图")
+            except Exception:
+                recorder.close()
+                exit_code = max(exit_code, 7)
+                logger.exception("保存曲线图失败；CSV 数据仍保留")
+
+        logger.info("程序结束，退出码=%d", exit_code)
+    return exit_code
+
+
+def main(argv=None):
+    args = parse_args(argv)
+    return run(args)
+
+
+if __name__ == "__main__":
+    import sys
+
+    sys.exit(main())