fix(B05,B06): 레거시 기성막이 관 지점 이관·스테일 테스트·죽은 코드 정리 통합 (PR #9)

migrate 400 회귀를 pipe_points 이관으로 해소. B06_Section_Engine_Revetment.py 삭제.

B05_Profile/B05_Profile_Structures_Migration.py

@@ -5,14 +5,18 @@
 
 - 배관: 옮기지 않는다 — `pipe_points.json`이 정본이다.
 - 기성막이 → 기슭막이(`revetment`). 기존 이름이 오기였다 (2026-08-16 사용자 확정).
+  기슭막이는 2026-08-28 이관으로 관 지점 시설이 됐다 — 구조물 목록이 아니라
+  `pipe_points.json`으로 간다 (2026-09-01 사용자 확정, 건너뛰기는 데이터 유실).
 - 대피로 → 대피소(`refuge`)로 통합. 폭 값은 대피소 너비 옵션으로 옮긴다 (동 확정).
 - 그 밖: 이름을 살려 `기타`.
 
-구간형으로 바뀌는 타입(기슭막이·대피소)은 기존 데이터에 종점이 없다. 기점만 알고 있으므로
+구간형으로 바뀌는 타입(대피소)은 기존 데이터에 종점이 없다. 기점만 알고 있으므로 최소
-최소 구간을 임시로 주고, 사용자가 화면에서 종점을 조정한다 — 값을 지어내는 것보다 낫다.
+구간을 임시로 주고, 사용자가 화면에서 종점을 조정한다 — 값을 지어내는 것보다 낫다.
+관 시설로 가는 기슭막이는 레지스트리 기본값(사용자 확정분)을 승계한다.
 """
 
 import re
+from dataclasses import dataclass, field
 from typing import Any, Iterable
 
 from B05_Profile.B05_Profile_Structures_Schema import (
@@ -20,8 +24,21 @@ from B05_Profile.B05_Profile_Structures_Schema import (
     station_planting_labels,
     structure_type_map,
 )
+from common_util.common_util_drainage_pipes import PIPE_SOURCE_USER, PipePoint
 
 PIPE_STRUCTURE_NAME = "배관"
+# 구조물 정본이 아니라 관 지점 정본이 관리한다는 표식(레지스트리 `managed_by`).
+PIPE_POINTS_OWNER = "pipe_points"
+
+
+@dataclass
+class MigrationPlan:
+    """이관 결과 — 정본이 갈리므로 목록도 갈린다(2026-09-01 사용자 확정)."""
+
+    structures: list[StructureInstance] = field(default_factory=list)
+    pipe_facilities: list[PipePoint] = field(default_factory=list)
+
+
 # 기존 타입 이름 → 신규 type_id.
 LEGACY_TYPE_MAP = {
     "기성막이": "revetment",
@@ -40,8 +57,14 @@ _ESCAPE_LABEL = re.compile(r"^대피로(?:\s+([\d.]+)\s*m)?", re.UNICODE)
 def _managed_elsewhere_labels() -> set[str]:
     """서버가 종단 정본에 직접 심는 라벨 — 구조물로 재이관하면 안 된다(정본 이중화 →
     "기타" 고스트). 규칙은 측점 생성과 **같은 자리**에서 온다(station_planting_labels).
+
+    관 지점 정본이 관리하는 타입 이름도 같이 막는다 — 그 측점은 이미 관 시설에서
+    투영된 것이라 되옮기면 이름을 못 알아보고 "기타"로 굳는다(기슭막이 2026-08-28 이관).
     """
-    return station_planting_labels()
+    owned = {
+        item.name for item in structure_type_map().values() if item.managed_by == PIPE_POINTS_OWNER
+    }
+    return station_planting_labels() | owned
 
 
 def _is_managed_elsewhere(entry: dict[str, Any]) -> bool:
@@ -84,10 +107,16 @@ def _escape_width_of(entry: dict[str, Any]) -> float | None:
     return None
 
 
-def migrate_irregular_stations(entries: Iterable[dict[str, Any]]) -> list[StructureInstance]:
+def migrate_irregular_stations(entries: Iterable[dict[str, Any]]) -> MigrationPlan:
-    """기존 비정규 측점 목록을 구조물 인스턴스로 옮긴다(같은 입력이면 같은 결과)."""
+    """기존 비정규 측점 목록을 정본별로 나눠 옮긴다(같은 입력이면 같은 결과).
+
+    타입이 다른 정본 소관(`managed_by`)이면 구조물 목록이 아니라 그 정본으로 간다 —
+    기슭막이가 관 지점 시설로 옮겨간 뒤(2026-08-28) 구조물로 저장하면 저장소가 거절한다
+    (2026-09-01 사용자 확정: 건너뛰지 않고 관 지점으로 옮긴다).
+    """
     types = structure_type_map()
-    migrated: list[StructureInstance] = []
+    structures: list[StructureInstance] = []
+    facilities: list[PipePoint] = []
     seen: set[tuple[str, float]] = set()
 
     for entry in entries:
@@ -104,6 +133,10 @@ def migrate_irregular_stations(entries: Iterable[dict[str, Any]]) -> list[Struct
             continue
         seen.add(key)
 
+        if definition.managed_by == PIPE_POINTS_OWNER:
+            facilities.append(_pipe_facility_for(definition, chainage))
+            continue
+
         data: dict[str, Any] = {
             "type_id": type_id,
             "placement": definition.placement,
@@ -116,9 +149,34 @@ def migrate_irregular_stations(entries: Iterable[dict[str, Any]]) -> list[Struct
             data["memo"] = "구 비정규 측점 이관 — 종점 확인 필요"
         else:
             data["chainage_m"] = chainage
-        migrated.append(StructureInstance.model_validate(data))
+        structures.append(StructureInstance.model_validate(data))
 
-    return migrated
+    return MigrationPlan(structures=structures, pipe_facilities=facilities)
+
+
+def _pipe_facility_for(definition, chainage: float) -> PipePoint:
+    """관 지점 정본이 관리하는 타입 하나를 관 시설로 만든다.
+
+    레거시 측점에는 제원이 없다(누가거리 + 표시 문자열뿐). 값을 지어내지 않고
+    **레지스트리 기본값**을 그대로 승계한다 — 기슭막이의 형태·높이·길이·전/후는
+    사용자 확정 기본값이다(2026-08-17·08-19). 옵션 키는 옛 전용 키 한 벌로 두면
+    B05 폼(`legacyRevetOptions`)과 B06 세트(`_revet_set`)가 양쪽 벽으로 펼쳐 읽는다.
+    """
+    options = {
+        option.key: option.default
+        for option in definition.options
+        if option.phase == "b05" and option.default is not None
+    }
+    before = float(options.get("before_m") or 0.0)
+    after = float(options.get("after_m") or 0.0)
+    return PipePoint(
+        chainage_m=chainage,
+        source=PIPE_SOURCE_USER,
+        facility=definition.type_id,
+        start_m=chainage - before,
+        end_m=chainage + after,
+        options=options,
+    )
 
 
 def _options_for(type_id: str, entry: dict[str, Any]) -> dict[str, Any]:

B05_Profile/B05_Profile_Structures_Router.py

@@ -36,6 +36,12 @@ from B05_Profile.B05_Profile_Structures_Schema import (
     load_structure_types,
     registry_schema_version,
 )
+from common_util.common_util_drainage_pipes import (
+    PipePoint,
+    append_pipe_points_file,
+    pipe_points_path_in,
+    read_pipe_points_file,
+)
 from common_util.common_util_storage import resolve_stored_project_path
 from config.config_db import get_db_pool
 
@@ -91,6 +97,36 @@ async def _route_length(project_id: UUID) -> float | None:
         return None
 
 
+def _migrate_pipe_facilities(
+    root: str, facilities: list[PipePoint], migrated_before: set[str]
+) -> tuple[int, set[str]]:
+    """관 지점 정본이 관리하는 이관 후보를 그 정본에 덧붙인다. (옮긴 수, 이력 표식).
+
+    관 지점 파일이 없으면(= B04 배수유역 산출물 없음) 아무것도 쓰지 않고 이력도 남기지
+    않는다 — 원천 비정규 측점은 그대로 있으므로, 산출물이 생긴 뒤 다시 이관된다.
+    표식은 구조물 쪽과 같은 "타입@위치"라 한 벌로 셈해도 부딪히지 않는다.
+    """
+    if not facilities:
+        return 0, set()
+
+    def key_of(point: PipePoint) -> str:
+        return f"{point.facility}@{round(float(point.chainage_m), 3)}"
+
+    path = pipe_points_path_in(Path(root))
+    if not path.is_file():
+        logger.info("B05 이관: 관 지점 정본이 없어 관 시설 %d건을 미뤘습니다.", len(facilities))
+        return 0, set()
+    occupied = {key_of(point) for point in read_pipe_points_file(path)}
+    fresh = [
+        point
+        for point in facilities
+        if key_of(point) not in occupied and key_of(point) not in migrated_before
+    ]
+    if fresh and append_pipe_points_file(path, fresh) is None:
+        return 0, set()
+    return len(fresh), {key_of(point) for point in facilities}
+
+
 async def _invalidate_downstream(project_id: UUID) -> bool:
     """구조물이 바뀌었으니 B06(stage 3) 이후의 완료 단계를 STALE로 되돌린다.
 
@@ -156,7 +192,7 @@ async def migrate_structures(project_id: UUID, payload: StructureMigrateRequest)
             return JSONResponse(status_code=404, content=_PROJECT_PATH_MISSING)
         revision, existing = load_structures(root)
         migrated_before = load_migrated_legacy(root)
-        candidates = migrate_irregular_stations(payload.stations)
+        plan = migrate_irregular_stations(payload.stations)
         # 표식 = "타입@위치". 원천(종단 정본의 비정규 측점)은 원복용으로 그대로 두고,
         # 옮긴 이력만 남긴다 — 그래야 사용자가 지운 구조물이 재진입 때 되살아나지 않는다
         # (2026-08-24 사용자: 초기 계산값은 원복용, 사용자 수정 1세트가 최종본).
@@ -167,11 +203,16 @@ async def migrate_structures(project_id: UUID, payload: StructureMigrateRequest)
         occupied = {key_of(item) for item in existing}
         fresh = [
             item
-            for item in candidates
+            for item in plan.structures
             if key_of(item) not in occupied and key_of(item) not in migrated_before
         ]
+        # 관 지점 정본이 관리하는 타입(기슭막이)은 구조물 목록에 넣을 수 없다 — 그쪽
+        # 정본에 덧붙인다(2026-09-01 사용자 확정: 건너뛰지 않고 옮긴다).
+        moved_pipes, pipe_history = _migrate_pipe_facilities(
+            root, plan.pipe_facilities, migrated_before
+        )
         # 이번에 건너뛴 것(이미 있던 자리)도 이력에 남긴다 — 그 자리는 이관이 끝난 자리다.
-        history = {key_of(item) for item in candidates}
+        history = {key_of(item) for item in plan.structures} | pipe_history
         if not fresh:
             # 새로 옮길 건 없어도 아직 이력에 없는 자리가 있으면 이력만 남긴다 — 그래야
             # 다음 진입에서 그 자리가 다시 후보로 잡히지 않는다. 이력이 이미 다 있으면
@@ -184,7 +225,14 @@ async def migrate_structures(project_id: UUID, payload: StructureMigrateRequest)
                     max_chainage_m=await _route_length(project_id),
                     migrated_legacy=history,
                 )
-            return JSONResponse(content={"status": "success", "migrated": 0, "revision": revision})
+            return JSONResponse(
+                content={
+                    "status": "success",
+                    "migrated": moved_pipes,
+                    "revision": revision,
+                    "pipe_facilities": moved_pipes,
+                }
+            )
         new_revision = save_structures(
             root,
             [*existing, *fresh],
@@ -192,9 +240,19 @@ async def migrate_structures(project_id: UUID, payload: StructureMigrateRequest)
             max_chainage_m=await _route_length(project_id),
             migrated_legacy=history,
         )
-        logger.info("B05 구 비정규 측점 이관: project_id=%s, %d건", project_id, len(fresh))
+        logger.info(
+            "B05 구 비정규 측점 이관: project_id=%s, 구조물 %d건 · 관 시설 %d건",
+            project_id,
+            len(fresh),
+            moved_pipes,
+        )
         return JSONResponse(
-            content={"status": "success", "migrated": len(fresh), "revision": new_revision}
+            content={
+                "status": "success",
+                "migrated": len(fresh) + moved_pipes,
+                "revision": new_revision,
+                "pipe_facilities": moved_pipes,
+            }
         )
     except LookupError:
         return JSONResponse(status_code=404, content=_PROJECT_PATH_MISSING)

B06_Section/B06_Section_Engine_Revetment.py

@@ -1,90 +0,0 @@
-"""독립 기슭막이(구조물 정본 D군) 제원을 횡단 측점에 얹는다.
-
-배관 유입·유출에 딸린 기슭막이는 관 정본이 관리하지만(`B06_Section_Engine_Culvert`),
-배관과 무관한 **독립 기슭막이**는 구조물 정본(`structures.json`)이 정본이다. 여기서는
-구간(시작~종료) 안의 측점에 형태·높이·설치 측을 붙이기만 한다 — 치수 결정·도형은
-화면(`B06_Section_UI_Cross_Revetment`)과 3D가 같은 산식으로 그린다(2026-08-28 사용자 확정).
-
-측점 자체는 `B05_Profile_Engine_Sections.resolve_extra_stations`가 시작·기준·종료에 심는다.
-"""
-
-import logging
-from pathlib import Path
-from typing import Any
-
-from B05_Profile.B05_Profile_Structures_Repository import load_structures
-from B05_Profile.B05_Profile_Structures_Schema import structure_type_map
-
-logger = logging.getLogger(__name__)
-
-# 구간 경계 측점을 구간 안으로 볼 허용 오차(m) — 정본이 누가거리를 0.01m로 끊어 쓴다.
-_EDGE_TOLERANCE_M = 0.02
-
-
-def load_revetments(project_root: Path) -> list[dict[str, Any]]:
-    """구조물 정본에서 독립 기슭막이(D군 구간형) 목록을 읽는다. 실패하면 빈 목록."""
-    try:
-        types = structure_type_map()
-        found: list[dict[str, Any]] = []
-        for structure in load_structures(str(project_root))[1]:
-            definition = types.get(structure.type_id)
-            if definition is None or definition.group != "D" or definition.placement != "interval":
-                continue
-            start, end = structure.start_m, structure.end_m
-            if start is None or end is None:
-                continue
-            options = structure.options or {}
-            found.append(
-                {
-                    "structure_id": structure.structure_id,
-                    "type_id": structure.type_id,
-                    "name": definition.name,
-                    "start_m": float(min(start, end)),
-                    "end_m": float(max(start, end)),
-                    "anchor_m": float(structure.anchor_m()),
-                    # 형태·높이는 **배관 유출 키 한 벌**이 정본이다(B05 폼 2026-08-28
-                    # 이관). 옛 전용 키(form·height_m)는 읽을 때만 폴백으로 본다 —
-                    # 이걸 안 보면 형태가 비어 도형이 늘 메쌓기로 그려졌다(2026-08-30).
-                    "form": options.get("outlet_revet_form") or options.get("form"),
-                    "height_m": (
-                        options.get("outlet_revet_height_m")
-                        if options.get("outlet_revet_height_m") is not None
-                        else options.get("height_m")
-                    ),
-                    "side": options.get("side"),
-                    # 다단 — 1이면 단일 벽. 전개 규칙은 화면·3D가 같은 산식으로 푼다.
-                    "tiers": options.get("tiers"),
-                    # 자리 이동(사용자 조작) — 사면 위로 올림 / 좌우 이동.
-                    "lift_m": options.get("lift_m"),
-                    "shift_m": options.get("shift_m"),
-                }
-            )
-        return found
-    except Exception:  # noqa: BLE001 — 정본을 못 읽어도 횡단 조회는 이어 간다
-        logger.exception("B06 독립 기슭막이 정본을 읽지 못했습니다 (없는 것으로 본다)")
-        return []
-
-
-def attach_revetments(project_root: Path, cross_sections: list[dict[str, Any]]) -> int:
-    """구간 안 측점의 횡단 dict에 `revetment` 키를 얹는다. 얹은 개수를 돌려준다.
-
-    한 측점에 여러 개가 겹치면 **먼저 시작한 것**을 쓴다 — 겹침 정리는 사용자 몫이다.
-    """
-    revetments = load_revetments(project_root)
-    if not revetments:
-        return 0
-    attached = 0
-    for section in cross_sections:
-        chainage = section.get("chainage_m")
-        if not isinstance(chainage, (int, float)):
-            continue
-        for spec in revetments:
-            if (
-                spec["start_m"] - _EDGE_TOLERANCE_M
-                <= float(chainage)
-                <= spec["end_m"] + _EDGE_TOLERANCE_M
-            ):
-                section["revetment"] = spec
-                attached += 1
-                break
-    return attached

common_util/common_util_drainage_pipes.py

@@ -33,6 +33,7 @@ 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, interpolate_vertex
 from config.config_system import (
+    DRAINAGE_CACHE_DIRNAME,
     DRAINAGE_DETAIL_FILENAME,
     DRAINAGE_EDITS_DIRNAME,
     DRAINAGE_PIPE_POINTS_FILENAME,
@@ -113,6 +114,21 @@ def pipe_points_path(stored_path: str) -> Path:
     return edits_dir(stored_path) / DRAINAGE_PIPE_POINTS_FILENAME
 
 
+def pipe_points_path_in(project_root: Path) -> Path:
+    """`pipe_points_path`와 같은 자리를 **프로젝트 실경로**로 가리킨다.
+
+    B05는 스토리지 상대경로가 아니라 실경로를 쥐고 있어 `resolve_stored_project_path`를
+    다시 태울 수 없다(절대경로를 거절한다).
+    """
+    return (
+        project_root
+        / "B04_PreProcess"
+        / DRAINAGE_CACHE_DIRNAME
+        / DRAINAGE_EDITS_DIRNAME
+        / DRAINAGE_PIPE_POINTS_FILENAME
+    )
+
+
 def detail_basins_path(stored_path: str) -> Path:
     return drainage_dir(stored_path) / DRAINAGE_DETAIL_FILENAME
 
@@ -330,6 +346,51 @@ def save_pipe_points(
     return len(points)
 
 
+def read_pipe_points_file(path: Path) -> list[PipePoint]:
+    """저장분을 **노선 지문과 무관하게** 그대로 읽는다. 파일이 없거나 깨졌으면 빈 목록.
+
+    지문 대조는 "이 노선에 그려도 되는가"를 가리는 것이고, 여기서 알고 싶은 것은
+    "그 자리에 이미 시설이 있는가"다(이관 중복 방지). 두 물음이 달라 읽기도 다르다.
+    """
+    if not path.is_file():
+        return []
+    try:
+        with path.open("r", encoding="utf-8") as file:
+            document = json.load(file)
+    except (OSError, json.JSONDecodeError):
+        logger.warning("배수유역: 관 지점 파일을 읽지 못했습니다 (%s).", path)
+        return []
+    return parse_pipe_points(document.get("points"))
+
+
+def append_pipe_points_file(path: Path, points: list[PipePoint]) -> int | None:
+    """관 지점 정본에 시설을 덧붙인다. 덧붙인 개수, 파일이 없으면 None.
+
+    저장 당시 노선 지문과 이미 있는 관은 그대로 둔다 — 여기서 지문을 새로 만들면
+    읽는 쪽이 노선이 바뀐 것으로 보고 저장분을 통째로 버린다(`load_pipe_points_file`).
+    지문을 알 수 없는 상황(파일 없음 = B04 배수유역 산출물 없음)에서는 쓰지 않고
+    None으로 알린다 — 지어낸 지문으로 쓰면 다음 읽기에서 사라진다.
+    """
+    if not path.is_file():
+        return None
+    try:
+        with path.open("r", encoding="utf-8") as file:
+            document = json.load(file)
+    except (OSError, json.JSONDecodeError):
+        logger.warning("배수유역: 관 지점 파일을 읽지 못해 덧붙이지 못했습니다 (%s).", path)
+        return None
+    stored = list(document.get("points") or [])
+    atomic_write_json(
+        path,
+        {
+            "route_signature": str(document.get("route_signature") or ""),
+            "points": [*stored, *(point.as_dict() for point in points)],
+        },
+    )
+    logger.info("배수유역: 관 지점 %d개를 덧붙였습니다 (%s).", len(points), path.name)
+    return len(points)
+
+
 def clear_pipe_points(stored_path: str) -> bool:
     """저장된 관 지점과 그 파생물을 지운다. 하나라도 지웠으면 True.