eomsangdonandClaude Opus 5.5
44561251ea
- PipePoint.id · ensure_pipe_ids(빈 id = p<누가거리 cm> · 겹치면 꼬리) — 읽기(parse) · 되붙이기(carry · 원본 id) · 저장(save) 세 자리 · 옛 자료는 다음 저장에 남음 · 노선 투영 · 새 자동 관 = 새 id - 화면 관 모양 DetailPipePoint.id · 배관 편집기 관이 id 를 쥐고 요청에 실음 · reconcilePipes 제자리 · 하나 옮긴 관 id 따라감 - 시험 test_347_pipe_id(8 · station_key = pipe:<id> · 화면 거울) · test_b04_pipe_points_facility 옛 자료 = id 하나만 더함 · test_312 옛 분석 = 낡음(409) 뒤 지문 찍고 자동 배치(새 지문 동작에 맞춤) Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01XWh5wNNycMkbwMpACz1rte
98 lines
4.3 KiB
Python
98 lines
4.3 KiB
Python
"""347 덧 관 고정 id — 측점 열쇠 `pipe:<id>` · 옮겨도 · 노선이 바뀌어도 · 다시 셈에도 그대로.
|
|
|
|
- 옛 자료(id 없음)는 읽을 때 `p<누가거리 cm>` · 두 번 읽어도 같음 · 다음 저장에 남음.
|
|
- 세부유역 다시 셈(계산기 왕복)은 허용 오차 안 원본 id 를 되붙임 · 새 자동 관 = 새 id · 겹치면 꼬리.
|
|
- 노선 바뀜(좌표 투영)에도 id 유지 · 측점 열쇠 = pipe:<id>(common_util_station_key).
|
|
"""
|
|
|
|
import json
|
|
from pathlib import Path
|
|
from types import SimpleNamespace
|
|
|
|
from B04_PreProcess import B04_PreProcess_Router_Basins as basins
|
|
from common_util import common_util_drainage_pipes as pipes_mod
|
|
from common_util.common_util_drainage_detail import DrainageDetail
|
|
from common_util.common_util_drainage_pipes import (
|
|
PipePoint,
|
|
carry_facility_attributes,
|
|
ensure_pipe_ids,
|
|
parse_pipe_points,
|
|
project_pipe_points,
|
|
read_pipe_points_file,
|
|
)
|
|
from common_util.common_util_route_geometry import RouteVertex, StructureCandidate
|
|
from common_util.common_util_station_key import station_key
|
|
|
|
|
|
def test_old_data_gets_stable_ids():
|
|
raw = [{"chainage_m": 120.0, "source": "stream"}, {"chainage_m": 40.25}]
|
|
first = parse_pipe_points(raw)
|
|
again = parse_pipe_points(raw)
|
|
assert [p.id for p in first] == ["p4025", "p12000"] == [p.id for p in again]
|
|
kept = parse_pipe_points([{"chainage_m": 120.0, "id": "k7"}])
|
|
assert kept[0].id == "k7" and kept[0].as_dict()["id"] == "k7"
|
|
|
|
|
|
def test_collision_gets_suffix():
|
|
points = [PipePoint(chainage_m=100.0, id="p10000"), PipePoint(chainage_m=100.004)]
|
|
assert [p.id for p in ensure_pipe_ids(points)] == ["p10000", "p10000-2"]
|
|
|
|
|
|
def test_recompute_carries_ids_and_names_new_pipe():
|
|
reference = [PipePoint(chainage_m=100.0, id="a1", facility="ford_bridge")]
|
|
rebuilt = [PipePoint(chainage_m=100.004), PipePoint(chainage_m=250.0, source="spacing")]
|
|
carried = carry_facility_attributes(rebuilt, reference)
|
|
assert carried[0].id == "a1" and carried[0].facility == "ford_bridge"
|
|
assert carried[1].id == "p25000"
|
|
|
|
|
|
def test_save_writes_ids(tmp_path, monkeypatch):
|
|
monkeypatch.setattr(pipes_mod, "pipe_points_path", lambda stored: tmp_path / "p.json")
|
|
pipes_mod.save_pipe_points("x", "sig", [PipePoint(chainage_m=55.5)])
|
|
saved = json.loads((tmp_path / "p.json").read_text("utf-8"))
|
|
assert saved["points"][0]["id"] == "p5550"
|
|
assert read_pipe_points_file(tmp_path / "p.json")[0].id == "p5550"
|
|
|
|
|
|
def test_route_change_keeps_id():
|
|
vertices = [RouteVertex(0.0, 0.0, 0.0, 0.0), RouteVertex(0.0, 300.0, 0.0, 300.0)]
|
|
moved = project_pipe_points([PipePoint(chainage_m=10.0, x=0.0, y=120.0, id="a1")], vertices)
|
|
assert moved[0].chainage_m == 120.0 and moved[0].id == "a1"
|
|
|
|
|
|
def test_station_key_is_pipe_id():
|
|
point = parse_pipe_points([{"chainage_m": 120.0, "id": "a1"}])[0]
|
|
assert station_key("pipe", point.chainage_m, 20.0, point.id) == "pipe:a1"
|
|
|
|
|
|
def test_build_keeps_request_ids(monkeypatch):
|
|
detail = DrainageDetail(
|
|
pipes=[
|
|
StructureCandidate(chainage_m=100.0, x=0.0, y=0.0, reason="confirmed"),
|
|
StructureCandidate(chainage_m=200.0, x=0.0, y=0.0, reason="confirmed"),
|
|
]
|
|
)
|
|
monkeypatch.setattr(basins, "build_detail", lambda *args: detail)
|
|
monkeypatch.setattr(basins, "drainage_dir", lambda path: Path("."))
|
|
points = [PipePoint(chainage_m=100.0, id="a1"), PipePoint(chainage_m=200.0, id="b2")]
|
|
_, carried = basins._build("x", SimpleNamespace(vertices=[]), points)
|
|
assert [p.id for p in carried] == ["a1", "b2"]
|
|
|
|
|
|
def test_screen_reconcile_carries_id():
|
|
"""화면 `reconcilePipes` — 제자리 관 · 하나 옮긴 관은 id 그대로 · 새 관 · 둘 넘게 바뀐 관은 id 없음."""
|
|
import os
|
|
import subprocess
|
|
|
|
helper = os.path.join(os.path.dirname(__file__), "helper_347_pipe_id_reconcile.cjs")
|
|
proc = subprocess.run(
|
|
["node", helper], capture_output=True, text=True, timeout=60, encoding="utf-8"
|
|
)
|
|
assert proc.returncode == 0, proc.stderr
|
|
result = json.loads(proc.stdout)
|
|
assert result["same"] is None
|
|
assert [p.get("id") for p in result["moved"]] == ["a1", "b2"]
|
|
assert result["moved"][1]["reason"] == "confirmed"
|
|
assert [p.get("id") for p in result["added"]] == ["a1", "b2", None]
|
|
assert [p.get("id") for p in result["twoMoved"]] == [None, None]
|