新增两个命令行参数python test_valve.py --opening 60 --dt 0.01
修改传感器通道号
This commit is contained in:
@@ -13,12 +13,12 @@ MT2AM8_PORT = 502
|
||||
MT2AM8_SLAVE_ID = 1
|
||||
|
||||
# MT2AM8Client 用这些量程把模拟量转换成物理量。
|
||||
PRESSURE_RANGE_KPA = 1600.0
|
||||
PRESSURE_RANGE_KPA = 400.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
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
@@ -49,8 +49,8 @@ PID_NEAR_KD = 0.0
|
||||
|
||||
OPENING_MIN_PCT = 0.0
|
||||
OPENING_MAX_PCT = 100.0
|
||||
MAX_OPENING_RATE_FAR_PCT_S = 20.0
|
||||
MAX_OPENING_RATE_NEAR_PCT_S = 2.0
|
||||
MAX_OPENING_RATE_FAR_PCT_S = 200.0
|
||||
MAX_OPENING_RATE_NEAR_PCT_S = 20.0
|
||||
|
||||
# FAR -> NEAR：绝对误差连续满足 near 阈值指定周期数后切换。
|
||||
# NEAR -> FAR：绝对误差连续达到 far 阈值指定周期数后切换。
|
||||
|
||||
+1
-1
@@ -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 通道
|
||||
|
||||
+63
-19
@@ -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:
|
||||
|
||||
Reference in New Issue
Block a user