도면 출력 추가: 수리 종단면도, 관망도(색인도·부분도·수압표), 범례·색상 선택, DXF(AutoCAD 2018)

- 데모 v2 이후 개발 PC의 변경을 옮김 (이 저장소의 이름 변경·README 정리는 유지)
- 서버 기본 DEM, DEM 등고선, 속성 테이블 범위 선택, WNTR 교차 검증 시험 포함
- 새 라이브러리: ezdxf(Python), maplibre-contour(npm)
- 변경 내역: docs/SPipe_패치내역_도면출력_2026-11-06.txt

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

backend/tests/test_drawing_dxf.py

@@ -0,0 +1,35 @@
+"""/api/drawing/dxf: browser R12 DXF (Unicode escapes) -> Korean code page (ANSI_949)."""
+import os
+import tempfile
+
+import pyogrio
+from fastapi.testclient import TestClient
+
+from backend.main import app, dxf_to_cp949
+
+DXF = "\r\n".join([
+    "0", "SECTION", "2", "HEADER", "9", "$ACADVER", "1", "AC1009", "0", "ENDSEC",
+    "0", "SECTION", "2", "ENTITIES",
+    "0", "TEXT", "8", "TITLE", "10", "1", "20", "2", "30", "0", "40", "2.5", "1", r"\U+C218\U+B9AC \U+C885\U+B2E8\U+BA74\U+B3C4 \U+1F600",
+    "0", "ENDSEC", "0", "EOF", "",
+])
+
+
+def test_conversion_reads_back_as_korean():
+    data = dxf_to_cp949(DXF)
+    assert b"$DWGCODEPAGE\r\n3\r\nANSI_949" in data
+    with tempfile.TemporaryDirectory() as tmp:
+        p = os.path.join(tmp, "t.dxf")
+        open(p, "wb").write(data)
+        meta, _, _, fields = pyogrio.raw.read(p)
+        text = fields[list(meta["fields"]).index("Text")][0]
+    assert text.startswith("수리 종단면도")
+    assert r"\U+1F600" in data.decode("cp949")  # outside CP949: kept escaped
+
+
+def test_endpoint():
+    with TestClient(app) as c:
+        r = c.post("/api/drawing/dxf", json={"dxf": DXF, "name": "수리 종단면도"})
+        assert r.status_code == 200 and "attachment" in r.headers["content-disposition"]
+        assert "수리".encode("cp949") in r.content
+        assert c.post("/api/drawing/dxf", json={"dxf": "nope"}).status_code == 400

backend/tests/test_drawing_ezdxf.py

@@ -0,0 +1,56 @@
+"""Drawing sheet -> DXF R2018 (ezdxf): Unicode text, layers, lineweights, hatches."""
+import io
+import os
+import tempfile
+
+import ezdxf
+import pyogrio
+import pytest
+from fastapi.testclient import TestClient
+
+from backend.batch import BatchError
+from backend.drawing_dxf import drawing_to_dxf
+from backend.main import app
+
+DRAWING = {
+    "width": 420, "height": 297, "title": "t",
+    "items": [
+        {"kind": "line", "layer": "FRAME", "pts": [[0, 0], [420, 0], [420, 297], [0, 297]], "closed": True, "width": 0.7},
+        {"kind": "line", "layer": "HGL_1", "pts": [[50, 200], [300, 150]], "width": 0.45, "color": "#1d4ed8", "dash": "dashed"},
+        {"kind": "fill", "layer": "LEGEND", "pts": [[10, 10], [30, 10], [30, 20], [10, 20]], "fill": "#ffffff", "opacity": 0.9},
+        {"kind": "circle", "layer": "NODE", "x": 100, "y": 100, "r": 1.2, "fill": "#dc2626", "color": "#dc2626"},
+        {"kind": "text", "layer": "TITLE", "x": 210, "y": 280, "text": "송수관로 수리 종단면도 ○ ‰ m³", "size": 7, "anchor": "middle", "vAnchor": "middle", "bold": True},
+        {"kind": "text", "layer": "TABLE", "x": 60, "y": 60, "text": "J-101", "size": 2, "rotate": 90, "anchor": "middle", "vAnchor": "middle"},
+    ],
+}
+
+
+def test_r2018_with_unicode_layers_and_styles():
+    data = drawing_to_dxf(DRAWING)
+    doc = ezdxf.read(io.StringIO(data.decode("utf-8")))
+    assert doc.dxfversion == "AC1032" and doc.units == ezdxf.units.MM
+    msp = doc.modelspace()
+    texts = [e.dxf.text for e in msp.query("TEXT")]
+    assert "송수관로 수리 종단면도 ○ ‰ m³" in texts
+    hgl = msp.query('LWPOLYLINE[layer=="HGL_1"]')[0]
+    assert hgl.dxf.linetype == "DASHED" and hgl.dxf.lineweight == 40 and hgl.rgb == (29, 78, 216)
+    assert len(msp.query("HATCH")) == 1 and len(msp.query("CIRCLE")) == 1  # white mask box dropped
+    assert {"FRAME", "HGL_1", "NODE", "TITLE", "TABLE"} <= {l.dxf.name for l in doc.layers}
+    rotated = [e for e in msp.query("TEXT") if e.dxf.text == "J-101"][0]
+    assert rotated.dxf.rotation == 90
+    # GDAL (what GIS tools use) reads the Korean text back too.
+    with tempfile.TemporaryDirectory() as tmp:
+        p = os.path.join(tmp, "d.dxf")
+        open(p, "wb").write(data)
+        meta, _, _, fields = pyogrio.raw.read(p)
+        gtexts = [t for t in fields[list(meta["fields"]).index("Text")] if t]
+    assert any("송수관로" in t for t in gtexts)
+
+
+def test_errors_and_endpoint():
+    with pytest.raises(BatchError):
+        drawing_to_dxf({"items": []})
+    with TestClient(app) as c:
+        r = c.post("/api/drawing/dxf", json={"drawing": DRAWING, "name": "종단면도"})
+        assert r.status_code == 200 and b"AC1032" in r.content
+        assert "송수관로".encode("utf-8") in r.content

backend/tests/test_wntr_crosscheck.py

@@ -0,0 +1,78 @@
+"""Cross-check SPipe's server analyses against USEPA WNTR (an independent EPANET
+front end). Skipped when wntr is not installed (it is a dev-only dependency).
+
+WNTR works in SI (m, m³/s); Net1/Net3 are US units (psi, GPM), so values are
+converted before comparing. WNTR bundles its own EPANET build, so small
+numerical differences are allowed.
+"""
+import os
+import tempfile
+from pathlib import Path
+
+import numpy as np
+import pytest
+
+from backend.batch import Network
+from backend.epanet import EN
+from backend.jobs import job_criticality, job_fireflow
+
+wntr = pytest.importorskip("wntr")
+
+PUBLIC = Path(__file__).resolve().parents[2] / "public"
+PSI_TO_M = 0.3048 / 0.4333  # EPANET's own psi <-> ft of water factor
+GPM_TO_M3S = 6.30902e-5
+
+
+def wntr_pressure(wn) -> "object":
+    with tempfile.TemporaryDirectory() as tmp:
+        res = wntr.sim.EpanetSimulator(wn).run_sim(file_prefix=os.path.join(tmp, "w"))
+    return res.node["pressure"]
+
+
+def spipe_pressure(inp: bytes) -> dict:
+    with Network(inp) as n:
+        js = n.junction_indices()
+        r = n.simulate(node_props=(EN.PRESSURE,), nodes=js)
+        return {n.node_ids[j - 1]: r.nodes[EN.PRESSURE][:, k] * PSI_TO_M for k, j in enumerate(js)}, r.times
+
+
+@pytest.mark.parametrize("name", ["Net1.inp", "Net3.inp"])
+def test_baseline_pressures_match(name):
+    path = PUBLIC / name
+    ours, times = spipe_pressure(path.read_bytes())
+    p = wntr_pressure(wntr.network.WaterNetworkModel(str(path)))
+    diffs = []
+    for nid, series in ours.items():
+        w = p[nid].reindex(times).to_numpy()
+        diffs.append(np.nanmax(np.abs(series - w)))
+    assert max(diffs) < 0.01, f"max pressure difference {max(diffs):.3f} m"
+
+
+def test_pipe_criticality_matches_closing_pipes_in_wntr():
+    path = PUBLIC / "Net1.inp"
+    ours = {e["id"]: e for e in job_criticality(path.read_bytes(), {"threshold": 40})["links"]}
+    checked = 0
+    for pid, e in ours.items():
+        if e.get("disconnected") or e.get("minPressure") is None:
+            continue  # isolated nodes: both engines report meaningless negative heads
+        wn = wntr.network.WaterNetworkModel(str(path))
+        wn.get_link(pid).initial_status = wntr.network.LinkStatus.Closed
+        p = wntr_pressure(wn)
+        junctions = [n for n, _ in wn.junctions()]
+        wmin = float(p[junctions].min().min())
+        assert abs(e["minPressure"] * PSI_TO_M - wmin) < 0.01, f"pipe {pid}: {e['minPressure'] * PSI_TO_M:.2f} vs {wmin:.2f} m"
+        checked += 1
+    assert checked >= 10
+
+
+def test_fire_flow_pressure_drop_matches_extra_demand_in_wntr():
+    path = PUBLIC / "Net1.inp"
+    flow_gpm = 500.0
+    r = job_fireflow(path.read_bytes(), {"nodes": ["22"], "flow": flow_gpm, "start": 0, "end": 24})
+    ours = r["hydrants"][0]["firePressure"] * PSI_TO_M
+    wn = wntr.network.WaterNetworkModel(str(path))
+    # A demand without a pattern would get Net1's default pattern (up to 1.6x); fire flow is constant.
+    wn.add_pattern("fire_const", [1.0])
+    wn.get_node("22").add_demand(flow_gpm * GPM_TO_M3S, "fire_const", "fire")
+    p = wntr_pressure(wn)
+    assert abs(ours - float(p["22"].min())) < 0.01