flow_control/test_valve_model.py

"""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_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

    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


if __name__ == "__main__":
    tests = [
        test_f_ratio,
        test_abs_pressure,
        test_valve_model_roundtrip,
        test_non_monotonic_raises,
        test_identify_end_to_end,
        test_merge_multiple_csvs,
        test_direction_tagging,
        test_resolve_csv_paths_glob,
    ]
    for test in tests:
        test()
        print(f"PASS {test.__name__}")
    print("全部通过")