feat(B05): 계획노선 두 벌 + 노선 갈아 끼우기·초기화 서버 경로

노선을 예상노선(원본)·계획노선(수정본) 두 벌로 나눔 (PLAN 0-7 사용자 확정).

- 예상노선 정본 `B05_Profile/route/expected_route.csv` 신설 — 자동 체인이 한 번
  씀. 초기값 스냅샷 안에도 같은 CSV 가 있으나 그 폴더는 재확정 체인이 지우므로
  스냅샷 밖에 한 벌 둠.
- 계획노선 수정본 `B05_Profile/route/planned_route.csv` — 설계 계통(`load_design_route`)이
  수정본 → 예상노선 → 스냅샷 순으로 읽음. 노선 초기화는 수정본을 지우는 것.
- `GET /route/plan` 두 노선 정점 반환(모달이 점선·실선으로 그림).
- `POST /route/replan` 고친 노선을 수정본에 쓰고(조밀화) 재확정 체인 재사용 —
  배수유역 다시 → 관 새로 → 계획선·종횡단 재생성 → 옛 측점 설계 누가거리 이월.
- `POST /route/replan/reset` 수정본 삭제 후 같은 재계산.
- 횡단배수 지점은 노선이 바뀌면 저장분을 버리고 새로 계산(사용자 확정).
  구조물은 프로젝트 정본이라 옛 측점값 그대로 남음.

Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>

B03_FileInput/B03_FileInput_Service_Chain.py

@@ -148,6 +148,10 @@ async def run_auto_design_chain(
         mark_designing,
         save_initial_snapshot,
     )
+    from common_util.common_util_route_geometry import (
+        expected_route_csv_path,
+        write_route_csv,
+    )
     from common_util.common_util_storage import resolve_stored_project_path
     from common_util.common_util_surface_confirmation import get_surface_confirmation_params
     from config.config_db import get_db_pool
@@ -195,6 +199,12 @@ async def run_auto_design_chain(
             mark_design_failed(project_root, "계획노선이 없어 초기 노선을 세울 수 없습니다.")
             return None
 
+        # 2.5) 예상노선(원본) 정본을 남긴다 — 초기값 스냅샷과 달리 재확정 체인이 지우지
+        #      않는 자리라, 노선 초기화가 언제나 이 값으로 돌아갈 수 있다(2026-09-06 PLAN 0-7).
+        expected_path = expected_route_csv_path(project_root)
+        if not expected_path.is_file():
+            write_route_csv(expected_path, points)
+
         # 3) B05 경로 계산
         request = RouteSolveRequest(
             filter_key=str(defaults["source_filter"]),

B05_Profile/B05_Profile_Router_Replan.py

@@ -0,0 +1,301 @@
+"""계획노선 편집 — 노선 두 벌(예상노선·계획노선)과 그 뒤 재계산의 서버 몫.
+
+노선은 두 벌이다(2026-09-06 사용자 확정, PLAN 0-7).
+
+  - **예상노선**(원본) — 파일 업로드 자동 체인이 낸 노선. `initial_snapshot/planned_route.csv`.
+    어떤 경로로도 고치지 않는다.
+  - **계획노선**(수정본) — 예상노선과 **같은 값으로 시작**해 사용자가 고쳐 쓰는 노선.
+    `B05_Profile/route/planned_route.csv`. 설계 계통은 이것이 있으면 이것을 읽는다
+    (`load_design_route`).
+
+노선 초기화는 수정본 파일을 지우는 것이다 — 그러면 원본을 읽으므로 「원본을 수정본으로
+복사」와 결과가 같다.
+
+[확인]을 눌렀을 때만 계산이 돈다. 재계산은 **초기 업로드 체인의 로직을 그대로 재사용**한다
+(2026-09-06 사용자 제안) — `run_redesign_chain`이 배수유역 다시 분석 → 기본 관 저장 → 관
+정착 계획선 재산출 → 종횡단 재생성 → 옛 측점 설계를 누가거리로 이월 → B05·B06 확정까지 한
+줄로 돈다. 여기서 하는 일은 그 앞에 **노선을 갈아 끼우는 것**뿐이다.
+
+시설 처리(2026-09-06 사용자 확정) — 횡단배수 지점은 노선 자리에 따라 생기고 없어지므로
+저장분을 버리고 **새로 계산한 값**을 쓴다. 그 밖의 구조물은 **옛 측점값을 그대로** 이어받는다
+(`structures.json`은 프로젝트 정본이라 손대지 않으면 그대로 남는다).
+
+  GET  /api/projects/{id}/route/plan         → 예상노선·계획노선 정점(사업지 좌표계)
+  POST /api/projects/{id}/route/replan       → 고친 계획노선으로 갈아 끼우고 재계산
+  POST /api/projects/{id}/route/replan/reset → 계획노선을 예상노선으로 되돌리고 재계산
+"""
+
+import asyncio
+import csv
+import logging
+from pathlib import Path
+from typing import Any
+from uuid import UUID
+
+from fastapi import APIRouter
+from fastapi.responses import JSONResponse
+from pydantic import BaseModel, Field
+
+from B03_FileInput.B03_FileInput_Repository import get_project_storage_relative_path
+from common_util.common_util_initial_snapshot import design_route_csv_path
+from common_util.common_util_route_geometry import (
+    densify_route,
+    expected_route_csv_path,
+    planned_route_working_path,
+    read_planned_route_csv,
+    write_route_csv,
+)
+from common_util.common_util_storage import resolve_stored_project_path
+from config.config_db import get_db_pool
+from config.config_system import (
+    ROUTE_DIRECT_LINK_CELL_FACTOR,
+    ROUTE_GRID_RES_M,
+    ROUTE_PLANNED_DENSIFY_SAFETY,
+)
+
+logger = logging.getLogger(__name__)
+router = APIRouter(prefix="/api/projects", tags=["B05 Route Design"])
+
+_PROJECT_PATH_MISSING = {"status": "error", "message": "프로젝트 저장 경로를 찾을 수 없습니다."}
+
+
+class RouteVertexInput(BaseModel):
+    """계획노선 정점 하나 — 사업지 좌표계(m)."""
+
+    x: float
+    y: float
+
+
+class RouteReplanRequest(BaseModel):
+    """고친 계획노선. 정점은 시점 → 종점 순서."""
+
+    vertices: list[RouteVertexInput] = Field(default_factory=list)
+
+
+async def _project_paths(project_id: UUID) -> tuple[Path, str] | None:
+    """(프로젝트 실경로, 저장소 상대경로). 없으면 None."""
+    pool = get_db_pool()
+    async with pool.acquire() as connection:
+        stored = await get_project_storage_relative_path(connection, project_id)
+    if not stored:
+        return None
+    return Path(resolve_stored_project_path(stored)), str(stored)
+
+
+def _vertices_of(path: Path) -> list[list[float]]:
+    """CSV 정점을 [[x, y], …]로. 없거나 못 읽으면 빈 목록."""
+    if not path.is_file():
+        return []
+    route = read_planned_route_csv(path)
+    if route is None:
+        return []
+    return [[float(vertex.x), float(vertex.y)] for vertex in route.vertices]
+
+
+def _ensure_expected_route(project_root: Path) -> str:
+    """예상노선(원본) 정본이 없으면 만들어 둔다. 어디서 씨앗을 얻었는지 돌려준다.
+
+    새 프로젝트는 자동 체인이 만들어 두지만(2026-09-06), 그 전에 만들어진 프로젝트는
+    초기값 스냅샷 안에만 있거나 그마저 재확정 체인이 지운 뒤일 수 있다. 노선 초기화가
+    성립하려면 이 파일이 반드시 있어야 하므로 여기서 한 번 세워 둔다.
+    """
+    target = expected_route_csv_path(project_root)
+    if target.is_file():
+        return "already"
+    for source, label in (
+        (design_route_csv_path(project_root), "snapshot"),
+        (planned_route_working_path(project_root), "working"),
+    ):
+        vertices = _vertices_of(source)
+        if len(vertices) >= 2:
+            write_route_csv(target, [{"x": x, "y": y} for x, y in vertices])
+            logger.info("예상노선 정본을 %s 에서 세웠습니다: %s", label, target)
+            return label
+    return "none"
+
+
+def _write_working_route(path: Path, vertices: list[tuple[float, float]]) -> int:
+    """계획노선(수정본) CSV를 쓴다. 열 이름은 `read_planned_route_csv()`가 아는 것.
+
+    쓰기 전에 **조밀화**한다 — 정점이 성기면 B05 격자 탐색이 원좌표를 그대로 잇지 못하고
+    제 나름의 길을 찾아 사용자가 그린 선과 달라진다(`densify_route` 주석). 평면 형상은
+    바뀌지 않고 같은 선 위에 점만 더 찍힌다.
+    """
+    dense = densify_route(
+        vertices,
+        ROUTE_DIRECT_LINK_CELL_FACTOR * ROUTE_GRID_RES_M * ROUTE_PLANNED_DENSIFY_SAFETY,
+    )
+    path.parent.mkdir(parents=True, exist_ok=True)
+    with path.open("w", encoding="utf-8", newline="") as file:
+        writer = csv.writer(file)
+        writer.writerow(("sequence", "x", "y"))
+        writer.writerows((index, round(x, 4), round(y, 4)) for index, (x, y) in enumerate(dense))
+    return len(dense)
+
+
+async def _recompute(project_id: UUID, project_root: Path, stored_path: str) -> dict[str, Any]:
+    """노선을 갈아 끼운 뒤의 재계산 — 초기 업로드 체인과 같은 로직을 그대로 탄다."""
+    import aiomysql
+
+    from B03_FileInput.B03_FileInput_Service_Chain import (
+        _planned_route_points_in_project_crs,
+        run_redesign_chain,
+    )
+    from B04_PreProcess.B04_PreProcess_Service import find_surface_model_for_selection
+    from B05_Profile.B05_Profile_Repository import get_latest_route
+    from common_util.common_util_drainage_pipes import clear_pipe_points
+    from common_util.common_util_surface_confirmation import get_surface_confirmation_params
+    from common_util.common_util_workflow_state import get_workflow_state, start_stage
+
+    # 횡단배수 지점은 새 노선에서 새로 계산한다 — 저장분을 남기면 옛 자리가 살아난다.
+    await asyncio.to_thread(clear_pipe_points, stored_path)
+
+    pool = get_db_pool()
+    async with pool.acquire() as connection:
+        selection = await get_surface_confirmation_params(connection, str(project_id))
+        latest = await get_latest_route(connection, project_id)
+        # 지금 노선이 쓰던 지표면 모델이 1순위다 — 확정 선택값으로 다시 찾으면 프로젝트마다
+        # 모델 조합이 달라 못 찾는 경우가 있다(용화: filter=classification/method=dtm 무매칭).
+        surface_model_id = (
+            int(latest["surface_model_id"]) if latest and latest.get("surface_model_id") else None
+        )
+        if surface_model_id is None:
+            try:
+                surface_model_id = await find_surface_model_for_selection(
+                    connection, project_id, selection
+                )
+            except Exception:
+                surface_model_id = None
+        async with connection.cursor() as cursor:
+            await cursor.execute(
+                "SELECT route_start_m, route_end_m FROM projects WHERE id = %s",
+                (str(project_id),),
+            )
+            range_row = await cursor.fetchone()
+    if surface_model_id is None:
+        return {"error": "확정된 지표면 모델을 찾지 못해 노선을 다시 계산할 수 없습니다."}
+    route_range = (range_row[0], range_row[1]) if range_row else None
+
+    # 갈아 끼운 노선을 stage 2 제어점으로 세운다 — 재확정 체인이 여기서 BP·EP·경유점을
+    # 읽어 경로를 다시 푼다. 읽기는 설계 계통과 같은 한 곳(`load_design_route`)을 지난다.
+    points = await asyncio.to_thread(
+        _planned_route_points_in_project_crs, project_root, selection, route_range
+    )
+    if not points or len(points) < 2:
+        return {"error": "계획노선을 읽지 못했습니다(정점이 2개 미만)."}
+
+    # 수정본도 예상노선 정본도 없던 프로젝트 — 지금 읽은 것이 곧 예상노선이므로 여기서 세운다.
+    expected_path = expected_route_csv_path(project_root)
+    if not expected_path.is_file() and not planned_route_working_path(project_root).is_file():
+        await asyncio.to_thread(write_route_csv, expected_path, points)
+        logger.info("예상노선 정본을 원본 재판독으로 세웠습니다: %s", expected_path)
+
+    pool = get_db_pool()
+    async with pool.acquire() as connection:
+        async with connection.cursor(aiomysql.DictCursor) as cursor:
+            state = await get_workflow_state(cursor, str(project_id))
+        stage2 = next(
+            (s for s in (state or {}).get("stages", []) if int(s.get("stage_no", -1)) == 2), None
+        )
+        params = dict((stage2 or {}).get("params") or {})
+        params["points"] = {
+            "bp": points[0],
+            "ep": points[-1],
+            "cp": [{**point, "order": index} for index, point in enumerate(points[1:-1], start=1)],
+        }
+        async with connection.cursor(aiomysql.DictCursor) as cursor:
+            await start_stage(cursor, str(project_id), 2, params)
+        await connection.commit()
+
+    await run_redesign_chain(project_id, int(surface_model_id), selection)
+
+    pool = get_db_pool()
+    async with pool.acquire() as connection:
+        made = await get_latest_route(connection, project_id)
+    return {
+        "route_id": int(made["id"]) if made else None,
+        "total_length_m": float(made.get("total_length_m") or 0.0) if made else None,
+        "vertex_count": len(points),
+    }
+
+
+@router.get("/{project_id}/route/plan", response_model=None)
+async def read_route_plan(project_id: UUID) -> dict[str, Any] | JSONResponse:
+    """예상노선(원본)과 계획노선(수정본) 정점을 함께 돌려준다 — 편집 모달이 둘 다 그린다."""
+    paths = await _project_paths(project_id)
+    if paths is None:
+        return JSONResponse(status_code=404, content=_PROJECT_PATH_MISSING)
+    project_root, _ = paths
+    # 예상노선 정본이 먼저, 없으면 옛 자리(초기값 스냅샷)를 본다.
+    expected = await asyncio.to_thread(_vertices_of, expected_route_csv_path(project_root))
+    if not expected:
+        expected = await asyncio.to_thread(_vertices_of, design_route_csv_path(project_root))
+    working_path = planned_route_working_path(project_root)
+    planned = await asyncio.to_thread(_vertices_of, working_path)
+    return {
+        "status": "success",
+        "project_id": str(project_id),
+        "expected": expected,
+        # 고친 적이 없으면 계획노선 = 예상노선(같은 값으로 시작한다).
+        "planned": planned or expected,
+        "edited": bool(planned),
+    }
+
+
+@router.post("/{project_id}/route/replan", response_model=None)
+async def replan_route(
+    project_id: UUID, request: RouteReplanRequest
+) -> dict[str, Any] | JSONResponse:
+    """고친 계획노선으로 갈아 끼우고 배수유역부터 다시 계산한다(모달 [확인])."""
+    if len(request.vertices) < 2:
+        return JSONResponse(
+            status_code=400,
+            content={"status": "error", "message": "계획노선은 정점이 2개 이상이어야 합니다."},
+        )
+    paths = await _project_paths(project_id)
+    if paths is None:
+        return JSONResponse(status_code=404, content=_PROJECT_PATH_MISSING)
+    project_root, stored_path = paths
+
+    # 고치기 전에 예상노선(원본)이 서 있는지 본다 — 초기화가 돌아갈 자리다.
+    await asyncio.to_thread(_ensure_expected_route, project_root)
+    written = await asyncio.to_thread(
+        _write_working_route,
+        planned_route_working_path(project_root),
+        [(vertex.x, vertex.y) for vertex in request.vertices],
+    )
+    logger.info(
+        "계획노선 갈아 끼움: project_id=%s 정점 %d→%d(조밀화)",
+        project_id,
+        len(request.vertices),
+        written,
+    )
+    result = await _recompute(project_id, project_root, stored_path)
+    if "error" in result:
+        return JSONResponse(
+            status_code=409, content={"status": "error", "message": result["error"]}
+        )
+    return {"status": "success", "project_id": str(project_id), **result}
+
+
+@router.post("/{project_id}/route/replan/reset", response_model=None)
+async def reset_route_plan(project_id: UUID) -> dict[str, Any] | JSONResponse:
+    """계획노선을 예상노선으로 되돌리고 다시 계산한다(노선 초기화).
+
+    수정본 파일을 지우면 설계 계통이 원본을 읽으므로 「원본을 수정본으로 복사」와 같다.
+    """
+    paths = await _project_paths(project_id)
+    if paths is None:
+        return JSONResponse(status_code=404, content=_PROJECT_PATH_MISSING)
+    project_root, stored_path = paths
+    await asyncio.to_thread(_ensure_expected_route, project_root)
+    working_path = planned_route_working_path(project_root)
+    if working_path.is_file():
+        working_path.unlink()
+        logger.info("계획노선 초기화(예상노선으로): project_id=%s", project_id)
+    result = await _recompute(project_id, project_root, stored_path)
+    if "error" in result:
+        return JSONResponse(
+            status_code=409, content={"status": "error", "message": result["error"]}
+        )
+    return {"status": "success", "project_id": str(project_id), **result}

common_util/common_util_route_geometry.py

@@ -212,6 +212,40 @@ def find_planned_route_file(input_dir: Path) -> Path | None:
     return None
 
 
+def expected_route_csv_path(project_root: Path) -> Path:
+    """예상노선(원본) CSV 자리 — 자동 체인이 낸 노선을 **그대로** 보관한다.
+
+    초기값 스냅샷(`initial_snapshot/`) 안에도 같은 CSV가 있지만 그 폴더는 재확정 체인이
+    통째로 지운다(`discard_initial_snapshot`). 노선 초기화는 언제나 예상노선으로 돌아갈 수
+    있어야 하므로 스냅샷 **밖**에 한 벌 둔다(2026-09-06).
+    """
+    return Path(project_root) / "B05_Profile" / "route" / "expected_route.csv"
+
+
+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)
+    with path.open("w", encoding="utf-8", newline="") as file:
+        writer = csv.writer(file)
+        writer.writerow(("sequence", "x", "y"))
+        writer.writerows(
+            (index, round(float(point["x"]), 4), round(float(point["y"]), 4))
+            for index, point in enumerate(points)
+        )
+
+
+def planned_route_working_path(project_root: Path) -> Path:
+    """계획노선(수정본) CSV 자리 — 사용자가 노선을 고치면 여기에 쓴다.
+
+    노선은 두 벌이다(2026-09-06 사용자 확정): **예상노선**(원본,
+    `initial_snapshot/planned_route.csv`)은 자동 체인이 한 번 쓰고 안 바뀌며,
+    **계획노선**(수정본)은 예상노선과 같은 값으로 시작해 사용자가 고쳐 쓴다.
+    설계 계통은 수정본이 있으면 그것을 읽는다 — 노선 초기화는 이 파일을 지우는 것이며,
+    그것이 곧 「원본을 수정본으로 복사」와 같다.
+    """
+    return Path(project_root) / "B05_Profile" / "route" / "planned_route.csv"
+
+
 def load_design_route(
     project_root: Path,
     surface_params: dict[str, Any] | None = None,
@@ -244,9 +278,15 @@ def load_design_route(
     # 지표면·노선이 바뀌면 `discard_initial_snapshot()`이 폴더째 지우므로 이 경로가 저절로
     # 닫히고 원본 재판독으로 되돌아간다. 트림 **전** 원본이 필요한 호출(도엽 범위 —
     # surface_params 없음)은 여기를 타지 않는다.
+    # 읽는 순서 — 계획노선(수정본) → 예상노선(원본) → 초기값 스냅샷(옛 자리).
     if surface_params:
-        master = design_route_csv_path(project_root)
-        if master.is_file():
+        for master in (
+            planned_route_working_path(project_root),
+            expected_route_csv_path(project_root),
+            design_route_csv_path(project_root),
+        ):
+            if not master.is_file():
+                continue
             stored = read_planned_route_csv(master)
             if stored is not None and len(stored.vertices) >= 2:
                 return replace_vertices(

main.py

@@ -43,6 +43,7 @@ from B04_PreProcess.B04_PreProcess_Router_Watershed import router as b04_watersh
 from B05_Profile.B05_Profile_Router import router as b05_route_router
 from B05_Profile.B05_Profile_Router_Corridor import router as b05_corridor_router
 from B05_Profile.B05_Profile_Router_Lifecycle import router as b05_route_lifecycle_router
+from B05_Profile.B05_Profile_Router_Replan import router as b05_route_replan_router
 from B05_Profile.B05_Profile_Structures_Router import router as b05_structures_router
 from B06_Section.B06_Section_Router import router as b06_section_router
 from B06_Section.B06_Section_Router_Confirm import (
@@ -418,6 +419,7 @@ app.include_router(tiles_router, dependencies=protected_with_company)
 app.include_router(b05_route_router, dependencies=protected_with_company)
 app.include_router(b05_route_lifecycle_router, dependencies=protected_with_company)
 app.include_router(b05_corridor_router, dependencies=protected_with_company)
+app.include_router(b05_route_replan_router, dependencies=protected_with_company)
 app.include_router(b05_structures_router, dependencies=protected_with_company)
 app.include_router(b06_section_router, dependencies=protected_with_company)
 app.include_router(b06_section_confirm_router, dependencies=protected_with_company)