98 lines
4.7 KiB
Python
98 lines
4.7 KiB
Python
from datetime import datetime, timezone
|
|
from types import SimpleNamespace
|
|
|
|
import pytest
|
|
|
|
from scripts.benchmark.calibration import assess, configure_inp, pressure_preview, select_calibrated_area
|
|
|
|
|
|
START = datetime(2026, 10, 8, 12, tzinfo=timezone.utc)
|
|
PIPE = SimpleNamespace(
|
|
pipe_id="test-pipe",
|
|
pressure_bindings=[SimpleNamespace(device_id="P1", node_id="N1"),
|
|
SimpleNamespace(device_id="P2", node_id="N2")],
|
|
flow_bindings=[SimpleNamespace(device_id="F1", link_id="L1")],
|
|
)
|
|
RULE = SimpleNamespace(
|
|
primary_pressure_drop_percent=30, corroborating_pressure_drop_percent=10,
|
|
flow_change_percent=20, consecutive_samples=2, relative_change_epsilon=1e-6,
|
|
)
|
|
|
|
|
|
def result(p1=(30, 30, 30, 30, 30), p2=(30, 30, 30, 30, 30), flow=(10, 10, 10, 10, 10)):
|
|
return {"status": "completed", "warnings": [],
|
|
"units": {"flow": "LPS", "pressure": "MTR", "quality": "NONE"},
|
|
"times": {"report_start": 0, "report_step": 900, "sim_duration": 3600, "num_periods": 5},
|
|
"nodes": {"N1": [{"pressure": v} for v in p1], "N2": [{"pressure": v} for v in p2]},
|
|
"links": {"L1": [{"flow": v} for v in flow]}}
|
|
|
|
|
|
def check(burst, normal=None):
|
|
return assess(PIPE, normal or result(), burst, RULE, duration=3600, step=900)
|
|
|
|
|
|
def test_file_configuration_preserves_source_and_matches_burst_service():
|
|
source = "[OPTIONS]\nDEMAND MODEL DDA\nREQUIRED PRESSURE 20\n[TIMES]\nDURATION 0\nPATTERN START 0:00\n[REPORT]\nNODES ALL\nLINKS ALL\n[EMITTERS]\n[END]\n"
|
|
configured = configure_inp(source, START, 18000, emitter=("N1", 100), pda_options={"HEADERROR": "0.0001"})
|
|
assert "DEMAND MODEL DDA" in source
|
|
assert "DEMAND MODEL DDA" not in configured
|
|
assert "DEMAND MODEL PDA" in configured
|
|
assert "REQUIRED PRESSURE 10.0000" in configured
|
|
assert "HEADERROR 0.0001" in configured
|
|
assert "DURATION 5:00:00" in configured
|
|
assert "PATTERN START 20:00:00" in configured
|
|
coefficient = float(configured.split("[EMITTERS]\nN1 ")[1].splitlines()[0])
|
|
assert coefficient == pytest.approx(28.7767267075)
|
|
assert "NODES NONE" in configured and "LINKS NONE" in configured
|
|
|
|
|
|
def test_requires_two_consecutive_primary_samples():
|
|
assert not check(result(p1=(20, 30, 20, 30, 20), p2=(25,) * 5))["passed"]
|
|
assert check(result(p1=(20, 20, 30, 30, 30), p2=(25,) * 5))["passed"]
|
|
|
|
|
|
def test_corroboration_must_be_concurrent_and_a_different_pressure_device():
|
|
assert not check(result(p1=(20, 20, 30, 30, 30)))["passed"]
|
|
assert not check(result(p1=(20, 20, 30, 30, 30), p2=(30, 30, 25, 25, 30)))["passed"]
|
|
assert check(result(p1=(20,) * 5, flow=(12,) * 5))["passed"]
|
|
|
|
|
|
def test_pressure_rise_and_zero_flow_baseline_do_not_supply_false_evidence():
|
|
assert not check(result(p1=(40,) * 5, p2=(25,) * 5, flow=(100,) * 5))["passed"]
|
|
assert not check(result(p1=(20,) * 5, flow=(10,) * 5), normal=result(flow=(0,) * 5))["passed"]
|
|
|
|
|
|
def test_negative_pressure_or_bad_normal_pressure_invalidates_candidate():
|
|
assert "negative_monitor_pressure:P2" in check(result(p1=(20,) * 5, p2=(-1,) * 5))["blockers"]
|
|
assert "nonpositive_normal_pressure:P1" in check(result(p1=(0,) * 5), normal=result(p1=(0,) * 5))["blockers"]
|
|
|
|
|
|
@pytest.mark.parametrize("fault", ["warning", "missing_period", "unit"])
|
|
def test_solver_warning_incomplete_time_grid_and_wrong_unit_fail(fault):
|
|
burst = result(p1=(20,) * 5, p2=(25,) * 5)
|
|
if fault == "warning":
|
|
burst["warnings"] = ["WARNING: System unbalanced"]
|
|
elif fault == "missing_period":
|
|
burst["times"]["num_periods"] = 4
|
|
else:
|
|
burst["units"]["flow"] = "CMH"
|
|
assert not check(burst)["passed"]
|
|
|
|
|
|
def test_actual_replay_rule_uses_warmup_and_reports_first_signal_time():
|
|
history = {"N1": [{"pressure": 30}] * 4, "N2": [{"pressure": 30}] * 4}
|
|
normal = pressure_preview(PIPE, START, 3600, 900, history, result()["nodes"], RULE)
|
|
event = pressure_preview(PIPE, START, 3600, 900, history, result(p1=(20,) * 5)["nodes"], RULE)
|
|
assert not normal["raised"]
|
|
assert event["raised"]
|
|
assert event["first_signal_at"] == "2026-10-08T12:15:00+00:00"
|
|
|
|
|
|
def test_valid_burst_with_a_weak_signal_is_retained_separately_from_invalid_physics():
|
|
pipe = SimpleNamespace(candidate_orifice_areas_cm2={"light": 1, "medium": 2, "severe": 4})
|
|
weak = [{"blockers": ["event_rule_did_not_trigger", "insufficient_consecutive_pressure_and_corroboration"]}]
|
|
assert select_calibrated_area(pipe, [], weak, True)["status"] == "weak_signal_candidate"
|
|
assert select_calibrated_area(pipe, [], weak, False)["status"] == "unsuitable"
|
|
assert select_calibrated_area(pipe, [], [{"blockers": ["burst_solver_warning"]}], True)["status"] == "unsuitable"
|
|
assert select_calibrated_area(pipe, ["severe", "medium"], weak, True)["area_cm2"] == 2
|