diff --git a/README.md b/README.md index d9b198a..9bcce50 100644 --- a/README.md +++ b/README.md @@ -75,6 +75,18 @@ python main.py --target 50 `--target` 的单位为 SLM,必须位于 `TARGET_FLOW_MIN_SLM~TARGET_FLOW_MAX_SLM`(当前为 `0~300`)范围内。按 `Ctrl+C` 停止;默认 `AUTO_STOP_ON_STEADY=False`,所以达到稳态后仍会继续控制。如需首次达到稳态后自动结束,将该配置改为 `True`。 +### 运行中切换目标流量 + +闭环运行过程中无需重启即可切换目标流量:在终端按 `S`(`TARGET_SWITCH_KEY`)进入输入模式,屏幕会提示当前目标并等待输入,输入数字后按回车生效,按 `Esc` 取消。输入模式期间: + +- 控制循环照常运行,原目标流量继续生效,直到回车确认新值; +- 终端输出整体静默(状态行、性能报告均不打印),避免与输入回显混在一起; +- 逐周期 CSV 与 `logs/flow_control.log` 照常写入,不丢数据。 + +新目标生效后会重置稳态判定(`PerformanceReporter.reset()`),因此切换后的目标仍会重新判稳并输出一次性能报告。目标值需位于 `TARGET_FLOW_MIN_SLM~TARGET_FLOW_MAX_SLM` 范围内,非法或超范围输入会被拒绝并提示,不影响控制。 + +> 该热键仅在真实控制台(Windows 的 `msvcrt`)下可用;`Ctrl+C` 无论是否处于输入模式都会停止控制。 + 启动流程为: ```text @@ -207,6 +219,7 @@ CSV 同时记录 `is_ratio_smaller_than_0.528`:压力比小于临界值 `0.528 | `MOTOR_OUTPUT_CHANNEL` | `1` | 电机 AO 通道 | | `FLOW_METER_RANGE_SLM` / `PRESSURE_RANGE_KPA` | `300` / `400` | 传感器量程 | | `TARGET_FLOW_MIN_SLM` / `TARGET_FLOW_MAX_SLM` | `0` / `300` | 允许的目标范围 | +| `TARGET_SWITCH_KEY` | `s` | 运行中按此键进入"输入新目标流量"模式 | | `CONTROL_PERIOD_S` / `MAX_CONTROL_DT_S` | `0.1` / `0.3` | 目标周期 / 超时阈值 | | `PID_KP` / `PID_KI` / `PID_KD` | `0.5` / `0.5` / `0.0` | 当前 PI 参数,需按现场继续整定 | | `OPENING_MIN_PCT` / `OPENING_MAX_PCT` | `0` / `100` | PID 输出开度范围 | diff --git a/config.py b/config.py index 46f1203..eaa7237 100644 --- a/config.py +++ b/config.py @@ -28,6 +28,7 @@ MOTOR_OUTPUT_CHANNEL = 1 TARGET_FLOW_MIN_SLM = 0.0 TARGET_FLOW_MAX_SLM = 300.0 ZERO_FLOW_THRESHOLD_SLM = 0.5 +TARGET_SWITCH_KEY = "s" # 运行中按此键进入"输入新目标流量"模式 CONTROL_PERIOD_S = 0.1 # 100 ms,10 Hz MAX_CONTROL_DT_S = 0.3 # 超过此值视为控制循环严重超时 diff --git a/main.py b/main.py index c8c53d7..73d06c8 100644 --- a/main.py +++ b/main.py @@ -13,6 +13,11 @@ 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 @@ -20,6 +25,17 @@ 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) @@ -39,8 +55,10 @@ def build_logger(): console_handler = logging.StreamHandler() console_handler.setFormatter(formatter) + console_gate = ConsoleGate() + console_handler.addFilter(console_gate) logger.addHandler(console_handler) - return logger + return logger, console_gate def build_control_loop(hardware, logger): @@ -110,7 +128,7 @@ def format_status(result): def run(target_flow_slm): config.validate_config() - logger = build_logger() + logger, console_gate = build_logger() try: from PcControl import MT2AM8Client except ModuleNotFoundError as exc: @@ -180,12 +198,79 @@ def run(target_flow_slm): plot_dpi=config.PLOT_DPI, ) logger.info("逐周期数据将保存到 %s", recorder.csv_path) - logger.info("闭环控制已启动;按 Ctrl+C 停止") + 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) @@ -207,14 +292,18 @@ def run(target_flow_slm): 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() diff --git a/performance_reporter.py b/performance_reporter.py index b9f1d50..fd42b42 100644 --- a/performance_reporter.py +++ b/performance_reporter.py @@ -113,6 +113,24 @@ class PerformanceReporter: if not self._reached and self.target_flow_slm is not None: self._log("info", "本次运行未达到稳态,未生成控制性能报告") + def reset(self): + """清空判稳状态,使切换目标后能重新判定稳态并再次输出报告。 + + ``initial_flow_slm`` 一并置空,观察器会在下一个有效样本上重新把它 + 初始化为切换目标时刻的实测流量,从而保证新目标的超调方向判断正确。 + """ + self._reached = False + self.target_flow_slm = None + self.settling_time_s = None + self.steady_mean = None + self.steady_std = None + self.steady_error_slm = None + self._first_timestamp = None + self._peak_flow_slm = None + self._min_flow_slm = None + self.initial_flow_slm = None + self._window.clear() + @property def steady_reached(self): """是否已判定达到稳态(供主循环决定是否自动停止)。"""