"""气体流量闭环控制程序入口。 示例： python main.py --target 50 程序启动后直接从当前阀门开度（默认全开）开始闭环。Ctrl+C、控制故障或 其他异常会断开 MT2-AM8；阀门不被驱动时默认回到全开。 """ import argparse import logging from pathlib import Path import sys import time try: import msvcrt # Windows 专用：非阻塞键盘输入，用于运行中切换目标流量 except ImportError: # 非 Windows 平台无此模块，退化为仅周期控制 msvcrt = None import config from controllers import IncrementalPID from data_logger import FlowRunRecorder from flow_control import FlowControlFault, FlowControlLoop from performance_reporter import PerformanceReporter class ConsoleGate(logging.Filter): """终端输出闸门：输入模式下挡住终端日志，文件日志不受影响。""" def __init__(self): super().__init__() self.blocked = False def filter(self, _record): return not self.blocked def build_logger(): log_dir = Path(__file__).resolve().parent / config.LOG_DIRECTORY log_dir.mkdir(parents=True, exist_ok=True) log_path = log_dir / config.LOG_FILE_NAME logger = logging.getLogger("flow_control") 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) console_gate = ConsoleGate() console_handler.addFilter(console_gate) logger.addHandler(console_handler) return logger, console_gate def build_control_loop(hardware, logger): pid = IncrementalPID( kp=config.PID_KP, ki=config.PID_KI, kd=config.PID_KD, dt=config.CONTROL_PERIOD_S, out_min=config.OPENING_MIN_PCT, out_max=config.OPENING_MAX_PCT, output_rate_limit=config.MAX_OPENING_RATE_PCT_S, ) return FlowControlLoop( hardware=hardware, pid=pid, period_s=config.CONTROL_PERIOD_S, flow_channel=config.FLOW_INPUT_CHANNEL, pressure_channel=config.PRESSURE_INPUT_CHANNEL, motor_channel=config.MOTOR_OUTPUT_CHANNEL, motor_open_position=config.MOTOR_OPEN_POSITION, motor_closed_position=config.MOTOR_CLOSED_POSITION, target_min_slm=config.TARGET_FLOW_MIN_SLM, target_max_slm=config.TARGET_FLOW_MAX_SLM, zero_flow_threshold_slm=config.ZERO_FLOW_THRESHOLD_SLM, flow_valid_min_slm=config.FLOW_VALID_MIN_SLM, flow_valid_max_slm=config.FLOW_VALID_MAX_SLM, max_pressure_kpa=config.MAX_PRESSURE_KPA, max_control_dt_s=config.MAX_CONTROL_DT_S, max_consecutive_flow_failures=config.MAX_CONSECUTIVE_FLOW_FAILURES, max_consecutive_pressure_failures=( config.MAX_CONSECUTIVE_PRESSURE_FAILURES ), logger=logger, ) def parse_args(argv=None): parser = argparse.ArgumentParser(description="MT2-AM8 气体流量 PID 控制") parser.add_argument( "--target", type=float, required=True, help=( "目标流量 SLM，允许范围 " f"{config.TARGET_FLOW_MIN_SLM}~{config.TARGET_FLOW_MAX_SLM}" ), ) return parser.parse_args(argv) def format_status(result): flow_text = ( "--" if result.measured_flow_slm is None else f"{result.measured_flow_slm:.2f}" ) pressure_text = ( "--" if result.pressure_kpa is None else f"{result.pressure_kpa:.2f}" ) return ( f"状态={result.status} 目标={result.target_flow_slm:.2f} SLM " f"实际={flow_text} SLM 压力={pressure_text} kPa " f"开度={result.opening_pct:.2f}% " f"行程={result.motor_position:.1f} dt={result.actual_dt_s:.3f}s" ) def run(target_flow_slm): config.validate_config() logger, console_gate = 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, ) controller = build_control_loop(hardware, logger) controller.set_target_flow(target_flow_slm) connected = False recorder = None reporter = None exit_code = 0 try: logger.info( "正在连接 MT2-AM8 %s:%s，目标流量=%.3f SLM", config.MT2AM8_HOST, config.MT2AM8_PORT, target_flow_slm, ) connected = bool(hardware.connect()) if not connected: raise FlowControlFault( "DEVICE_CONNECT_FAILED", "STARTUP", "无法连接 MT2-AM8", { "host": config.MT2AM8_HOST, "port": config.MT2AM8_PORT, "slave_id": config.MT2AM8_SLAVE_ID, }, ) # 控制前读取一次当前流量，作为超调量方向的基准。 try: initial_flow_slm = hardware.get_flow(config.FLOW_INPUT_CHANNEL) except Exception: initial_flow_slm = None reporter = PerformanceReporter( controller.pid, window_s=config.STEADY_STATE_WINDOW_S, sample_period_s=config.CONTROL_PERIOD_S, band_pct=config.STEADY_STATE_BAND_PCT, std_pct=config.STEADY_STATE_STD_PCT, flow_range_min_slm=config.TARGET_FLOW_MIN_SLM, flow_range_max_slm=config.TARGET_FLOW_MAX_SLM, initial_flow_slm=initial_flow_slm, logger=logger, ) controller.start(initial_opening=config.INITIAL_OPENING_PCT) recorder = FlowRunRecorder( Path(__file__).resolve().parent / config.DATA_DIRECTORY, plot_dpi=config.PLOT_DPI, ) logger.info("逐周期数据将保存到 %s", recorder.csv_path) logger.info( "闭环控制已启动；按 Ctrl+C 停止，按 %s 切换目标流量", config.TARGET_SWITCH_KEY.upper(), ) input_mode = False input_buf = "" def apply_target(): """把输入缓冲区解析为新的目标流量并生效。""" raw = input_buf.strip() if not raw: logger.warning("未输入目标流量，已忽略") return try: value = float(raw) except ValueError: logger.warning("无法解析目标流量 %r，已忽略", raw) return try: controller.set_target_flow(value) except ValueError as exc: logger.warning("目标流量设置失败：%s", exc) return reporter.reset() logger.info("目标流量已切换为 %.3f SLM", value) def process_keys(): """非阻塞处理键盘输入；输入模式下控制循环照常运行。""" nonlocal input_mode, input_buf if msvcrt is None: return while msvcrt.kbhit(): ch = msvcrt.getch() if not input_mode: if ch == b"\x03": # Ctrl+C，交给外层 KeyboardInterrupt 处理 raise KeyboardInterrupt key = ch.decode("ascii", errors="ignore").lower() if key == config.TARGET_SWITCH_KEY: input_mode = True input_buf = "" console_gate.blocked = True print( f"当前目标={controller.target_flow_slm:.2f} SLM，" f"请输入新目标流量 (回车确认, Esc 取消): ", end="", flush=True, ) else: if ch in (b"\r", b"\n"): print() console_gate.blocked = False input_mode = False apply_target() elif ch == b"\x1b": # Esc 取消 console_gate.blocked = False input_mode = False print(" (已取消)") elif ch in (b"\x08", b"\x7f"): # 退格删除最后一位 if input_buf: input_buf = input_buf[:-1] print("\b \b", end="", flush=True) else: text = ch.decode("ascii", errors="ignore") if text and text.isprintable(): input_buf += text print(text, end="", flush=True) next_deadline = time.perf_counter() next_status_time = next_deadline while True: now = time.perf_counter() process_keys() result = controller.step(now=now) recorder.record(result) reporter.observe(result) if config.AUTO_STOP_ON_STEADY and reporter.steady_reached: logger.info("已达到稳态，自动停止控制") break if now >= next_status_time or result.status != "OK": logger.info(format_status(result)) next_status_time = now + config.STATUS_PRINT_PERIOD_S next_deadline += config.CONTROL_PERIOD_S sleep_s = next_deadline - time.perf_counter() if sleep_s > 0: time.sleep(sleep_s) else: # 丢弃已经错过的节拍，避免连续快速补跑 PID。 next_deadline = time.perf_counter() except KeyboardInterrupt: console_gate.blocked = False logger.info("收到 Ctrl+C，正在停止控制") except FlowControlFault as exc: console_gate.blocked = False exit_code = 2 logger.critical("控制故障：%s", exc) except Exception: console_gate.blocked = False exit_code = 3 logger.exception("未处理异常，系统将退出") finally: console_gate.blocked = False if connected: 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 数据仍保留") if reporter is not None: try: reporter.finalize() except Exception: logger.exception("生成控制性能报告时发生异常") logger.info("程序结束，退出码=%d", exit_code) return exit_code def main(argv=None): args = parse_args(argv) return run(args.target) if __name__ == "__main__": sys.exit(main())