支持运行中切换目标流量:按 S 进入输入模式,回车生效、Esc 取消

- 输入模式下控制循环照常运行、原目标继续生效,直到回车确认新值
- 输入期间终端输出整体静默(状态行/性能报告均不打印),文件日志与 CSV 照常写入
- 新增 ConsoleGate 过滤器挂到 console_handler,退出路径统一解除静默
- 新增 PerformanceReporter.reset(),切换目标后重新判稳并再次输出性能报告
- 新增 config.TARGET_SWITCH_KEY 配置项,README 同步说明
This commit is contained in:
2026-08-21 03:36:17 +08:00
parent 7732c36c55
commit 292815e907
4 changed files with 124 additions and 3 deletions
+92 -3
View File
@@ -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()