159 lines
7.2 KiB
Python
159 lines
7.2 KiB
Python
# -*- coding: utf-8 -*-
|
|
"""510-1 [자동 제안] = B05 배수유역 영역 후보 → 사용자가 골라 넣음(요구 320-2).
|
|
|
|
못박는 것
|
|
- 후보 = 지금 노선 × 세류선 교차(5 m 솎기 · 정수 m · 라벨) + 간격 보충(흐름 세기 · 분석 있을 때만).
|
|
- 상태 = 라벨 짝 관 · 교차 5 m 안 관 = held · 지움 표지 짝 = dismissed(라벨 = 그 표지) · 나머지 new.
|
|
- 고른 교차 후보 = 라벨 실은 관 → 계산 왕복(carry) · 저장(keep_labels) 지나도 라벨 · 지운 관을 다시
|
|
넣으면 지움 표지 걷음.
|
|
- 넣은 라벨 관 = 노선 바뀐 [저장] 따라감 · 새 관 안 더함(겹침 없음).
|
|
- 화면: 아무것도 미리 안 고름 · [넣기] = 고른 것만 · held 는 목록 밖 · 관 속성 보관소가 라벨 왕복.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import json
|
|
import shutil
|
|
import subprocess
|
|
import sys
|
|
from pathlib import Path
|
|
|
|
import numpy as np
|
|
import pytest
|
|
|
|
ROOT = Path(__file__).resolve().parents[2]
|
|
if str(ROOT) not in sys.path:
|
|
sys.path.insert(0, str(ROOT))
|
|
|
|
import B05_Profile.B05_Profile_Route_Save as save # noqa: E402
|
|
from B04_PreProcess.B04_PreProcess_Engine_Watershed_Analyze import _base_pipes # noqa: E402
|
|
from B05_Profile.B05_Profile_Route_Save_Parts import vertices_of # noqa: E402
|
|
from common_util.common_util_drainage_pipes import ( # noqa: E402
|
|
PipePoint,
|
|
carry_facility_attributes,
|
|
parse_pipe_points,
|
|
)
|
|
from common_util.common_util_stream_crossing import keep_labels, propose_pipes # noqa: E402
|
|
|
|
HERE = Path(__file__).resolve().parent
|
|
TSC = ROOT / "config" / "node_modules" / "typescript"
|
|
|
|
|
|
def _stream(ufid: str, x: float) -> dict:
|
|
return {
|
|
"type": "Feature",
|
|
"properties": {"UFID": ufid},
|
|
"geometry": {"type": "LineString", "coordinates": [[x, -50.0], [x, 120.0]]},
|
|
}
|
|
|
|
|
|
LINE = [[0.0, 0.0], [300.0, 0.0], [600.0, 0.0]]
|
|
STREAMS = [_stream("S1", 50.4), _stream("S2", 150.0), _stream("S3", 250.0), _stream("S4", 320.0)]
|
|
|
|
|
|
def _case():
|
|
vertices = vertices_of(LINE)
|
|
crossings = _base_pipes(vertices, STREAMS)
|
|
s2 = {"stream": "S2", "order": 1, "x": 150.0, "y": 0.0}
|
|
pipes = [
|
|
PipePoint(chainage_m=150.0, source="user", stream_label=s2, id="p15000"),
|
|
PipePoint(chainage_m=252.0, source="user", id="p25200"), # 라벨 없음 · S3 교차 5 m 안
|
|
]
|
|
dismissed = [{"stream": "S1", "order": 1, "x": 50.4, "y": 0.0}]
|
|
return vertices, crossings, pipes, dismissed
|
|
|
|
|
|
def test_후보_교차_상태_라벨():
|
|
vertices, crossings, pipes, dismissed = _case()
|
|
found = propose_pipes(vertices, crossings, pipes, dismissed)
|
|
assert [(f["chainage_m"], f["kind"], f["state"]) for f in found] == [
|
|
(50.0, "stream", "dismissed"), # 정수 m
|
|
(150.0, "stream", "held"),
|
|
(250.0, "stream", "held"),
|
|
(320.0, "stream", "new"),
|
|
]
|
|
assert found[0]["stream_label"] == dismissed[0] # 지운 관 = 그 표지 그대로(다시 넣으면 걷음)
|
|
assert found[3]["stream_label"] == {"stream": "S4", "order": 1, "x": 320.0, "y": 0.0}
|
|
|
|
|
|
def test_간격_보충_분석_있을_때만():
|
|
vertices, crossings, pipes, dismissed = _case()
|
|
curve = np.ones(601)
|
|
found = propose_pipes(vertices, crossings, pipes, dismissed, curve)
|
|
spacing = [f for f in found if f["kind"] == "spacing"]
|
|
assert spacing and all(f["state"] == "new" and "stream_label" not in f for f in spacing)
|
|
# 지금 관 + 교차 후보(새 · 지운 관) 사이 최대 간격(100 m) 넘는 구간만 · 320 뒤 280 m ·
|
|
# 지운 교차(50) 곁에 간격 보충이 붙지 않음(화면 실측 = 지운 385 옆 378)
|
|
taken = [50.0, 150.0, 252.0, 320.0]
|
|
assert all(min(abs(f["chainage_m"] - t) for t in taken) > 5.0 for f in spacing)
|
|
assert any(f["chainage_m"] > 320.0 for f in spacing)
|
|
assert not [f for f in propose_pipes(vertices, crossings, pipes, dismissed) if f["kind"] == "spacing"]
|
|
|
|
|
|
def test_고른_라벨_관_계산_왕복_저장_지나도_라벨():
|
|
label = {"stream": "S4", "order": 1, "x": 320.0, "y": 0.0}
|
|
request = parse_pipe_points([{"chainage_m": 320.0, "source": "user", "stream_label": label}])
|
|
rebuilt = carry_facility_attributes([PipePoint(chainage_m=320.0, source="user")], request)
|
|
assert rebuilt[0].stream_label == label # 고치기 전 = 계산 왕복에서 사라짐
|
|
dismissed: list = []
|
|
keep_labels(rebuilt, [], dismissed)
|
|
assert rebuilt[0].stream_label == label and dismissed == []
|
|
|
|
|
|
def test_지운_관_다시_넣으면_지움_표지_걷음():
|
|
mark = {"stream": "S1", "order": 1, "x": 50.4, "y": 0.0}
|
|
other = {"stream": "S9", "order": 1, "x": 9.0, "y": 0.0}
|
|
dismissed = [mark, other]
|
|
keep_labels([PipePoint(chainage_m=50.0, stream_label=dict(mark))], [], dismissed)
|
|
assert dismissed == [other]
|
|
|
|
|
|
# 507-2 시험과 같은 꼴 — 앞 노선을 북으로 옮겨 세류선 교차가 30 m 옮음
|
|
OLD = [[0.0, 0.0], [100.0, 0.0], [200.0, 0.0], [300.0, 0.0], [400.0, 0.0]]
|
|
NEW1 = [[0.0, 0.0], [100.0, 0.0], [200.0, 60.0], [300.0, 0.0], [400.0, 0.0]]
|
|
NEW2 = [[0.0, 0.0], [100.0, 0.0], [200.0, 90.0], [300.0, 0.0], [400.0, 0.0]]
|
|
ONE = [_stream("S1", 150.0)]
|
|
|
|
|
|
def _follow(old, old_xy, new_xy):
|
|
points = [{"x": x, "y": y} for x, y in new_xy]
|
|
pairs, crossing, _ = save.followed_pipes(old, old_xy, points, ONE, [])
|
|
return [pipe for _, pipe in pairs], crossing
|
|
|
|
|
|
def test_넣은_라벨_관_노선_바꿈_따라감_겹침_없음():
|
|
[cand] = propose_pipes(vertices_of(OLD), _base_pipes(vertices_of(OLD), ONE), [], [])
|
|
assert cand["state"] == "new"
|
|
picked = PipePoint(
|
|
chainage_m=cand["chainage_m"], source="user", x=150.0, y=0.0,
|
|
stream_label=cand["stream_label"], id="p15000",
|
|
) # fmt: skip
|
|
pipes, crossing = _follow([picked], OLD, NEW1)
|
|
assert crossing == 0 and len(pipes) == 1 and pipes[0].stream_label == cand["stream_label"]
|
|
again = parse_pipe_points([pipe.as_dict() for pipe in pipes])
|
|
pipes2, crossing2 = _follow(again, NEW1, NEW2)
|
|
assert crossing2 == 0 and len(pipes2) == 1
|
|
# 새 노선에서 다시 물으면 = 있는 관(held) · 후보 아님
|
|
[again_cand] = propose_pipes(vertices_of(NEW2), _base_pipes(vertices_of(NEW2), ONE), pipes2, [])
|
|
assert again_cand["state"] == "held"
|
|
|
|
|
|
@pytest.mark.skipif(shutil.which("node") is None or not TSC.is_dir(), reason="node · typescript 없음")
|
|
def test_화면_고른_것만_넣음_라벨_왕복():
|
|
run = subprocess.run(
|
|
["node", str(HERE / "helper_510_propose.cjs")],
|
|
capture_output=True, text=True, encoding="utf-8", timeout=60, check=True,
|
|
) # fmt: skip
|
|
got = json.loads(run.stdout)
|
|
assert got["sent"] == {"url": "http://fake/projects/p1/drainage/pipe-proposals", "points": 1}
|
|
assert got["rows"] == [
|
|
"측점 50 m — 세류선 교차 · 지운 관",
|
|
"측점 320 m — 세류선 교차(작은골)",
|
|
"측점 420 m — 간격 보충(흐름이 모이는 자리)",
|
|
] # held 는 목록 밖
|
|
assert got["title"] == "자동 제안 3건 · 있는 관 1 뺌"
|
|
assert got["addOff"] is True and got["noCircles"] # 배수유역도 점선 원 없음(326-6)
|
|
assert got["added"] == [[320, "S4"], [420, None]] # 고른 둘만
|
|
assert got["afterClose"] == {"hidden": True}
|
|
assert got["carried"] == {"stream": "S4", "order": 1} # 응답 → 보관소 → 요청
|