feat(배수유역): 347 덧 관 고정 id — 측점 열쇠 pipe:<id> · 옮겨도 · 노선 바뀜 · 다시 셈에도 그대로

- 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

B04_PreProcess/B04_PreProcess_Api_Fetch.ts

@@ -510,6 +510,8 @@ export type PipeFacility = "pipe" | "box_culvert" | "ford_pavement" | "ford_brid
 
 export interface DetailPipePoint {
   chainage_m: number;
+  /** 관 고정 id(347 덧 · 측점 열쇠 `pipe:<id>`) — 서버가 붙임 · 옮겨도 그대로. */
+  id?: string;
   lonlat: [number, number];
   source: PipeSource;
   /** 시설 종류 — 응답에서 생략되면 기본 배관. */

B05_Profile/B05_Profile_UI_Drainage_Panel.ts

@@ -358,6 +358,7 @@ export function createDrainagePanel(callbacks: DrainagePanelCallbacks = {}): Dra
       response.pipe_points.map((pipe) => ({
         chainage_m: pipe.chainage_m,
         reason: pipe.source,
+        id: pipe.id,
       })),
     );
     // 시설 확장 정보(종류·구간·옵션)는 응답이 정본 — 통째로 다시 채운다(기다리던 월류 높이를 먼저 얹음).
@@ -395,6 +396,7 @@ export function createDrainagePanel(callbacks: DrainagePanelCallbacks = {}): Dra
       pipeEditor.pipes().map((pipe) => ({
         chainage_m: pipe.chainage_m,
         source: (pipe.reason || "user") as PipeSource,
+        ...(pipe.id ? { id: pipe.id } : {}),
       })),
     );
 

B05_Profile/B05_Profile_UI_Drainage_Parts.ts

@@ -354,9 +354,16 @@ export function reconcilePipes(
   if (next.length === current.length && next.every((value, index) => value === current[index])) {
     return null;
   }
+  const near = (a: number, b: number): boolean => Math.abs(a - b) < PIPE_MATCH_M;
+  // 사라진 관 하나 · 새 관 하나 = 옮긴 것 — 고정 id 를 따라 옮김(시설 속성 `follow` 와 같은 규칙)
+  const gone = before.filter((pipe) => !next.some((value) => near(pipe.chainage_m, value)));
+  const added = next.filter((value) => !before.some((pipe) => near(pipe.chainage_m, value)));
   return next.map((chainage) => {
-    const matched = before.find((pipe) => Math.abs(pipe.chainage_m - chainage) < PIPE_MATCH_M);
-    return { chainage_m: chainage, reason: matched?.reason ?? "confirmed" };
+    const matched =
+      before.find((pipe) => near(pipe.chainage_m, chainage)) ??
+      (gone.length === 1 && added.length === 1 && added[0] === chainage ? gone[0] : undefined);
+    const reason = matched && near(matched.chainage_m, chainage) ? matched.reason : "confirmed";
+    return { chainage_m: chainage, reason, ...(matched?.id ? { id: matched.id } : {}) };
   });
 }
 

B05_Profile/B05_Profile_UI_Drainage_Pipes.ts

@@ -12,6 +12,8 @@ import { snapStructureStation } from "../common_util/common_util_structure_stati
 export interface PipePoint {
   chainage_m: number;
   reason: string;
+  /** 관 고정 id(347 덧 · 측점 열쇠 `pipe:<id>`) — 옮겨도 그대로 · 새 관은 없음(서버가 붙임). */
+  id?: string;
 }
 
 interface RoutePointLike {

common_util/common_util_drainage_pipes.py

@@ -101,6 +101,9 @@ class PipePoint:
     summary_item: dict[str, Any] | None = None
     # 옵션 그룹(M02 PLAN 46)마다 고른 집계 항목 `{그룹 id: summary_item}` — 본체는 `summary_item`
     group_items: dict[str, Any] | None = None
+    # 관 고정 id(347 덧) — 측점 열쇠 `pipe:<id>`(common_util_station_key) · 옮겨도 · 노선이
+    # 바뀌어도 그대로 · 빈 id 는 `ensure_pipe_ids` 가 읽기 · 되붙이기 · 저장 때 채움.
+    id: str | None = None
 
     def as_dict(self) -> dict[str, Any]:
         # 구 형식 저장분이 확장 필드 없이 그대로 다시 저장되도록 기본값은 생략한다.
@@ -112,6 +115,8 @@ class PipePoint:
             "chainage_m": round(float(self.chainage_m), 3),
             "source": self.source,
         }
+        if self.id:
+            payload["id"] = self.id
         if self.facility != PIPE_FACILITY_PIPE:
             payload["facility"] = self.facility
         if self.start_m is not None and self.end_m is not None:
@@ -129,6 +134,26 @@ class PipePoint:
         return payload
 
 
+def ensure_pipe_ids(points: list[PipePoint]) -> list[PipePoint]:
+    """id 없는 관에 id 를 채운다(제자리) — `p<누가거리 cm>` · 겹치면 `-2` 꼬리.
+
+    누가거리로 지어 같은 자료를 두 번 읽어도 같은 id(저장 전 GET 을 되풀이해도 열쇠가 안 흔들림).
+    한 번 붙은 id 는 관을 옮겨도 그대로다.
+    """
+    taken = {point.id for point in points if point.id}
+    for point in points:
+        if point.id:
+            continue
+        base = f"p{int(round(float(point.chainage_m) * 100))}"
+        candidate, order = base, 1
+        while candidate in taken:
+            order += 1
+            candidate = f"{base}-{order}"
+        point.id = candidate
+        taken.add(candidate)
+    return points
+
+
 def edits_dir(stored_path: str) -> Path:
     return drainage_dir(stored_path) / DRAINAGE_EDITS_DIRNAME
 
@@ -402,10 +427,11 @@ def parse_pipe_points(values: Any) -> list[PipePoint]:
                 ),
                 summary_item=parse_summary_item(item.get("summary_item")),
                 group_items=parse_group_items(item.get("group_items")),
+                id=str(item["id"]) if item.get("id") else None,
             )
         )
     points.sort(key=lambda point: point.chainage_m)
-    return points
+    return ensure_pipe_ids(points)
 
 
 # 시설 속성을 붙이는 허용 오차(m) — 측점 반올림(0.01m)을 덮되 이웃 관에는 안 닿는 폭.
@@ -440,7 +466,7 @@ def carry_facility_attributes(base: list[PipePoint], reference: list[PipePoint])
     붙이기는 `carry_indices` — 허용 오차 밖의 관은 속성 없이 둔다(옆 관 속성을 가져가지 않음).
     """
     if not reference:
-        return base
+        return ensure_pipe_ids(base)
     picks = carry_indices(
         [point.chainage_m for point in base], [ref.chainage_m for ref in reference]
     )
@@ -454,7 +480,8 @@ def carry_facility_attributes(base: list[PipePoint], reference: list[PipePoint])
         point.options = ref.options
         point.summary_item = ref.summary_item
         point.group_items = ref.group_items
-    return base
+        point.id = ref.id
+    return ensure_pipe_ids(base)
 
 
 def save_pipe_points(
@@ -470,6 +497,7 @@ def save_pipe_points(
     """
     if vertices:
         fill_pipe_coordinates(points, vertices)
+    ensure_pipe_ids(points)
     path = pipe_points_path(stored_path)
     path.parent.mkdir(parents=True, exist_ok=True)
     atomic_write_json(

resources/tester/helper_347_pipe_id_reconcile.cjs

@@ -0,0 +1,39 @@
+/* 347 덧 관 고정 id — `reconcilePipes`(되돌리기 · 구조물 폼으로 관 목록이 바뀔 때)가 id 를 따라 옮기는지.
+ * `B05_Profile_UI_Drainage_Parts.ts` 를 그 자리에서 트랜스파일 · 그 밖 import 는 빈 껍데기(틀 = helper_281_1).
+ */
+const fs = require("fs");
+const path = require("path");
+const ts = require(path.join(__dirname, "..", "..", "config", "node_modules", "typescript"));
+
+const ROOT = path.join(__dirname, "..", "..");
+const stub = new Proxy(function () {}, { get: () => stub, apply: () => stub });
+
+function loadTs(file, real) {
+  const js = ts.transpileModule(fs.readFileSync(file, "utf8"), {
+    compilerOptions: { module: ts.ModuleKind.CommonJS, target: ts.ScriptTarget.ES2020 },
+  }).outputText;
+  const box = { exports: {} };
+  const localRequire = (request) =>
+    real.some((name) => request.endsWith(name))
+      ? loadTs(path.join(path.dirname(file), `${request}.ts`), [])
+      : stub;
+  new Function("exports", "module", "require", js)(box.exports, box, localRequire);
+  return box.exports;
+}
+
+const { reconcilePipes } = loadTs(
+  path.join(ROOT, "B05_Profile", "B05_Profile_UI_Drainage_Parts.ts"),
+  ["design_selection"],
+);
+const before = [
+  { chainage_m: 100, reason: "stream", id: "a1" },
+  { chainage_m: 200, reason: "user", id: "b2" },
+];
+process.stdout.write(
+  JSON.stringify({
+    same: reconcilePipes(before, [100, 200]),
+    moved: reconcilePipes(before, [100, 250]),
+    added: reconcilePipes(before, [100, 200, 300]),
+    twoMoved: reconcilePipes(before, [150, 250]),
+  }),
+);

resources/tester/test_312_empty_pipes.py

@@ -115,7 +115,19 @@ def test_POST_빈_목록은_관_0_유역_0(api) -> None:
     assert len(_file_points(api)) > 0  # 계산만 · 저장 안 함
 
 
+def _fresh_analysis(api) -> None:
+    """사본의 배수유역 분석에 지금 노선 지문을 찍음 — 옛 분석(지문 없음)은 낡음(342-3)."""
+    signature = api.get(f"{BASE}/drainage/pipe-points").json()["route_signature"]
+    path = api.root / "B04_PreProcess/drainage/00_watershed_response.json"
+    saved = json.loads(path.read_text(encoding="utf-8"))
+    path.write_text(json.dumps({**saved, "route_signature": signature}), encoding="utf-8")
+
+
 def test_칸_없음_파일_없음_DELETE_노선_바뀜은_자동_배치(api) -> None:
+    # 옛 분석(노선 지문 없음) = 낡음 — 관 목록이 없으면 자동 배치 대신 409(342-3 지문)
+    stale = api.post(f"{BASE}/drainage/detail-basins", json={})
+    assert stale.status_code == 409 and stale.json()["status"] == "stale"
+    _fresh_analysis(api)
     auto = api.post(f"{BASE}/drainage/detail-basins", json={}).json()["pipe_points"]
     assert len(auto) > 0  # 요청에 칸 없음 = 자동 배치
     api.put(f"{BASE}/drainage/pipe-points", json={"points": []})

resources/tester/test_347_pipe_id.py

@@ -0,0 +1,97 @@
+"""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]

resources/tester/test_b04_pipe_points_facility.py

@@ -38,9 +38,9 @@ def test_legacy_bare_number_still_parses():
 
 
 def test_legacy_dict_roundtrip_stays_compact():
-    """구 형식 저장분은 확장 필드 없이 그대로 다시 저장된다 (파일 불변)."""
+    """구 형식 저장분은 확장 필드 없이 다시 저장된다 — 고정 id 만 더함(347 덧 · 다음 저장)."""
     point = parse_pipe_points([{"chainage_m": 60.0, "source": "user"}])[0]
-    assert point.as_dict() == {"chainage_m": 60.0, "source": PIPE_SOURCE_USER}
+    assert point.as_dict() == {"chainage_m": 60.0, "source": PIPE_SOURCE_USER, "id": "p6000"}
 
 
 # ── 시설 종류 ───────────────────────────────────────────────────────────────