bjkim1
21ebb16fb7
- SPipe_server_2026-10-30.zip 내용을 풀어서 등록 (압축 파일은 올리지 않음) - README에 데모 v2 서버 업데이트 순서 추가, 저장소에 없는 문서 목차 정리 - 패치내역을 docs/로 이동, _worklog.md 추가 - 임시 파일(backend/temp.*)과 API 키가 든 파일은 제외
61 lines
2.7 KiB
Python
61 lines
2.7 KiB
Python
"""Criticality, service area (trace), VSP and energy jobs against the real DLL."""
|
|
import re
|
|
from pathlib import Path
|
|
|
|
import pytest
|
|
|
|
from backend.batch import BatchError
|
|
from backend.jobs import job_criticality, job_energy, job_servicearea, job_vsp
|
|
|
|
PUBLIC = Path(__file__).resolve().parents[2] / "public"
|
|
NET1 = (PUBLIC / "Net1.inp").read_bytes()
|
|
NET3 = (PUBLIC / "Net3.inp").read_bytes()
|
|
|
|
|
|
def test_criticality_ranks_the_supply_main_first():
|
|
r = job_criticality(NET1, {"threshold": 40})
|
|
assert len(r["links"]) == 12 # pipes only, the pump is skipped
|
|
top = r["links"][0]
|
|
# Pipe 10 is the only path from the source: closing it cuts off every junction.
|
|
assert top["id"] == "10" and top["affectedNodes"] == 8 and top["disconnected"]
|
|
loops = [e for e in r["links"] if e["id"] in ("111", "112", "121", "122")]
|
|
assert all(e["affectedNodes"] == 0 for e in loops)
|
|
|
|
|
|
def test_criticality_subset_and_limits():
|
|
r = job_criticality(NET1, {"threshold": 40, "links": ["111"]})
|
|
assert [e["id"] for e in r["links"]] == ["111"]
|
|
with pytest.raises(BatchError):
|
|
job_criticality(NET1, {"links": ["nope"]})
|
|
|
|
|
|
def test_service_area_shares_sum_to_at_most_100():
|
|
r = job_servicearea(NET3, {})
|
|
assert r["sources"] == ["River", "Lake"]
|
|
assert r["unreached"] == 0 and r["notes"] # 24 h run extended to 72 h
|
|
assert max(sum(v["shares"].values()) for v in r["nodes"].values()) <= 100.5
|
|
assert r["nodes"]["River"]["source"] == "River" and r["nodes"]["Lake"]["source"] == "Lake"
|
|
|
|
|
|
def test_vsp_hits_the_target_or_reports_the_bound():
|
|
r = job_vsp(NET1, {"pump": "9", "node": "31", "target": 125, "hours": [0, 18], "maxSpeed": 1.5})
|
|
by_hour = {row["hour"]: row for row in r["rows"]}
|
|
assert by_hour[0]["status"] == "max" and by_hour[0]["pressure"] < 125
|
|
assert by_hour[18]["status"] == "ok" and abs(by_hour[18]["pressure"] - 125) < 0.05
|
|
assert 1.0 < by_hour[18]["speed"] < 1.5
|
|
with pytest.raises(BatchError):
|
|
job_vsp(NET1, {"pump": "10", "node": "31", "target": 100}) # a pipe
|
|
|
|
|
|
def test_energy_report_matches_epanet_energy_section():
|
|
r = job_energy(NET1, {})
|
|
(p,) = r["pumps"]
|
|
# EPANET's own Net1 energy report: 57.71 % utilisation, 75 % efficiency, ~96 kW average.
|
|
assert p["utilization"] == pytest.approx(57.71, abs=0.05)
|
|
assert p["efficiency"] == pytest.approx(75.0)
|
|
assert p["avgKw"] == pytest.approx(96.25, abs=0.5)
|
|
assert p["kwhPerM3"] > 0 and r["totalCost"] == 0.0 # Net1 has no price
|
|
priced = re.sub(rb"Global Price\s+0\.0", b"Global Price 0.1", NET1)
|
|
assert priced != NET1
|
|
assert job_energy(priced, {})["pumps"][0]["cost"] == pytest.approx(p["kwh"] * 0.1, rel=1e-3)
|