bjkim1
21ebb16fb7
- SPipe_server_2026-10-30.zip 내용을 풀어서 등록 (압축 파일은 올리지 않음) - README에 데모 v2 서버 업데이트 순서 추가, 저장소에 없는 문서 목차 정리 - 패치내역을 docs/로 이동, _worklog.md 추가 - 임시 파일(backend/temp.*)과 API 키가 든 파일은 제외
88 lines
3.5 KiB
Python
88 lines
3.5 KiB
Python
"""backend/batch.py against the real EPANET DLL."""
|
|
from pathlib import Path
|
|
|
|
import numpy as np
|
|
|
|
from backend.batch import Network, parse_time
|
|
from backend.epanet import EN
|
|
|
|
NET1 = (Path(__file__).resolve().parents[2] / "public" / "Net1.inp").read_bytes()
|
|
|
|
|
|
def test_open_lookup_and_baseline():
|
|
with Network(NET1) as net:
|
|
assert net.node_count == 11 and net.link_count == 13
|
|
assert net.node_ids[net.node_index("22") - 1] == "22"
|
|
res = net.simulate(node_props=(EN.PRESSURE,), link_props=(EN.FLOW,))
|
|
assert len(res.times) == 25 and res.times[0] == 0 and res.times[-1] == 24 * 3600
|
|
assert res.nodes[EN.PRESSURE].shape == (25, 11)
|
|
assert res.links[EN.FLOW].shape == (25, 13)
|
|
|
|
|
|
def test_windowed_extra_demand_only_acts_in_window():
|
|
with Network(NET1) as net:
|
|
j = net.node_index("22")
|
|
base = net.simulate(nodes=[j]).nodes[EN.PRESSURE][:, 0]
|
|
net.window_pattern("FIREWIN", 6 * 3600, 9 * 3600)
|
|
cat = net.add_demand(j, 500.0, "FIREWIN", "fire")
|
|
fire = net.simulate(nodes=[j]).nodes[EN.PRESSURE][:, 0]
|
|
hours = np.arange(25)
|
|
inside = (hours >= 6) & (hours < 9)
|
|
assert np.all(fire[inside] < base[inside] - 1.0)
|
|
# Tank levels carry the draw-down a little past the window; before it, nothing changes.
|
|
assert np.allclose(fire[hours < 6], base[hours < 6], atol=1e-6)
|
|
net.delete_demand(j, cat)
|
|
again = net.simulate(nodes=[j]).nodes[EN.PRESSURE][:, 0]
|
|
assert np.allclose(again, base, atol=1e-6)
|
|
|
|
|
|
def test_timer_control_closes_pipe_in_window():
|
|
with Network(NET1) as net:
|
|
p = net.link_index("12")
|
|
c1 = net.add_timer_control(p, 0, 3 * 3600)
|
|
c2 = net.add_timer_control(p, 1, 5 * 3600)
|
|
flow = net.simulate(node_props=(), link_props=(EN.FLOW,), links=[p]).links[EN.FLOW][:, 0]
|
|
assert abs(flow[3]) < 1e-6 and abs(flow[4]) < 1e-6
|
|
assert abs(flow[2]) > 1e-3 and abs(flow[6]) > 1e-3
|
|
net.delete_control(c2)
|
|
net.delete_control(c1)
|
|
|
|
|
|
def test_parse_time():
|
|
assert parse_time("6:30") == 6 * 3600 + 1800
|
|
assert parse_time(2.5) == 9000
|
|
assert parse_time(None, 7) == 7
|
|
|
|
|
|
def test_fireflow_job_and_available_flow():
|
|
from backend.jobs import job_fireflow
|
|
|
|
r = job_fireflow(NET1, {"nodes": ["22"], "flow": 500, "start": 6, "end": 9, "residual": 20, "threshold": 50})
|
|
assert r["windowLabel"] == "6:00~9:00"
|
|
h = r["hydrants"][0]
|
|
assert h["firePressure"] < h["basePressure"] - 1
|
|
assert r["nodes"][0]["diff"] <= r["nodes"][-1]["diff"] # most affected first
|
|
# available flow keeps the hydrant at (just above) the residual pressure
|
|
q = h["available"]
|
|
assert q > 0
|
|
with Network(NET1) as net:
|
|
j = net.node_index("22")
|
|
net.window_pattern("W", 6 * 3600, 9 * 3600)
|
|
c = net.add_demand(j, q, "W", "x")
|
|
p = net.simulate(nodes=[j]).nodes[EN.PRESSURE][6:9, 0].min()
|
|
assert p >= 20 - 1e-6
|
|
net.delete_demand(j, c)
|
|
net.add_demand(j, q * 1.02, "W", "x")
|
|
assert net.simulate(nodes=[j]).nodes[EN.PRESSURE][6:9, 0].min() < 20 + 0.5
|
|
|
|
|
|
def test_fireflow_rejects_bad_input():
|
|
import pytest
|
|
from backend.batch import BatchError
|
|
from backend.jobs import job_fireflow
|
|
|
|
with pytest.raises(BatchError):
|
|
job_fireflow(NET1, {"nodes": [], "flow": 10})
|
|
with pytest.raises(BatchError):
|
|
job_fireflow(NET1, {"nodes": ["NOPE"], "flow": 10})
|