diff --git a/README.md b/README.md new file mode 100644 index 0000000..ac05e2b --- /dev/null +++ b/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/ # 运行日志 +``` diff --git a/config.py b/config.py index fcc119e..b6f7074 100644 --- a/config.py +++ b/config.py @@ -19,7 +19,7 @@ FLOW_METER_RANGE_SLM = 300.0 # AI/AO 通道均从 0 开始。AI0 和 AO0 属于不同的寄存器区，可以同时使用。 FLOW_INPUT_CHANNEL = 1 PRESSURE_INPUT_CHANNEL = 0 # 没有压力传感器时改为 None -MOTOR_OUTPUT_CHANNEL = 0 +MOTOR_OUTPUT_CHANNEL = 1 # --------------------------------------------------------------------------- # 控制目标和控制周期 @@ -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 diff --git a/flow_control.py b/flow_control.py index 846bf2a..7db4a4c 100644 --- a/flow_control.py +++ b/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 diff --git a/logs/open_loop.log b/logs/open_loop.log new file mode 100644 index 0000000..4c5fd1c --- /dev/null +++ b/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 diff --git a/main.py b/main.py index f531eea..e0212a6 100644 --- a/main.py +++ b/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: diff --git a/open_loop.py b/open_loop.py new file mode 100644 index 0000000..d624983 --- /dev/null +++ b/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())