fix(배수유역): 관 지점에 좌표를 남겨 어느 노선으로 읽든 선 위에 앉힌다

사용자 지적(2026-08-30): "결국 노선 위에 위치해야 한다".

관 자리를 정한 선(계획노선 CSV)과 화면에 그려지는 선(B05 최적 경로)은 같은 자리를
지나면서 연장이 다르다 — 실측 350.11m vs 354.83m, 전 구간 이격은 평균 0.03m·최대
1.12m뿐인데 길이가 4.7m 차이난다. pipe_points.json이 누가거리만 저장해서 읽는 쪽이
쥔 선에 따라 같은 값이 3.3~4.3m 미끄러졌고, 노선 지문도 늘 어긋나 B05는 저장된 관을
매번 통째로 버렸다("노선 지문이 달라 저장된 관 지점 4건을 쓰지 않습니다").

- PipePoint에 x·y를 두고 저장 시 채운다(save_pipe_points에 노선을 넘긴다).
- 지문이 달라도 좌표가 있으면 읽는 쪽 노선에 투영해 이월한다(project_pipe_points).
  구간(start_m·end_m)은 기준점이 옮겨간 만큼 같이 민다 — 시설 치수는 불변.
- 좌표 없는 구 저장분만 종전대로 버린다.
- B05는 같은 로더(load_pipe_points_file)를 쓴다 — 두 화면의 판정 기준을 하나로.

검증: 프로젝트 5d18ebe3 실데이터 — 좌표 4/4 저장, B05 계획선으로 읽어 4건 전부 이월,
관에서 계획선까지 거리 최대 0.016m(종전 4건 전량 폐기). 누가거리는 +3.45~+4.29m
이동(연장 차이 그대로). pytest 11 passed(투영 이월·구 저장분 폐기 테스트 추가).

B04_PreProcess/B04_PreProcess_Router_Basins.py

@@ -237,7 +237,9 @@ async def _resolve(
 
     requested = parse_pipe_points((payload or {}).get("points"))
     signature = route_signature(context.vertices)
-    stored = load_pipe_points(context.stored_path, signature) if use_stored else None
+    stored = (
+        load_pipe_points(context.stored_path, signature, context.vertices) if use_stored else None
+    )
     points = requested or stored or None
 
     result = await asyncio.to_thread(_build, context.stored_path, context, points)
@@ -303,7 +305,10 @@ async def put_pipe_points(
     context, detail, points, _ = resolved
 
     signature = route_signature(context.vertices)
-    saved = await asyncio.to_thread(save_pipe_points, context.stored_path, signature, points)
+    # 좌표를 같이 남긴다 — 다른 선(B05 최적 경로)으로 읽어도 그 자리에 되놓는다.
+    saved = await asyncio.to_thread(
+        save_pipe_points, context.stored_path, signature, points, context.vertices
+    )
     await asyncio.to_thread(
         save_detail_basins, context.stored_path, _basin_features(context, detail, points)
     )

B05_Profile/B05_Profile_Engine_Sections.py

@@ -31,7 +31,7 @@ from common_util.common_util_drainage_pipes import (
     PIPE_FACILITY_PIPE,
     PIPE_FACILITY_REVET,
     PipePoint,
-    parse_pipe_points,
+    load_pipe_points_file,
     pipe_anchor_clearances,
     route_signature,
 )
@@ -111,9 +111,17 @@ def _load_pipe_points(project_root: Path, polyline: list[list[float]]) -> list[P
     계획선 정착과 구조물 측점 생성이 **같은 한 번의 읽기**를 쓴다 — 따로 읽으면 계획선이
     물린 자리와 측점 자리가 어긋난다.
 
-    관 지점 파일에는 저장 당시 노선 지문이 함께 있다 — 노선이 바뀌었으면 버린다
-    (옛 노선의 배관 자리로 계획선을 앉히면 전부 어긋난다). 파일이 없거나 못 읽으면 빈 목록.
+    관 지점 파일에는 저장 당시 노선 지문이 함께 있다. 지문이 달라도 저장분에 좌표가 있으면
+    **이 계획선에 투영해 이월한다** — B04가 관 자리를 정한 선(계획노선 CSV)과 여기 계획선은
+    같은 자리를 지나면서 연장이 다르다(실측 350.11m vs 354.83m). 그래서 지문은 거의 항상
+    달랐고 관이 통째로 빠졌다(2026-08-30 사용자 지적). 좌표가 없는 구 저장분만 버린다.
     """
+    vertices = [
+        RouteVertex(
+            x=float(p[0]), y=float(p[1]), z=float(p[2]) if len(p) > 2 else 0.0, chainage_m=0.0
+        )
+        for p in polyline
+    ]
     path = (
         project_root
         / "B04_PreProcess"
@@ -123,25 +131,12 @@ def _load_pipe_points(project_root: Path, polyline: list[list[float]]) -> list[P
     )
     if not path.is_file():
         return []
-    try:
-        document = json.loads(path.read_text(encoding="utf-8"))
-    except (OSError, json.JSONDecodeError):
-        logger.warning("B05 계획선: 관 지점 파일을 읽지 못했습니다 (%s)", path)
+    points = load_pipe_points_file(path, route_signature(vertices), vertices)
+    if points is None:
+        # 계획선 정착과 구조물 측점이 함께 빠지므로 남긴다(침묵 실패 금지).
+        logger.warning("B05 계획선: 저장된 관 지점을 쓰지 못했습니다 (좌표 없는 구 저장분).")
         return []
-    vertices = [
-        RouteVertex(
-            x=float(p[0]), y=float(p[1]), z=float(p[2]) if len(p) > 2 else 0.0, chainage_m=0.0
-        )
-        for p in polyline
-    ]
-    if str(document.get("route_signature") or "") != route_signature(vertices):
-        # 계획선 정착과 구조물 측점이 함께 빠지므로 버린 건수를 남긴다(침묵 실패 금지).
-        logger.warning(
-            "B05 계획선: 노선 지문이 달라 저장된 관 지점 %d건을 쓰지 않습니다.",
-            len(document.get("points") or []),
-        )
-        return []
-    return parse_pipe_points(document.get("points"))
+    return points
 
 
 def resolve_extra_stations(

common_util/common_util_drainage_pipes.py

@@ -5,9 +5,17 @@
 읽고 쓴다 — 관리자 화면에서 옮긴 관이 사용자 화면에서 다르게 보이면 안 되기 때문이다
 (2026-08-01 사용자 지시).
 
-위치는 좌표가 아니라 **누가거리(chainage_m)** 로 저장한다. 지면 필터나 지표면 모델을 바꾸면
+위치는 **누가거리(chainage_m) + 좌표(x, y)** 로 저장한다. 지면 필터나 지표면 모델을 바꾸면
 종단 Z가 달라지지만 관이 놓인 자리는 그대로여야 하고, 그때는 세부유역만 다시 나누면 된다.
-노선 자체가 바뀌면(`route_signature` 불일치) 기준이 사라지므로 전량 버리고 다시 만든다.
+
+좌표를 같이 남기는 이유(2026-08-30 사용자 지적 — "결국 노선 위에 위치해야 한다"): 관 자리를
+정한 선(계획노선 CSV)과 화면에 그려지는 선(B05 최적 경로)은 **같은 자리를 지나면서 연장이
+다르다**(실측 350.11m vs 354.83m). 누가거리만 남기면 읽는 쪽이 쥔 선에 따라 같은 값이 3~4m
+미끄러져 관이 선 옆에 떨어진 것처럼 보인다. 좌표를 남겨 두면 어느 선으로 읽든 그 좌표를
+투영해 **항상 선 위에** 앉힐 수 있다.
+
+노선이 바뀌면(`route_signature` 불일치) 좌표가 있는 저장분은 새 노선에 투영해 이월하고,
+좌표가 없는 구 저장분만 버린다.
 """
 
 from __future__ import annotations
@@ -19,9 +27,11 @@ from dataclasses import dataclass
 from pathlib import Path
 from typing import Any
 
+from shapely.geometry import LineString, Point
+
 from B04_PreProcess.B04_PreProcess_Engine_Watershed_Export import drainage_dir
 from common_util.common_util_json import atomic_write_json
-from common_util.common_util_route_geometry import RouteVertex
+from common_util.common_util_route_geometry import RouteVertex, interpolate_vertex
 from config.config_system import (
     DRAINAGE_DETAIL_FILENAME,
     DRAINAGE_EDITS_DIRNAME,
@@ -72,6 +82,9 @@ class PipePoint:
     start_m: float | None = None
     end_m: float | None = None
     options: dict[str, Any] | None = None
+    # 관이 실제로 놓인 자리(사업지 CRS, m). 노선이 바뀌어도 이 자리는 그대로다.
+    x: float | None = None
+    y: float | None = None
 
     def as_dict(self) -> dict[str, Any]:
         # 구 형식 저장분이 확장 필드 없이 그대로 다시 저장되도록 기본값은 생략한다.
@@ -86,6 +99,9 @@ class PipePoint:
             payload["end_m"] = round(float(self.end_m), 2)
         if self.options:
             payload["options"] = self.options
+        if self.x is not None and self.y is not None:
+            payload["x"] = round(float(self.x), 3)
+            payload["y"] = round(float(self.y), 3)
         return payload
 
 
@@ -113,9 +129,59 @@ def route_signature(vertices: list[RouteVertex]) -> str:
     return f"{len(vertices)}-{digest.hexdigest()[:16]}"
 
 
-def load_pipe_points(stored_path: str, signature: str) -> list[PipePoint] | None:
-    """저장된 관 지점을 읽는다. 파일이 없거나 노선이 바뀌었으면 None(= 다시 만들어야 함)."""
-    path = pipe_points_path(stored_path)
+def fill_pipe_coordinates(points: list[PipePoint], vertices: list[RouteVertex]) -> list[PipePoint]:
+    """좌표가 비어 있는 관에 그 누가거리의 노선 좌표를 채운다(제자리 수정)."""
+    if not vertices:
+        return points
+    for point in points:
+        if point.x is None or point.y is None:
+            x, y, _ = interpolate_vertex(vertices, float(point.chainage_m))
+            point.x, point.y = float(x), float(y)
+    return points
+
+
+def project_pipe_points(points: list[PipePoint], vertices: list[RouteVertex]) -> list[PipePoint]:
+    """저장된 좌표를 주어진 노선에 투영해 누가거리를 다시 매긴다.
+
+    관이 놓인 **자리**는 좌표가 정본이고 누가거리는 그 자리를 읽는 선에 종속된 값이다.
+    앞뒤 구간(start_m·end_m)은 기준점이 옮겨간 만큼 같이 민다 — 구간 길이는 시설 치수라
+    노선이 바뀌어도 변하지 않는다.
+    """
+    if not vertices:
+        return points
+    line = LineString([(vertex.x, vertex.y) for vertex in vertices])
+    if line.length <= 0:
+        return points
+    for point in points:
+        if point.x is None or point.y is None:
+            continue
+        moved = float(line.project(Point(point.x, point.y)))
+        shift = moved - float(point.chainage_m)
+        point.chainage_m = moved
+        if point.start_m is not None:
+            point.start_m = float(point.start_m) + shift
+        if point.end_m is not None:
+            point.end_m = float(point.end_m) + shift
+    points.sort(key=lambda item: item.chainage_m)
+    return points
+
+
+def load_pipe_points(
+    stored_path: str, signature: str, vertices: list[RouteVertex] | None = None
+) -> list[PipePoint] | None:
+    """저장된 관 지점을 읽는다. 파일이 없거나 이월할 수 없으면 None(= 다시 만들어야 함).
+
+    노선 지문이 다르면 예전에는 전량 버렸다. 저장분에 좌표가 있으면 `vertices`(읽는 쪽이
+    쓰는 노선)에 투영해 이월한다 — 같은 자리를 지나면서 연장만 다른 선끼리 관이 통째로
+    사라지던 것을 막는다(2026-08-30 사용자 지적).
+    """
+    return load_pipe_points_file(pipe_points_path(stored_path), signature, vertices)
+
+
+def load_pipe_points_file(
+    path: Path, signature: str, vertices: list[RouteVertex] | None = None
+) -> list[PipePoint] | None:
+    """`load_pipe_points`와 같되 파일 경로로 직접 읽는다 (B05는 프로젝트 루트를 쥔다)."""
     if not path.exists():
         return None
     try:
@@ -124,11 +190,19 @@ def load_pipe_points(stored_path: str, signature: str) -> list[PipePoint] | None
     except (OSError, json.JSONDecodeError):
         logger.warning("배수유역: 관 지점 파일을 읽지 못했습니다 (%s).", path)
         return None
+    points = parse_pipe_points(document.get("points"))
     stored_signature = str(document.get("route_signature") or "")
-    if stored_signature != signature:
-        logger.info("배수유역: 노선이 바뀌어 저장된 관 지점을 버립니다 (%s).", path.name)
-        return None
-    return parse_pipe_points(document.get("points"))
+    if stored_signature == signature:
+        return points
+    if vertices and points and all(p.x is not None and p.y is not None for p in points):
+        logger.info(
+            "배수유역: 노선이 바뀌어 관 지점 %d건을 좌표로 이월합니다 (%s).",
+            len(points),
+            path.name,
+        )
+        return project_pipe_points(points, vertices)
+    logger.info("배수유역: 노선이 바뀌어 저장된 관 지점을 버립니다 (%s).", path.name)
+    return None
 
 
 def _parse_span(item: dict[str, Any], chainage: float) -> tuple[float | None, float | None]:
@@ -187,6 +261,7 @@ def parse_pipe_points(values: Any) -> list[PipePoint]:
         facility = str(item.get("facility") or PIPE_FACILITY_PIPE)
         start, end = _parse_span(item, float(chainage))
         options = item.get("options")
+        raw_x, raw_y = item.get("x"), item.get("y")
         points.append(
             PipePoint(
                 chainage_m=float(chainage),
@@ -194,6 +269,8 @@ def parse_pipe_points(values: Any) -> list[PipePoint]:
                 facility=facility if facility in _KNOWN_FACILITIES else PIPE_FACILITY_PIPE,
                 start_m=start,
                 end_m=end,
+                x=float(raw_x) if isinstance(raw_x, (int, float)) else None,
+                y=float(raw_y) if isinstance(raw_y, (int, float)) else None,
                 options=(
                     _migrate_protection(dict(options))
                     if isinstance(options, dict) and options
@@ -227,8 +304,19 @@ def carry_facility_attributes(base: list[PipePoint], reference: list[PipePoint])
     return base
 
 
-def save_pipe_points(stored_path: str, signature: str, points: list[PipePoint]) -> int:
-    """관 지점을 정본 파일에 쓴다. 저장된 개수를 돌려준다."""
+def save_pipe_points(
+    stored_path: str,
+    signature: str,
+    points: list[PipePoint],
+    vertices: list[RouteVertex] | None = None,
+) -> int:
+    """관 지점을 정본 파일에 쓴다. 저장된 개수를 돌려준다.
+
+    `vertices`를 주면 좌표가 빈 관을 그 노선 위 좌표로 채워 둔다 — 다음에 다른 선으로
+    읽어도 그 자리에 되놓을 수 있다.
+    """
+    if vertices:
+        fill_pipe_coordinates(points, vertices)
     path = pipe_points_path(stored_path)
     path.parent.mkdir(parents=True, exist_ok=True)
     atomic_write_json(