diff --git a/config.py b/config.py index 4c082e5..f714035 100644 --- a/config.py +++ b/config.py @@ -17,8 +17,8 @@ PRESSURE_RANGE_KPA = 1600.0 FLOW_METER_RANGE_SLM = 300.0 # AI/AO 通道均从 0 开始。AI0 和 AO0 属于不同的寄存器区,可以同时使用。 -FLOW_INPUT_CHANNEL = 1 -PRESSURE_INPUT_CHANNEL = 0 # 没有压力传感器时改为 None +FLOW_INPUT_CHANNEL = 2 +PRESSURE_INPUT_CHANNEL = 3 # 没有压力传感器时改为 None MOTOR_OUTPUT_CHANNEL = 1 # --------------------------------------------------------------------------- diff --git a/open_loop.py b/open_loop.py index ef70719..d01ad39 100644 --- a/open_loop.py +++ b/open_loop.py @@ -41,7 +41,7 @@ from flow_control import opening_to_motor_position # “前”= 控流阀入口侧(靠近减压阀一侧),为本次新增传感器,请按实际接线确认。 PRESSURE_AFTER_CHANNEL = config.PRESSURE_INPUT_CHANNEL # 后压力,默认 0 FLOW_AFTER_CHANNEL = config.FLOW_INPUT_CHANNEL # 后流量,默认 1 -PRESSURE_BEFORE_CHANNEL = 2 # 前压力(新增,需确认) +PRESSURE_BEFORE_CHANNEL = None # 前压力(新增,需确认) FLOW_BEFORE_CHANNEL = None # 前流量(新增,需确认) MOTOR_CHANNEL = config.MOTOR_OUTPUT_CHANNEL # AO 通道 diff --git a/test_valve.py b/test_valve.py index f31b949..b88ad47 100644 --- a/test_valve.py +++ b/test_valve.py @@ -1,5 +1,11 @@ -"""手动设置阀门开度,并每秒显示流量、压力和电机行程。""" +"""手动设置阀门开度,按 --dt 周期写 CSV,终端每秒显示一次。 +示例: + python test_valve.py + python test_valve.py --opening 60 --dt 0.01 +""" + +import argparse import contextlib import csv from datetime import datetime @@ -18,9 +24,10 @@ import config from flow_control import opening_to_motor_position -SAMPLE_PERIOD_S = 1.0 +SAMPLE_PERIOD_S = 1.0 # CSV 采样周期默认值(秒),可用 --dt 覆盖 +PRINT_PERIOD_S = 1.0 # 终端打印周期(秒),与采样周期解耦 KEY_POLL_PERIOD_S = 0.05 -INITIAL_OPENING_PCT = 100.0 +INITIAL_OPENING_PCT = 100.0 # 初始开度默认值(%),可用 --opening 覆盖 OUTPUT_DIRECTORY = Path(__file__).resolve().parent / "test_valve_data" @@ -193,7 +200,34 @@ def read_sensors(hardware, quiet=False): return flow, pressure -def main(): +def parse_args(argv=None): + """解析命令行参数:--dt 采样周期(秒)、--opening 初始开度(%)。""" + parser = argparse.ArgumentParser( + description="手动设置阀门开度:按 --dt 周期写 CSV,终端每秒显示一次。" + ) + parser.add_argument( + "--dt", + type=float, + default=SAMPLE_PERIOD_S, + help=f"CSV 采样周期(秒),默认 {SAMPLE_PERIOD_S:g}", + ) + parser.add_argument( + "--opening", + type=float, + default=INITIAL_OPENING_PCT, + help=f"初始阀门开度(%%),默认 {INITIAL_OPENING_PCT:g}", + ) + args = parser.parse_args(argv) + if args.dt <= 0: + parser.error("--dt 必须大于 0") + if not 0.0 <= args.opening <= 100.0: + parser.error("--opening 必须位于 0~100 之间") + return args + + +def main(argv=None): + args = parse_args(argv) + if msvcrt is None: print("本程序需要在 Windows 终端中运行。") return 1 @@ -213,7 +247,7 @@ def main(): ) connected = False - opening = INITIAL_OPENING_PCT + opening = args.opening position = motor_position_for(opening) input_mode = False input_buffer = "" @@ -238,6 +272,9 @@ def main(): print(f"CSV 将保存到:{recorder.csv_path}") next_sample_time = time.perf_counter() + next_print_time = time.perf_counter() + last_flow = None + last_pressure = None while True: while msvcrt.kbhit(): @@ -289,21 +326,28 @@ def main(): print(text, end="", flush=True) now = time.perf_counter() - if now >= next_sample_time: - flow, pressure = read_sensors(hardware, quiet=input_mode) - recorder.record(now, flow, opening, pressure, position) - if not input_mode: - timestamp = time.strftime("%H:%M:%S") - print( - f"[{timestamp}] " - f"流量={format_value(flow, 'SLM')} " - f"压力={format_value(pressure, 'kPa')} " - f"开度={opening:.2f}% 行程={position:.1f}" - ) - # 不追补错过的终端输出;输入期间的样本只写入 CSV。 - next_sample_time = time.perf_counter() + SAMPLE_PERIOD_S - time.sleep(KEY_POLL_PERIOD_S) + if now >= next_sample_time: + last_flow, last_pressure = read_sensors( + hardware, quiet=input_mode + ) + recorder.record( + now, last_flow, opening, last_pressure, position + ) + # 不追补错过的采样;输入期间的样本只写 CSV,不打印终端。 + next_sample_time = time.perf_counter() + args.dt + + if not input_mode and now >= next_print_time: + timestamp = time.strftime("%H:%M:%S") + print( + f"[{timestamp}] " + f"流量={format_value(last_flow, 'SLM')} " + f"压力={format_value(last_pressure, 'kPa')} " + f"开度={opening:.2f}% 行程={position:.1f}" + ) + next_print_time = time.perf_counter() + PRINT_PERIOD_S + + time.sleep(min(KEY_POLL_PERIOD_S, args.dt)) except KeyboardInterrupt: if input_mode: