649 lines
23 KiB
Python
649 lines
23 KiB
Python
"""气体流量开环扫点实验（不参与闭环反馈，也不被 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())
|