fix(765-8): 벌마다 평면노선 따로 — 받은 노선을 그 벌 지형으로 잘라 벌 예정노선으로 두고 따라감(요구 612)

- 벌 예정노선 terrain_route.csv(벌 몫) — 체인이 받은 노선을 그 지형으로 잘라 한 번 적음 · 초기 폴리라인(merged_follow) · 화면 예상노선 · 고칠 선 끝이 이것을 바탕으로(route_source_path · 없으면 공유 예상노선)
- 공유 예상노선 expected_route.csv = 설계 지형으로 자른 하나(추천 열쇠 그대로) · 고친 예정노선 expected_route_edit.csv 는 벌 몫(다른 벌에 안 번짐)
- 검증_sub5: 라이다 벌 1,352.9 m · 곡선 19(옛 값과 같음) · 수치지형도 벌 2,110.2 m · 곡선 28(예정노선 끝까지) · 칩 왕복 · 벌마다 [초기화] 값 같음 · 공유 예상노선 그대로
- 시험 test_765_design_sets — 벌 몫 가름에 벌 예정노선 · 고친 예정노선 · 벌 예정노선 자름 · 초기 폴리라인 바탕 둘 덧 · 지운 csf 필터 글자를 ptd · classification 으로(637 · 763_3 · 765)

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

B03_FileInput/B03_FileInput_Service_Chain.py

@@ -231,7 +231,8 @@ async def run_auto_design_chain(
         route_range = (range_row[0], range_row[1]) if range_row else None
         seeded_speed = range_row[2] if range_row else None
 
-        points = _planned_route_points_in_project_crs(project_root, defaults, route_range)
+        # 벌 예정노선 = 받은 노선을 이 벌 지형으로 자름(765-8 · 요구 612) — 벌마다 길이 · 곡선 다름
+        points = await asyncio.to_thread(_terrain_route_points, project_root, defaults, route_range)
         if not points:
             logger.warning("자동 설계 체인 중단(계획노선 없음): project_id=%s", project_id)
             mark_design_failed(project_root, "계획노선이 없어 초기 노선을 세울 수 없습니다.")
@@ -415,8 +416,8 @@ async def run_preprocess_chain(
             missing = "sheet" if design == "las" else "las"
             sets.note_unbuilt(root, missing, f"{missing} 지표면 모델 없음 — 한 벌만")
         else:
-            # 예상노선(공유 · 추천이 씀)은 설계 지형으로 자른 것 하나 — 먼저 적어 두면 두 벌이 같은
-            # 평면 노선으로 지어짐(지형만 다름 · 다른 지형 벌 [초기화]도 같은 노선)
+            # 예상노선(공유 · 추천 열쇠)은 설계 지형으로 자른 것 하나 — 벌 평면노선은 벌마다 그 지형으로
+            # 자른 벌 예정노선(`terrain_route.csv`)을 따라감(765-8 · 요구 612)
             await _seed_expected_route(project_id, root, params)
             try:
                 async with pool.acquire() as connection:
@@ -453,6 +454,35 @@ async def run_preprocess_chain(
         schedule_precompute(project_id)  # 추천은 지형 · 예정노선만 씀 — 실패 길에도(763-3)
 
 
+def _terrain_route_points(
+    root: Path, surface: dict[str, Any], route_range: tuple[Any, Any] | None
+) -> list[dict[str, float]] | None:
+    """벌 예정노선(`terrain_route.csv` · 벌 몫) — 있으면 그것 · 없으면 받은 노선을 이 지형으로
+    잘라 적음.
+
+    받은 노선 파일이 없는 옛 프로젝트 = 정본 차례(`load_design_route`) 그대로."""
+    from common_util.common_util_route_geometry import (
+        load_design_route,
+        read_planned_route_csv,
+        terrain_route_csv_path,
+        write_route_csv,
+    )
+
+    path = terrain_route_csv_path(root)
+    stored = read_planned_route_csv(path) if path.is_file() else None
+    if stored is not None and len(stored.vertices) >= 2:
+        return [{"x": v.x, "y": v.y} for v in stored.vertices]
+    planned = load_design_route(root, surface, route_range, from_source=True)
+    points = (
+        [{"x": v.x, "y": v.y} for v in planned.vertices]
+        if planned is not None
+        else _planned_route_points_in_project_crs(root, surface, route_range)
+    )
+    if points:
+        write_route_csv(path, points)
+    return points
+
+
 async def _seed_expected_route(project_id: UUID, root: Path, surface: dict[str, Any]) -> None:
     """예상노선 정본을 설계 지형으로 잘라 적음(없을 때만 · 자동 체인 2.5 와 같은 값)."""
     from common_util.common_util_route_geometry import expected_route_csv_path, write_route_csv

B05_Profile/B05_Profile_Router_FollowLand.py

@@ -15,8 +15,8 @@ from uuid import UUID
 
 from B05_Profile.B05_Profile_Engine_RouteInitial_Follow import FollowLand
 from common_util.common_util_route_geometry import (
-    expected_route_csv_path,
     planned_route_initial_path,
+    route_source_path,
 )
 from config.config_db import get_db_pool
 
@@ -63,7 +63,7 @@ async def follow_land(project_id: UUID, project_root: Path) -> FollowLand | None
 async def initial_land(project_id: UUID, project_root: Path) -> FollowLand | None:
     """초기 폴리라인을 새로 세울 때만 지형을 읽음(`_ensure_planned_initial` 과 같은 낡음 판정)."""
     target = planned_route_initial_path(project_root)
-    source = expected_route_csv_path(project_root)
+    source = route_source_path(project_root)
     if target.is_file():
         if not source.is_file() or source.stat().st_mtime <= target.stat().st_mtime:
             return None

B05_Profile/B05_Profile_Router_PlanSource.py

@@ -27,9 +27,9 @@ from shapely.geometry import LineString, Point
 from B05_Profile.B05_Profile_Router_Replan import _PROJECT_PATH_MISSING, _project_paths
 from common_util.common_util_route_geometry import (
     clip_route_by_chainage,
-    expected_route_csv_path,
     load_design_route,
     read_planned_route_csv,
+    route_source_path,
     trim_route_to_surface,
 )
 from common_util.common_util_surface_confirmation import get_surface_confirmation_params
@@ -72,8 +72,8 @@ def source_vertices(
 
 
 def clip_to_expected(root: Path, points: list[tuple[float, float]]) -> list[tuple[float, float]]:
-    """체인이 자른 예정노선 파일 양끝에서 자름(755-3) — 파일 없음 · 2점 미만 = 그대로."""
-    path = expected_route_csv_path(root)
+    """체인이 자른 벌 예정노선(765-8) 양끝에서 자름(755-3) — 파일 없음 · 2점 미만 = 그대로."""
+    path = route_source_path(root)
     stored = read_planned_route_csv(path) if path.is_file() else None
     if stored is None or len(stored.vertices) < 2 or len(points) < 2:
         return points

B05_Profile/B05_Profile_Router_Replan.py

@@ -61,6 +61,7 @@ from common_util.common_util_route_geometry import (
     planned_route_initial_path,
     planned_route_working_path,
     read_planned_route_csv,
+    route_source_path,
     write_route_csv,
 )
 from common_util.common_util_route_polyline import build_planned_polyline
@@ -217,7 +218,7 @@ def _ensure_planned_initial(project_root: Path, radius_m: float, land: Any = Non
     깔리는데 초기본이 옛 노선인 채로 남으면 노선 초기화가 옛 자리로 돌아간다.
     """
     target = planned_route_initial_path(project_root)
-    source = expected_route_csv_path(project_root)
+    source = route_source_path(project_root)  # 벌 예정노선(765-8) · 없으면 공유 예상노선
     if target.is_file():
         if not source.is_file() or source.stat().st_mtime <= target.stat().st_mtime:
             return None
@@ -391,7 +392,7 @@ async def read_route_plan(project_id: UUID) -> dict[str, Any] | JSONResponse:
     project_root, _ = paths
     # 추천 미리 셈은 전처리 체인에서만(505-1) — B05 열기는 저장본을 안 셈 · 안 씀
     # 예상노선 정본이 먼저, 없으면 옛 자리(초기값 스냅샷)를 본다.
-    expected = await asyncio.to_thread(_vertices_of, expected_route_csv_path(project_root))
+    expected = await asyncio.to_thread(_vertices_of, route_source_path(project_root))
     if not expected:
         expected = await asyncio.to_thread(_vertices_of, design_route_csv_path(project_root))
 

common_util/common_util_design_sets.py

@@ -53,7 +53,8 @@ _SET_DIRS = (
 )
 #: 공유를 뺀 나머지가 벌 몫인 폴더 → 공유 이름 앞머리
 _SPLIT_DIRS = {
-    "B05_Profile/route": ("cost_surface_", "expected_route", "initial_choice"),
+    # 공유 예상노선 = `expected_route.csv` 하나 — 고친 예정노선 · 벌 예정노선은 벌 몫(765-8)
+    "B05_Profile/route": ("cost_surface_", "expected_route.csv", "initial_choice"),
     "B04_PreProcess/drainage": DRAINAGE_SHARED_PREFIXES,
 }
 _ROUTE_CHILDREN = ("route_points", "route_statistics", "longitudinal_sections", "cross_sections")

common_util/common_util_route_geometry.py

@@ -229,6 +229,18 @@ def expected_route_edit_path(project_root: Path) -> Path:
     return Path(project_root) / "B05_Profile" / "route" / "expected_route_edit.csv"
 
 
+def terrain_route_csv_path(project_root: Path) -> Path:
+    """벌 예정노선(765-8 · 요구 612) — 받은 노선을 **그 벌 지형**으로 자른 것 · 벌 몫(묶음 안).
+    공유 `expected_route.csv` 는 설계 지형으로 자른 하나(추천 열쇠)."""
+    return Path(project_root) / "B05_Profile" / "route" / "terrain_route.csv"
+
+
+def route_source_path(project_root: Path) -> Path:
+    """초기 폴리라인 · 화면 예상노선의 바탕 — 벌 예정노선 · 없으면(옛 프로젝트) 공유 예상노선."""
+    own = terrain_route_csv_path(project_root)
+    return own if own.is_file() else expected_route_csv_path(project_root)
+
+
 def write_route_csv(path: Path, points: list[dict[str, float]]) -> None:
     """노선 정점을 CSV로 적는다. 열 이름은 `read_planned_route_csv()`가 아는 것."""
     path.parent.mkdir(parents=True, exist_ok=True)
@@ -312,6 +324,7 @@ def load_design_route(
         for master in (
             planned_route_working_path(project_root),
             planned_route_initial_path(project_root),
+            terrain_route_csv_path(project_root),  # 벌 예정노선(765-8)
             expected_route_csv_path(project_root),
             design_route_csv_path(project_root),
         ):

resources/tester/test_637_lidar.py

@@ -162,7 +162,7 @@ def test_DB_넣기_고치기만_갈음은_id_확정_살림(monkeypatch) -> None:
 
 
 def test_DB_지운_필터_행만_지움_확정_남김(monkeypatch) -> None:  # type: ignore[no-untyped-def]
-    gone = ["B04_PreProcess/models/dtm_csf.npz", "B04_PreProcess/models/dtm_csf_smooth.npz"]
+    gone = ["B04_PreProcess/models/dtm_ptd.npz", "B04_PreProcess/models/dtm_ptd_smooth.npz"]
     _, _, connection = _save(monkeypatch, keep=True, drop=gone)
     sql, args = [row for row in connection.sql if "DELETE" in row[0]][0]
     assert "status <> 'CONFIRMED'" in sql and "project_id = %s" in sql
@@ -354,13 +354,13 @@ def test_las_selection_적기_칩_열림(tmp_path: Path) -> None:
     assert confirm.analysis_terrain_sources(params, tmp_path) == ["sheet"]  # 칩 라이다 막힘
     cursor = _DictCursor(params)
     connection = type("C", (), {"cursor": lambda self, *_: cursor})()
-    pick = {"source_filter": "csf", "method": "dtm", "smooth": True}
+    pick = {"source_filter": "classification", "method": "dtm", "smooth": True}
     asyncio.run(confirm.stash_las_selection(connection, str(PID), pick))
     written = json.loads(cursor.sql[-1][1][0])
     assert written["las_selection"] == pick
     assert written["source_filter"] == "sheet_laplace"  # 설계 지형 그대로
     assert confirm.analysis_terrain_sources(written, tmp_path) == ["las", "sheet"]
-    assert confirm.analysis_terrain_params(written, "las")["source_filter"] == "csf"
+    assert confirm.analysis_terrain_params(written, "las")["source_filter"] == "classification"
 
 
 def test_지형_바꿈_force_같은_지형도_다시_셈(monkeypatch) -> None:  # type: ignore[no-untyped-def]

resources/tester/test_763_3_surface_change.py

@@ -106,7 +106,7 @@ def b04_world(tmp_path, monkeypatch):
                     "model_type": "implicit",
                     "generation_params": {"source_filter": "classification"},
                 },
-                {"id": 8, "model_type": "implicit", "generation_params": {"source_filter": "csf"}},
+                {"id": 8, "model_type": "implicit", "generation_params": {"source_filter": "ptd"}},
             ]
         ),
     )
@@ -165,7 +165,7 @@ def test_같은_서피스_다시_확정은_다시_셈_없음(b04_world) -> None:
 
 def test_다른_서피스_확정은_전처리_체인(b04_world) -> None:
     _confirm(8)
-    assert b04_world["confirm"] == [(8, "csf")]
+    assert b04_world["confirm"] == [(8, "ptd")]
     assert b04_world["chain"] == [8]
 
 

resources/tester/test_765_design_sets.py

@@ -7,6 +7,8 @@ storage = 시험 안 임시 폴더. 지킴:
   ② 꺼낼 때 새 노선 id · 자식 다시 잇기 · 코리도 이름 새 id
   ③ 공유 캐시(비용면 · 예정노선 · 배수 하강장 · 도엽 좌표 · 강우)는 안 옮기고 안 지움
   ④ 지문 어긋난(묵은) 벌은 버리고 그 자리에서 지음 · 실패면 통째 되돌림
+  ⑤ 벌 예정노선(765-8) = 받은 노선을 그 벌 지형으로 자름 · 고친 예정노선과 함께 벌 몫 · 초기
+     폴리라인 바탕 · 공유 예상노선은 설계 지형 하나(추천 열쇠)
 """
 
 from __future__ import annotations
@@ -17,6 +19,7 @@ import re
 import sqlite3
 import sys
 from pathlib import Path
+from types import SimpleNamespace
 from uuid import uuid4
 
 import pytest
@@ -162,6 +165,8 @@ def _seed_db() -> sqlite3.Connection:
 
 _SET_FILES = {
     "B05_Profile/route/planned_route.csv": "x,y\n1,2\n",
+    "B05_Profile/route/terrain_route.csv": "t",
+    "B05_Profile/route/expected_route_edit.csv": "e2",
     "B05_Profile/route/structures.json": '{"items": [1]}',
     "B05_Profile/route/route_main.geojson": "{}",
     "B05_Profile/corridor/corridor_0011.json": '{"route": 11}',
@@ -249,8 +254,11 @@ def test_벌_몫과_공유를_가름(tmp_path: Path) -> None:
     assert "B05_Profile/route/structures.json" in entries
     assert "B04_PreProcess/drainage/edits" in entries
     joined = " ".join(entries)
-    for shared in ("cost_surface_", "expected_route", "initial_choice", "descent_", "00_sheet"):
+    for shared in ("cost_surface_", "expected_route.csv", "initial_choice", "descent_", "00_sheet"):
         assert shared not in joined
+    # 벌 예정노선 · 고친 예정노선 = 벌 몫(765-8)
+    assert "B05_Profile/route/terrain_route.csv" in entries
+    assert "B05_Profile/route/expected_route_edit.csv" in entries
     assert "rainfall_" not in joined and "models" not in joined
 
 
@@ -296,7 +304,7 @@ def test_지문_어긋나면_묵은_벌(tmp_path: Path) -> None:
     assert sets.bundle_ready(tmp_path, "las", FP)
     assert sets.bundle_ready(tmp_path, "las", {**FP, "model_id": 99})  # 복제 · 갈음 id 는 안 봄
     assert not sets.bundle_ready(tmp_path, "las", {**FP, "model_digest": "bb"})
-    assert not sets.bundle_ready(tmp_path, "las", {**FP, "source_filter": "csf"})
+    assert not sets.bundle_ready(tmp_path, "las", {**FP, "source_filter": "ptd"})
     assert not sets.bundle_ready(tmp_path, "las", None)
     assert not sets.bundle_ready(tmp_path, "sheet", {**FP, "terrain": "sheet"})
     sets.note_unbuilt(tmp_path, "sheet", "수치지형도 모델 없음")
@@ -603,3 +611,47 @@ async def _save(snap, root: Path) -> None:
     (root / "initial_snapshot").rename(root / "_old_snapshot")  # 시드 사본 비켜 둠
     db = _seed_db()
     await snap.save_initial_snapshot(_Conn(db), root, 11)
+
+
+def test_벌_예정노선_그_지형으로_잘라_한_번_적음(monkeypatch, tmp_path: Path) -> None:
+    from B03_FileInput import B03_FileInput_Service_Chain as chain
+    from common_util import common_util_route_geometry as geometry
+
+    calls: list[str] = []
+
+    def load(_root, surface, _range, *, from_source=False):
+        calls.append(f"{surface['source_filter']}:{from_source}")
+        n = 5 if surface["source_filter"] == "sheet_laplace" else 3  # 수치지형도 = 끝까지
+        return SimpleNamespace(vertices=[SimpleNamespace(x=float(i), y=0.0) for i in range(n)])
+
+    monkeypatch.setattr(geometry, "load_design_route", load)
+    geometry.write_route_csv(geometry.expected_route_csv_path(tmp_path), [{"x": 0, "y": 0}] * 2)
+    sheet = {"source_filter": "sheet_laplace", "method": "dtm", "smooth": True}
+    points = chain._terrain_route_points(tmp_path, sheet, None)
+    assert len(points) == 5 and calls == ["sheet_laplace:True"]  # 공유 예상노선을 안 씀
+    own = geometry.terrain_route_csv_path(tmp_path)
+    assert len(geometry.read_planned_route_csv(own).vertices) == 5
+    assert geometry.route_source_path(tmp_path) == own
+    las = {"source_filter": "classification", "method": "implicit", "smooth": False}
+    assert len(chain._terrain_route_points(tmp_path, las, None)) == 5  # 있으면 그것(다시 안 자름)
+    assert calls == ["sheet_laplace:True"]
+
+
+def test_초기_폴리라인_바탕은_벌_예정노선(tmp_path: Path) -> None:
+    import os
+
+    from B05_Profile import B05_Profile_Router_Replan as replan
+    from common_util import common_util_route_geometry as geometry
+
+    own = [{"x": float(x), "y": 0.0} for x in range(0, 401, 10)]
+    geometry.write_route_csv(geometry.terrain_route_csv_path(tmp_path), own)
+    assert replan._ensure_planned_initial(tmp_path, 30.0) is not None
+    initial = geometry.planned_route_initial_path(tmp_path)
+    vertices = geometry.read_planned_route_csv(initial).vertices
+    assert (vertices[0].x, vertices[-1].x) == (0.0, 400.0)
+    # 공유 예상노선(다른 지형으로 자름)이 나중에 적혀도 벌 초기 폴리라인은 그대로
+    expected = geometry.expected_route_csv_path(tmp_path)
+    geometry.write_route_csv(expected, own[:10])
+    later = initial.stat().st_mtime + 5
+    os.utime(expected, (later, later))
+    assert replan._ensure_planned_initial(tmp_path, 30.0) is None