"""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 _write_sweep_csv(path, *, scale=1.0, visit_openings=(100.0, 50.0, 0.0)): """构造合成扫点 CSV：每个 step 30 个采样点，全程稳态。 ``visit_openings`` 为开度的访问顺序（模拟 open_loop.py 的打乱顺序）； A_eff 真值 = scale * (1000 - x) / 400，Q_ss 按阻塞流（F=1）反推。 """ rows = [] for step, opening in enumerate(visit_openings): x = 1000.0 - opening / 100.0 * 200.0 # 与 flow_control 的映射一致 a_eff = scale * (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": opening, "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 path.open("w", newline="", encoding="utf-8-sig") as f: writer = csv.DictWriter(f, fieldnames=list(rows[0].keys())) writer.writeheader() writer.writerows(rows) 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_area_below_measured_minimum_returns_fully_closed_stroke(): model = ValveModel( [(800.0, 0.7358347), (980.0, 0.002094)], motor_open=800.0, motor_closed=1000.0, ) assert model.stroke_from_area(0.002094) == 980.0 assert model.stroke_from_area(0.002093) == 1000.0 assert model.stroke_from_area(0.0) == 1000.0 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_monotonic_check_can_be_disabled(): import identify_valve p1 = 400.0 denom = p1 + P_ATM # p2=0 时为阻塞流，F=1。 steady = [ (800.0, 0.5 * denom, p1, 0.0, None), (900.0, 0.1 * denom, p1, 0.0, None), (1000.0, 0.2 * denom, p1, 0.0, None), ] checked, dropped, _, _ = identify_valve.compute_area_table( steady, check_monotonic=True, ) unchecked, unchecked_dropped, _, _ = identify_valve.compute_area_table( steady, check_monotonic=False, ) assert len(checked) == 2 assert len(dropped) == 1 assert len(unchecked) == 3 assert not unchecked_dropped model = ValveModel( unchecked, motor_open=800.0, motor_closed=1000.0, enforce_monotonic=False, ) assert not model.is_monotonic # A_eff=0.15 有两个交点，选择更接近关闭端的 x=950。 assert _close(model.stroke_from_area(0.15), 950.0) with tempfile.TemporaryDirectory() as tmp: path = Path(tmp) / "nonmonotonic.json" model.save(path) loaded = ValveModel.load(path, 800.0, 1000.0) assert not loaded.enforce_monotonic assert _close(loaded.stroke_from_area(0.15), 950.0) def test_identify_end_to_end(): import identify_valve with tempfile.TemporaryDirectory() as tmp: csv_path = Path(tmp) / "sweep.csv" out_path = Path(tmp) / "model.json" _write_sweep_csv(csv_path) model, steady, table, dropped, stats, area_points = 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) assert len(area_points) == 3 loaded = ValveModel.load(str(out_path), 800.0, 1000.0) assert _close(loaded.area_from_stroke(900.0), 0.25, rel=0.01) def test_merge_multiple_csvs(): import identify_valve with tempfile.TemporaryDirectory() as tmp: csv_a = Path(tmp) / "run_a.csv" csv_b = Path(tmp) / "run_b.csv" _write_sweep_csv(csv_a, scale=1.0) _write_sweep_csv(csv_b, scale=1.1) model, steady, table, dropped, stats, _ = identify_valve.identify( [csv_a, csv_b], tail=20 ) assert len(steady) == 6, steady assert stats["choked"] == 6 assert not dropped # 同一行程的两个点先平均：A_eff = 1.05 * 真值 for x, a in table: assert _close(a, 1.05 * (1000.0 - x) / 400.0, rel=0.01), (x, a) def test_direction_tagging(): import identify_valve with tempfile.TemporaryDirectory() as tmp: csv_path = Path(tmp) / "shuffled.csv" _write_sweep_csv(csv_path, visit_openings=(0.0, 100.0, 50.0)) points = identify_valve.load_steady_points(csv_path, tail=20) directions = [p[4] for p in points] assert directions == [None, "up", "down"], directions def test_resolve_csv_paths_glob(): import identify_valve with tempfile.TemporaryDirectory() as tmp: base = Path(tmp) (base / "b.csv").write_text("", encoding="utf-8") (base / "a.csv").write_text("", encoding="utf-8") (base / "c.txt").write_text("", encoding="utf-8") # glob 展开并按名称排序，只匹配 CSV paths = identify_valve.resolve_csv_paths([str(base / "*.csv")]) assert [Path(p).name for p in paths] == ["a.csv", "b.csv"], paths # 显式路径 + glob 混用时去重 paths = identify_valve.resolve_csv_paths( [str(base / "*.csv"), str(base / "a.csv")] ) assert [Path(p).name for p in paths] == ["a.csv", "b.csv"], paths def test_open_loop_appends_fine_closed_end_strokes(): import config import open_loop points = open_loop.generate_scan_points(0.0, 100.0, 50.0, seed=42) fine_strokes = open_loop.generate_fine_strokes( config.OPEN_LOOP_FINE_STROKE_START, config.MOTOR_CLOSED_POSITION, config.OPEN_LOOP_FINE_STROKE_STEP, ) assert open_loop.generate_fine_strokes(975.0, 1000.0, 5.0) == [ 975.0, 980.0, 985.0, 990.0, 995.0, 1000.0, ] fine_points = points[-len(fine_strokes):] assert [position for _, position in fine_points] == fine_strokes for opening, position in fine_points: assert _close( opening, open_loop.motor_position_to_opening( position, config.MOTOR_OPEN_POSITION, config.MOTOR_CLOSED_POSITION, ), ) if __name__ == "__main__": tests = [ test_f_ratio, test_abs_pressure, test_valve_model_roundtrip, test_area_below_measured_minimum_returns_fully_closed_stroke, test_non_monotonic_raises, test_monotonic_check_can_be_disabled, test_identify_end_to_end, test_merge_multiple_csvs, test_direction_tagging, test_resolve_csv_paths_glob, test_open_loop_appends_fine_closed_end_strokes, ] for test in tests: test() print(f"PASS {test.__name__}") print("全部通过")