louis
a03735387c
按 plan.md 新增三个交付物，全部默认关闭以保持与纯 PID 的 A/B 兼容： - valve_model.py：阀特性模型 Q_ss = A_eff(x)·P1_abs·F(r)，含 ISO 6358 椭圆 F(r)、行程/面积互逆插值、单调性校验、JSON 存取。 - identify_valve.py：从 open_loop.py 扫点 CSV 拟合 A_eff 表并输出 JSON。 - 控制器集成：config/flow_control/main/data_logger 新增阀前压读取 （channel 2）、前馈打底 + PI 修残差、按 P1 增益调度，并记录/绘制前馈项。 新增配置 FEEDFORWARD_ENABLED / GAIN_SCHEDULE_ENABLED 默认 False， VALVE_MODEL_PATH 默认空，未加载模型时行为与原先纯 PID 完全一致。
125 lines
3.8 KiB
Python
125 lines
3.8 KiB
Python
"""valve_model / identify_valve 离线自测（不碰硬件）。
|
|
|
|
运行：
|
|
python test_valve_model.py
|
|
"""
|
|
|
|
import csv
|
|
import math
|
|
import tempfile
|
|
from pathlib import Path
|
|
|
|
from valve_model import (
|
|
P_ATM,
|
|
CRITICAL_RATIO,
|
|
ValveModel,
|
|
abs_pressure,
|
|
f_ratio,
|
|
)
|
|
|
|
|
|
def _close(a, b, rel=1e-6):
|
|
return abs(a - b) <= rel * max(1.0, abs(a), abs(b))
|
|
|
|
|
|
def test_f_ratio():
|
|
assert f_ratio(0.4) == 1.0
|
|
assert f_ratio(CRITICAL_RATIO) == 1.0
|
|
assert f_ratio(1.0) == 0.0
|
|
assert f_ratio(1.1) == 0.0
|
|
expected = math.sqrt(
|
|
1.0 - ((0.7 - CRITICAL_RATIO) / (1.0 - CRITICAL_RATIO)) ** 2
|
|
)
|
|
assert _close(f_ratio(0.7), expected)
|
|
|
|
|
|
def test_abs_pressure():
|
|
assert _close(abs_pressure(0.0), P_ATM)
|
|
assert _close(abs_pressure(300.0), 300.0 + P_ATM)
|
|
|
|
|
|
def test_valve_model_roundtrip():
|
|
# A_eff 随行程线性递减（x 越大开度越小、面积越小）。
|
|
table = [(800.0, 0.5), (900.0, 0.25), (1000.0, 0.0)]
|
|
model = ValveModel(table, motor_open=800.0, motor_closed=1000.0)
|
|
|
|
assert _close(model.area_from_stroke(900.0), 0.25)
|
|
assert _close(model.area_from_stroke(850.0), 0.375)
|
|
assert _close(model.stroke_from_area(0.25), 900.0)
|
|
|
|
# 阻塞流（P2=0 表压 → r≈0.2 < 0.528），flow_ss 与 feedforward_stroke 互逆。
|
|
for x in (800.0, 850.0, 900.0, 1000.0):
|
|
q = model.flow_ss(x, 400.0, 0.0)
|
|
x_back = model.feedforward_stroke(q, 400.0, 0.0)
|
|
assert _close(x, x_back, rel=0.01), (x, q, x_back)
|
|
|
|
# 亚声速（P2=300 表压 → r≈0.8）。
|
|
q = model.flow_ss(900.0, 400.0, 300.0)
|
|
assert _close(model.feedforward_stroke(q, 400.0, 300.0), 900.0, rel=0.01)
|
|
|
|
|
|
def test_non_monotonic_raises():
|
|
table = [(800.0, 0.5), (900.0, 0.1), (1000.0, 0.3)]
|
|
try:
|
|
ValveModel(table, motor_open=800.0, motor_closed=1000.0)
|
|
except ValueError:
|
|
return
|
|
raise AssertionError("非单调表应抛 ValueError")
|
|
|
|
|
|
def test_identify_end_to_end():
|
|
import identify_valve
|
|
|
|
# 构造合成扫点 CSV：每个 step 30 个采样点，全程稳态。
|
|
rows = []
|
|
for step, x in enumerate((800.0, 900.0, 1000.0)):
|
|
a_eff = (1000.0 - x) / 400.0 # 单调递减的真值
|
|
q_ss = a_eff * (400.0 + P_ATM) # 阻塞流，F=1
|
|
for i in range(30):
|
|
rows.append({
|
|
"time_s": f"{step * 10 + i * 0.1:.6f}",
|
|
"step_index": step,
|
|
"opening_pct": 100.0 - step * 50.0,
|
|
"motor_position": x,
|
|
"flow_before_slm": "",
|
|
"flow_after_slm": q_ss,
|
|
"pressure_before_kpa": 400.0,
|
|
"pressure_after_kpa": 0.0,
|
|
"P_abs_ratio": "",
|
|
"is_ratio_smaller_than_0.528": "Y",
|
|
})
|
|
|
|
with tempfile.TemporaryDirectory() as tmp:
|
|
csv_path = Path(tmp) / "sweep.csv"
|
|
out_path = Path(tmp) / "model.json"
|
|
with csv_path.open("w", newline="", encoding="utf-8-sig") as f:
|
|
writer = csv.DictWriter(f, fieldnames=list(rows[0].keys()))
|
|
writer.writeheader()
|
|
writer.writerows(rows)
|
|
|
|
model, steady, table, dropped, stats = identify_valve.identify(
|
|
csv_path, tail=20, out_path=str(out_path)
|
|
)
|
|
assert len(table) == 3, table
|
|
assert stats["choked"] == 3
|
|
assert not dropped
|
|
for x, a in table:
|
|
assert _close(a, (1000.0 - x) / 400.0, rel=0.01), (x, a)
|
|
|
|
loaded = ValveModel.load(str(out_path), 800.0, 1000.0)
|
|
assert _close(loaded.area_from_stroke(900.0), 0.25, rel=0.01)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
tests = [
|
|
test_f_ratio,
|
|
test_abs_pressure,
|
|
test_valve_model_roundtrip,
|
|
test_non_monotonic_raises,
|
|
test_identify_end_to_end,
|
|
]
|
|
for test in tests:
|
|
test()
|
|
print(f"PASS {test.__name__}")
|
|
print("全部通过")
|