feat(B05): 노선 [확인] 때 고른 분석 지형 = 설계 지형 — 종 · 횡단 · 배수 · 유토 · 3D 전부 그 지형으로(133-2)

- [확인] 이 고른 지형으로 확정 모델 · stage 1 params 를 바꾼 뒤 재계산 체인을 돎(switch_design_terrain · B04 확정 표시도 따라감) · 체인 실패면 노선과 함께 되돌림
- 수치지형도로 갈 때 라이다 확정값을 stage 1 params 에 적어 둠 — 라이다로 되돌려 [확인]하면 원래 모델 · params
- 지형만 바꿔도 [확인]이 다시 셈 · 고르기 안 보이면(라이다 없는 프로젝트) 지형 안 실음 · 고른 값 없으면 지금 설계 지형 · 상태줄에 설계 지형 표시
- 분석 통로 terrain 기본 = 지금 설계 지형(None) · terrain-sources 가 design 을 함께 냄
- B06 구조물 측점 추가가 노선 확정 때 남긴 sampling.json 이 지금 지형과 다르면 stage 1 params 로 뜸(B04 재확정 때도 옛 지형으로 뜨던 흠)

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01UoSr2sqvYtBf1dtQdus88s

B04_PreProcess/B04_PreProcess_Service.py

@@ -10,7 +10,13 @@ from B04_PreProcess.B04_PreProcess_Repository import (
     list_surface_models,
 )
 from common_util.common_util_surface_confirmation import (
+    LAS_STASH_KEY,
+    AnalysisTerrain,
+    analysis_terrain_params,
+    design_terrain,
+    get_surface_confirmation_params,
     merge_surface_confirmation_params,
+    terrain_changes,
 )
 from common_util.common_util_workflow_state import complete_stage
 
@@ -52,3 +58,30 @@ async def confirm_surface_selection(
     async with connection.cursor() as cursor:
         await complete_stage(cursor, str(project_id), 1)
     return normalized
+
+
+async def switch_design_terrain(
+    connection: aiomysql.Connection,
+    project_id: UUID,
+    terrain: AnalysisTerrain | None,
+) -> tuple[int, AnalysisTerrain] | None:
+    """노선 [확인] 때 고른 분석 지형을 설계 지형으로(PLAN 133-2) — 확정 모델 · stage 1 params.
+
+    뒤 단계(종 · 횡단 · 배수 · 유토 · 3D)는 모두 stage 1 params · 확정 모델을 읽으므로 여기 한 곳만
+    바꾼다. 수치지형도로 갈 때 라이다 확정값을 적어 둠(되돌릴 자리). WF1 완료 전이는 안 건드림.
+    바뀌면 (새 확정 모델 id, 이전 설계 지형) · 그대로면 None. 트랜잭션은 부르는 쪽.
+    """
+    params = await get_surface_confirmation_params(connection, str(project_id))
+    target = analysis_terrain_params(params, terrain)
+    if not terrain_changes(params, target):
+        return None
+    model_id = await find_surface_model_for_selection(connection, project_id, target)
+    await confirm_surface_model(connection, project_id, model_id)
+    previous = design_terrain(params)
+    extra = (
+        {LAS_STASH_KEY: {key: params[key] for key in ("source_filter", "method", "smooth")}}
+        if previous == "las"
+        else None
+    )
+    await merge_surface_confirmation_params(connection, str(project_id), target, extra)
+    return model_id, previous

B05_Profile/B05_Profile_Api_Replan.ts

@@ -21,21 +21,29 @@ import { readState, writeState } from "../A00_Common/b_page_state";
 /** 편집 창 분석 지형 — las = 확정 지표면(라이다) · sheet = 수치지형도 서피스. */
 export type AnalysisTerrain = "las" | "sheet";
 
-/** 고른 분석 지형 — 화면 상태 보관소 pref 통(노선 편집 고정과 같은 자리) · 기본 las. */
-export const analysisTerrain = (projectId: string): AnalysisTerrain =>
-  readState<string>("routeedit-terrain", projectId) === "sheet" ? "sheet" : "las";
+/** 고른 분석 지형 — 화면 상태 보관소 pref 통(노선 편집 고정과 같은 자리). 없으면 `fallback`
+ *  (모달이 열 때 설계 지형으로 채움). [확인] 때 이 값이 설계 지형이 됨(PLAN 133-2). */
+export const analysisTerrain = (
+  projectId: string,
+  fallback: AnalysisTerrain = "las",
+): AnalysisTerrain => {
+  const saved = readState<string>("routeedit-terrain", projectId);
+  return saved === "sheet" || saved === "las" ? saved : fallback;
+};
 
 export const setAnalysisTerrain = (projectId: string, terrain: AnalysisTerrain): void =>
   writeState("routeedit-terrain", terrain, projectId);
 
-/** 이 프로젝트가 고를 수 있는 분석 지형 — 둘이면 고르기를 띄운다. */
-export async function fetchTerrainSources(projectId: string): Promise<AnalysisTerrain[]> {
-  const payload = await requestJson<{ sources: AnalysisTerrain[] }>(
+/** 이 프로젝트가 고를 수 있는 분석 지형(둘이면 고르기를 띄운다)과 지금 설계 지형. */
+export async function fetchTerrainSources(
+  projectId: string,
+): Promise<{ sources: AnalysisTerrain[]; design: AnalysisTerrain }> {
+  const payload = await requestJson<{ sources?: AnalysisTerrain[]; design?: AnalysisTerrain }>(
     `/projects/${projectId}/route/terrain-sources`,
     { method: "GET" },
     60000,
   );
-  return payload.sources ?? [];
+  return { sources: payload.sources ?? [], design: payload.design === "sheet" ? "sheet" : "las" };
 }
 
 /** 노선 재계산 대기 상한 — 배수유역 분석(90초대)까지 포함해 넉넉히 잡는다. */

B05_Profile/B05_Profile_Router_Replan.py

@@ -120,8 +120,8 @@ class RouteInitialRequest(BaseModel):
     criteria: RouteInitialCriteria | None = None
     # 추천 점수에 균형 종단(PLAN 63-4)을 넣을지 — 끄면 원지반만(63-4 전 점수). 파일에 안 적힘.
     with_profile: bool = True
-    # 분석 지형(PLAN 133-1) — 추천만 고른 지형으로 · [확인] 재계산은 확정 지표면 그대로.
-    terrain: Literal["las", "sheet"] = "las"
+    # 분석 지형(PLAN 133-1) — None = 지금 설계 지형 · [확인] 때 설계 지형이 됨(133-2).
+    terrain: Literal["las", "sheet"] | None = None
 
 
 class RouteReplanRequest(BaseModel):
@@ -133,6 +133,8 @@ class RouteReplanRequest(BaseModel):
     balanced_profile: bool = False
     """「종단 반영」 켬으로 [적용]한 추천 그대로인가 — 켜면 서버가 균형 종단을 다시 셈해
     1차 종단으로 쓴다(PLAN 74-4). 값은 받지 않는다 — 켬/끔 표시만."""
+    terrain: Literal["las", "sheet"] | None = None
+    """편집 창에서 고른 분석 지형 = 설계 지형(PLAN 133-2). None 이면 지금 설계 지형 그대로."""
 
 
 async def _project_paths(project_id: UUID) -> tuple[Path, str] | None:
@@ -385,6 +387,7 @@ async def _recompute(
     project_root: Path,
     stored_path: str,
     balanced: RouteInitialRequest | None = None,
+    terrain: Literal["las", "sheet"] | None = None,
 ) -> dict[str, Any]:
     """노선을 갈아 끼운 뒤의 재계산 — 초기 업로드 체인과 같은 로직을 그대로 탄다.
 
@@ -397,7 +400,10 @@ async def _recompute(
         _planned_route_points_in_project_crs,
         run_redesign_chain,
     )
-    from B04_PreProcess.B04_PreProcess_Service import find_surface_model_for_selection
+    from B04_PreProcess.B04_PreProcess_Service import (
+        find_surface_model_for_selection,
+        switch_design_terrain,
+    )
     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
@@ -410,6 +416,9 @@ async def _recompute(
 
     pool = get_db_pool()
     async with pool.acquire() as connection:
+        await connection.begin()  # 고른 지형 = 설계 지형 — 뒤 단계가 모두 이 확정값을 읽음
+        switched = await switch_design_terrain(connection, project_id, terrain)
+        await connection.commit()
         selection = await get_surface_confirmation_params(connection, str(project_id))
         latest = await get_latest_route(connection, project_id)
         # 지금 노선이 쓰던 지표면 모델이 1순위다 — 확정 선택값으로 다시 찾으면 프로젝트마다
@@ -417,6 +426,8 @@ async def _recompute(
         surface_model_id = (
             int(latest["surface_model_id"]) if latest and latest.get("surface_model_id") else None
         )
+        if switched:
+            surface_model_id = switched[0]
         if surface_model_id is None:
             try:
                 surface_model_id = await find_surface_model_for_selection(
@@ -475,6 +486,11 @@ async def _recompute(
     failure = await run_redesign_chain(project_id, int(surface_model_id), selection, target)
     marks.append(("재확정 체인", time.perf_counter()))
     _log_steps("노선 [확인] 재계산", marks)
+    if failure and switched:  # 노선과 함께 설계 지형도 되돌림
+        async with pool.acquire() as connection:
+            await connection.begin()
+            await switch_design_terrain(connection, project_id, switched[1])
+            await connection.commit()
     if failure:
         # 체인이 끊기면 **노선만 새것으로 갈아 끼워진 채** 종횡단은 옛 노선에 남아 어긋난다
         # (2026-09-06 실측: 배수유역·관은 새 노선, 종횡단은 옛 노선 → 배수관 측점 9 → 0).
@@ -613,6 +629,7 @@ async def replan_route(
         project_root,
         stored_path,
         (request.initial or RouteInitialRequest()) if request.balanced_profile else None,
+        request.terrain,
     )
     if "error" in result:
         # 노선을 되돌린다 — 실패했는데 새 노선만 남으면 화면·정본이 어긋난 채 굳는다.

B05_Profile/B05_Profile_Router_RouteInitial.py

@@ -211,11 +211,11 @@ async def balanced_profile_of(
     points: list[dict[str, float]],
     criteria: Any,
     design_curves: list[dict[str, Any]] | None = None,
-    terrain: Literal["las", "sheet"] = "las",
+    terrain: Literal["las", "sheet"] | None = None,
 ) -> dict[str, Any] | None:
     """노선 정점의 균형 종단 한 벌(`route_profile`) — [확인] 체인 · 편집 창 종단 띠가 함께 씀.
 
-    `terrain` 은 편집 창 종단 띠만 바꿈(PLAN 133-1) — [확인] 체인은 기본(확정 지표면).
+    `terrain` 은 편집 창 종단 띠만 바꿈(PLAN 133-1) — None = 지금 설계 지형([확인] 체인).
     """
     from B05_Profile.B05_Profile_Engine_Grade import legal_grade_limit_pct
     from B05_Profile.B05_Profile_Engine_RouteInitial import load_terrain_and_streams
@@ -293,7 +293,7 @@ class ProfilePreviewRequest(BaseModel):
 
     vertices: list[replan.RouteVertexInput] = Field(..., min_length=2)
     criteria: replan.RouteInitialCriteria | None = None
-    terrain: Literal["las", "sheet"] = "las"
+    terrain: Literal["las", "sheet"] | None = None
 
 
 @replan.router.post("/{project_id}/route/profile-preview", response_model=None)

B05_Profile/B05_Profile_Router_Terrain.py

@@ -12,8 +12,8 @@
   POST /api/projects/{id}/route/cross-preview → 고치던 노선의 한 측점 횡단 미리보기
   GET  /api/projects/{id}/route/terrain-sources → 고를 수 있는 분석 지형(라이다 · 수치지형도)
 
-분석 지형(`terrain`, PLAN 133-1) — las = 확정 지표면 · sheet = 수치지형도 서피스. 모달 분석만
-고른 지형을 쓴다 — 확정 지표면 · 종 · 횡단 정본은 안 바뀐다.
+분석 지형(`terrain`, PLAN 133-1) — las = 라이다 · sheet = 수치지형도 · None = 지금 설계 지형.
+모달 분석만 고른 지형을 쓴다 — 설계 지형(확정)은 [확인] 때만 바뀜(133-2 · `/route/replan`).
 """
 
 import asyncio
@@ -38,6 +38,7 @@ from common_util.common_util_surface_confirmation import (
     AnalysisTerrain,
     analysis_terrain_params,
     analysis_terrain_sources,
+    design_terrain,
     get_surface_confirmation_params,
 )
 from common_util.common_util_surface_sampler import build_surface_sampler
@@ -56,7 +57,7 @@ class ElevationRequest(BaseModel):
     """사업지 좌표계(m) 점 묶음 [[x, y], …]."""
 
     points: list[tuple[float, float]] = Field(..., min_length=1, max_length=MAX_POINTS)
-    terrain: AnalysisTerrain = "las"
+    terrain: AnalysisTerrain | None = None
 
 
 def _sample(project_root: Path, params: dict, points: list[tuple[float, float]]):
@@ -131,7 +132,7 @@ class CrossPreviewRequest(BaseModel):
     station_interval_m: float | None = None
     #: 횡단 반폭(m) — 화면이 적용해 둔 값. 비면 서버 기본값.
     cross_half_width_m: float | None = Field(None, gt=0)
-    terrain: AnalysisTerrain = "las"
+    terrain: AnalysisTerrain | None = None
 
 
 def _cross_preview(
@@ -254,4 +255,9 @@ async def read_terrain_sources(project_id: UUID) -> dict | JSONResponse:
         params = await get_surface_confirmation_params(connection, str(project_id))
     models_dir = Path(resolve_stored_project_path(stored)) / _MODELS_SUBDIR
     sources = await asyncio.to_thread(analysis_terrain_sources, params, models_dir)
-    return {"status": "success", "project_id": str(project_id), "sources": sources}
+    return {
+        "status": "success",
+        "project_id": str(project_id),
+        "sources": sources,
+        "design": design_terrain(params),  # 지금 설계 지형(PLAN 133-2)
+    }

B05_Profile/B05_Profile_UI_RouteEdit.ts

@@ -143,6 +143,11 @@ export async function openRouteEditModal(
   let meta: VWorldMeta | null = null;
   /** 바탕 등고선 한 벌 — LAS 것이거나 도엽 것. 고르기는 `_Contour` 몫. */
   let contours: RouteEditContours | null = null;
+  /** 지금 설계 지형(PLAN 133-2) — 고른 분석 지형이 이와 다르면 [확인]이 설계 지형을 바꾼다. */
+  let designTerrain: AnalysisTerrain = "las";
+  /** [확인]에 실을 분석 지형 — 고르기가 안 보이면 null(설계 지형 그대로). */
+  const pickedTerrain = (): AnalysisTerrain | null =>
+    terrainSelect.closest("label")!.hidden ? null : analysisTerrain(projectId, designTerrain);
   /** 등고선 말고 함께 깔 도엽 레이어(하천중심선). */
   let otherSheets: PreparedLayer[] = [];
   /** 고른 등고선 가닥 — 없으면 -1(계획서 0-9 ⑦). */
@@ -524,7 +529,9 @@ export async function openRouteEditModal(
     busy,
     projectId,
     nodes: () => ({ planned, curveOn, curveRadius }),
-    changed: () => history?.isDirty() ?? false,
+    // 분석 지형만 바꿔도 [확인]이 설계 지형을 바꿔 다시 셈한다(PLAN 133-2).
+    changed: () => (history?.isDirty() ?? false) || pickedTerrain() !== designTerrain,
+    terrain: pickedTerrain,
     initial,
     profileShown: strip.settle,
     close,
@@ -536,7 +543,7 @@ export async function openRouteEditModal(
     const [plan, drainage, sources] = await Promise.all([
       fetchRoutePlan(projectId),
       fetchDrainageLayers(projectId, () => {}),
-      fetchTerrainSources(projectId).catch((): AnalysisTerrain[] => []),
+      fetchTerrainSources(projectId).catch(() => null),
     ]);
     if (closed) return;
     expected = plan.expected as Vertex[];
@@ -561,9 +568,14 @@ export async function openRouteEditModal(
     if (!planned.length) planned = plannedLine.map((vertex) => [vertex[0], vertex[1]]);
     meta = drainage.meta;
     // 분석 지형 고르기(PLAN 133-1) — 라이다 · 수치지형도 둘 다 있을 때만 보임 · 기본 라이다.
-    const choosable = sources.includes("las") && sources.includes("sheet");
+    // 고른 값이 없으면 지금 설계 지형(133-2) · 못 고르는 프로젝트는 설계 지형 그대로.
+    designTerrain = sources?.design ?? "las";
+    const choosable = !!sources?.sources.includes("las") && !!sources?.sources.includes("sheet");
     terrainSelect.closest("label")!.hidden = !choosable;
-    if (!choosable && analysisTerrain(projectId) === "sheet") setAnalysisTerrain(projectId, "las");
+    setAnalysisTerrain(
+      projectId,
+      choosable ? analysisTerrain(projectId, designTerrain) : designTerrain,
+    );
     terrainSelect.value = analysisTerrain(projectId);
     // 등고선은 따로 고른다(LAS 우선 · 수치지형도를 고르면 도엽). 나머지 도엽 레이어(하천중심선)만 배경으로 깐다.
     const loadBackground = () =>
@@ -587,6 +599,7 @@ export async function openRouteEditModal(
       status.textContent =
         `${routeHead()} · ${plan.edited ? "고친 계획노선" : "초기 폴리라인"} · ` +
         `${contours?.source === "las" ? "LAS 등고선" : "도엽 등고선"} · ` +
+        `설계 지형 ${designTerrain === "sheet" ? "수치지형도" : "라이다"} · ` +
         curveHint();
     };
     useBackground(await loadBackground());

B05_Profile/B05_Profile_UI_RouteEdit_Apply.ts

@@ -25,6 +25,8 @@ export interface RouteApplyParams {
   nodes: () => { planned: Vertex[]; curveOn: boolean[]; curveRadius: Array<number | null> };
   /** 창을 연 뒤 노선이 바뀌었나 — 안 바뀌었으면 다시 계산하지 않는다. */
   changed: () => boolean;
+  /** 설계 지형으로 보낼 분석 지형(PLAN 133-2) — 고르기가 안 보이면 null(지금 설계 지형 그대로). */
+  terrain: () => string | null;
   /** 초기노선 갈래 — [확인]에 지금 초기값을 싣고, 결과 캐시를 비운 뒤 다시 적는다. */
   initial: {
     current: () => unknown;
@@ -94,6 +96,7 @@ export function bindRouteApply(params: RouteApplyParams): void {
           initial: params.initial.current(),
           // 띠 응답 전에 눌렀으면 기다린다 — 서버는 같은 노선 · 같은 셈으로 띠 계획선을 다시 낸다.
           balanced_profile: await params.profileShown(),
+          terrain: params.terrain(),
         },
       ),
     );

B06_Section/B06_Section_Router_Stations.py

@@ -62,9 +62,15 @@ async def _sampling_conditions(project_id: UUID, project_root: Path) -> dict[str
     from common_util.common_util_surface_confirmation import get_surface_confirmation_params
 
     snapshot = load_sampling_snapshot(project_root)
-    if snapshot is not None:
-        return snapshot
     params = await run_with_connection(get_surface_confirmation_params, str(project_id))
+    surface = (params or {}).get("source_filter"), (params or {}).get("method")
+    # 설계 지형이 바뀌었으면(B04 재확정 · 노선 [확인] 지형 바꿈 PLAN 133-2) ①은 옛 지형 — 버림.
+    if snapshot is not None and (
+        not all(surface)
+        or (snapshot["filter_key"], snapshot["method"], bool(snapshot.get("smooth")))
+        == (*surface, bool(params.get("smooth")))
+    ):
+        return snapshot
     if not params or not params.get("source_filter") or not params.get("method"):
         return None
     return {

common_util/common_util_surface_confirmation.py

@@ -18,6 +18,9 @@ from config.config_system import (
 AnalysisTerrain = Literal["las", "sheet"]
 #: 수치지형도 분석 지형 — LAS 없는 프로젝트 자동 확정(WF1)과 같은 모델
 SHEET_ANALYSIS_FILTER = f"sheet_{SHEET_SURFACE_DEFAULT_METHOD}"
+#: 설계 지형을 수치지형도로 바꿀 때 적어 두는 라이다 확정값 — 라이다로 되돌릴 자리(PLAN 133-2)
+LAS_STASH_KEY = "las_selection"
+_SURFACE_KEYS = ("source_filter", "method", "smooth")
 
 SURFACE_CONFIRM_PARAM_KEYS = (
     "source_filter",
@@ -67,6 +70,8 @@ async def get_surface_confirmation_params(
     for key in SURFACE_CONFIRM_PARAM_KEYS:
         if key in params and params[key] is not None:
             resolved[key] = params[key]
+    if isinstance(params.get(LAS_STASH_KEY), dict):
+        resolved[LAS_STASH_KEY] = params[LAS_STASH_KEY]
     return resolved
 
 
@@ -74,17 +79,38 @@ def _is_sheet(params: dict[str, Any]) -> bool:
     return str(params.get("source_filter") or "").startswith("sheet_")
 
 
-def analysis_terrain_params(params: dict[str, Any], terrain: AnalysisTerrain) -> dict[str, Any]:
-    """편집 모달 분석에 쓸 지표면 params — 확정 params 는 안 고침(정본 · 종 · 횡단 그대로)."""
-    if terrain != "sheet" or _is_sheet(params):
+def design_terrain(params: dict[str, Any]) -> AnalysisTerrain:
+    """지금 설계 지형 — 확정 params 가 수치지형도면 sheet."""
+    return "sheet" if _is_sheet(params) else "las"
+
+
+def analysis_terrain_params(
+    params: dict[str, Any], terrain: AnalysisTerrain | None
+) -> dict[str, Any]:
+    """고른 지형의 지표면 params — None · 지금 설계 지형과 같으면 확정 params 그대로.
+
+    설계 지형이 수치지형도여도 적어 둔 라이다 확정값(`LAS_STASH_KEY`)이 있으면 las 로 돌아감.
+    라이다 없는 프로젝트는 무엇을 받아도 확정 params 그대로.
+    """
+    if terrain is None or terrain == design_terrain(params):
         return params
-    return {**params, "source_filter": SHEET_ANALYSIS_FILTER, "method": "dtm", "smooth": True}
+    if terrain == "sheet":
+        return {**params, "source_filter": SHEET_ANALYSIS_FILTER, "method": "dtm", "smooth": True}
+    stash = params.get(LAS_STASH_KEY)
+    if not isinstance(stash, dict) or not stash.get("source_filter"):
+        return params
+    return {**params, **{key: stash[key] for key in _SURFACE_KEYS if key in stash}}
+
+
+def terrain_changes(params: dict[str, Any], target: dict[str, Any]) -> bool:
+    """지표면(필터 · 표현 · 평활)이 바뀌는가."""
+    return any(params.get(key) != target.get(key) for key in _SURFACE_KEYS)
 
 
 def analysis_terrain_sources(params: dict[str, Any], models_dir: Path) -> list[str]:
-    """고를 수 있는 분석 지형 — 확정이 LAS 이고 수치지형도 서피스가 있을 때만 둘."""
+    """고를 수 있는 분석 지형 — 라이다 확정값(지금 것 · 적어 둔 것)과 수치지형도가 둘 다 있으면 둘."""
     if _is_sheet(params):
-        return ["sheet"]
+        return ["las", "sheet"] if isinstance(params.get(LAS_STASH_KEY), dict) else ["sheet"]
     if (Path(models_dir) / f"dtm_{SHEET_ANALYSIS_FILTER}.npz").is_file():
         return ["las", "sheet"]
     return ["las"]
@@ -125,8 +151,9 @@ async def merge_surface_confirmation_params(
     connection: aiomysql.Connection,
     project_id: str,
     selection: dict[str, Any],
+    extra: dict[str, Any] | None = None,
 ) -> dict[str, Any]:
-    """기존 stage 1 params에 확정 선택값 전체를 병합 저장한다."""
+    """기존 stage 1 params에 확정 선택값 전체를 병합 저장한다. `extra` 는 함께 적을 칸."""
     normalized = {
         "source_filter": str(selection["source_filter"]),
         "method": str(selection["method"]),
@@ -148,6 +175,7 @@ async def merge_surface_confirmation_params(
             raise LookupError("WF1 단계 상태를 찾을 수 없습니다.")
         params = _decode_params(row.get("params") if row else None)
         params.update(normalized)
+        params.update(extra or {})
         await cursor.execute(
             """
             UPDATE project_workflow_stages

resources/tester/test_133_1_analysis_terrain.py

@@ -99,9 +99,9 @@ def test_sheet_고르면_dtm_sheet_로_표고(tmp_path, monkeypatch):
     _patch_db(monkeypatch, root, LAS_PARAMS)
     assert _elevations("las")["z"] == [LAS_Z]
     assert _elevations("sheet")["z"] == [SHEET_Z]
-    # 기본값 = 지금 동작(확정 지표면 · 라이다)
+    # 기본값 = 지금 설계 지형(확정 지표면 · 133-2)
     request = terrain_router.ElevationRequest(points=[(5.0, 5.0)])
-    assert request.terrain == "las"
+    assert request.terrain is None
 
 
 def test_확정_params_는_안_바뀜():
@@ -163,8 +163,8 @@ def test_화면_고르기_배선():
     assert '<label class="b05-routeedit__terrain" hidden' in chrome
     assert '<option value="las">라이다</option>' in chrome
     assert '<option value="sheet">수치지형도</option>' in chrome
-    assert 'sources.includes("las") && sources.includes("sheet")' in main
-    assert '=== "sheet" ? "sheet" : "las"' in api
+    assert 'sources?.sources.includes("las") && !!sources?.sources.includes("sheet")' in main
+    assert 'saved === "sheet" || saved === "las" ? saved : fallback' in api
     # 모달 분석 넷이 고른 지형을 실어 보냄 · [확인] 재계산은 안 실음
     assert api.count("terrain: analysisTerrain(projectId)") == 3
     assert "terrain: analysisTerrain(projectId)" in initial_api

resources/tester/test_133_2_design_terrain.py

@@ -0,0 +1,297 @@
+"""PLAN 133-2 — 노선 편집 [확인] 때 고른 분석 지형 = 설계 지형(뒤 단계 전부).
+
+수치지형도 고르고 [확인] → 설계 지형 sheet(확정 모델 · stage 1 params) · 종 · 횡단 · 배수 · 유토가
+stage 1 params 로 sheet sampler · 라이다로 되돌려 [확인] → 라이다 · [확인] 안 하면 그대로 ·
+라이다 없는 프로젝트 그대로 · 체인 실패면 설계 지형도 되돌림.
+"""
+
+from __future__ import annotations
+
+import asyncio
+import json
+from contextlib import asynccontextmanager
+from pathlib import Path
+from uuid import UUID
+
+import numpy as np
+
+import B03_FileInput.B03_FileInput_Service_Chain as chain
+import B04_PreProcess.B04_PreProcess_Service as service
+import B05_Profile.B05_Profile_Repository as route_repo
+import common_util.common_util_drainage_pipes as drainage_pipes
+import common_util.common_util_workflow_state as workflow_state
+from B05_Profile import B05_Profile_Router_Replan as replan
+from B06_Section import B06_Section_Router_Stations as stations
+from common_util.common_util_surface_confirmation import (
+    LAS_STASH_KEY,
+    SHEET_ANALYSIS_FILTER,
+    analysis_terrain_sources,
+    design_terrain,
+    get_surface_confirmation_params,
+)
+from common_util.common_util_surface_sampler import build_surface_sampler
+
+ROOT = Path(__file__).resolve().parents[2]
+PID = UUID("1bb0f876-983f-4ca5-9cc7-f59ee1f77e0a")
+LAS_ID, SHEET_ID = 582, 584
+LAS_Z, SHEET_Z = 100.0, 250.0
+LAS_STAGE1 = {"source_filter": "classification", "method": "dtm", "smooth": True}
+SHEET_STAGE1 = {"source_filter": SHEET_ANALYSIS_FILTER, "method": "dtm", "smooth": True}
+
+
+class _Cursor:
+    def __init__(self, db: _Db):
+        self.db = db
+        self.row = None
+        self.rowcount = 1
+
+    async def __aenter__(self):
+        return self
+
+    async def __aexit__(self, *_):
+        return False
+
+    async def execute(self, sql: str, args=()):
+        if "SELECT params" in sql:
+            self.row = {"params": json.dumps(self.db.stage1)}
+        elif "UPDATE project_workflow_stages" in sql:
+            self.db.stage1 = json.loads(args[0])
+            self.db.writes += 1
+        elif "route_start_m" in sql:
+            self.row = (None, None)
+
+    async def fetchone(self):
+        return self.row
+
+
+class _Db:
+    """stage 1 params 한 행 · 지표면 모델 표 — 실제 DB 대신."""
+
+    def __init__(self, stage1: dict, models: list[dict]):
+        self.stage1 = dict(stage1)
+        self.models = models
+        self.writes = 0
+
+    def cursor(self, *_):
+        return _Cursor(self)
+
+    async def begin(self):
+        return None
+
+    async def commit(self):
+        return None
+
+    async def rollback(self):
+        return None
+
+    def confirmed(self) -> int | None:
+        return next((m["id"] for m in self.models if m["status"] == "CONFIRMED"), None)
+
+
+def _models(las: bool) -> list[dict]:
+    rows = [
+        {
+            "id": SHEET_ID,
+            "model_type": "dtm",
+            "status": "COMPLETE" if las else "CONFIRMED",
+            "generation_params": {"source_filter": SHEET_ANALYSIS_FILTER},
+        }
+    ]
+    if las:
+        rows.insert(
+            0,
+            {
+                "id": LAS_ID,
+                "model_type": "dtm",
+                "status": "CONFIRMED",
+                "generation_params": {"source_filter": "classification"},
+            },
+        )
+    return rows
+
+
+def _patch_models(monkeypatch, db: _Db) -> None:
+    async def listed(_connection, _project_id):
+        return db.models
+
+    async def confirm(_connection, _project_id, model_id):
+        for model in db.models:
+            if model["status"] == "CONFIRMED":
+                model["status"] = "COMPLETE"
+        next(m for m in db.models if m["id"] == model_id)["status"] = "CONFIRMED"
+
+    monkeypatch.setattr(service, "list_surface_models", listed)
+    monkeypatch.setattr(service, "confirm_surface_model", confirm)
+
+
+def _switch(db: _Db, terrain):
+    return asyncio.run(service.switch_design_terrain(db, PID, terrain))
+
+
+def _params(db: _Db) -> dict:
+    return asyncio.run(get_surface_confirmation_params(db, str(PID)))
+
+
+def _models_dir(tmp_path: Path) -> Path:
+    models = tmp_path / "B04_PreProcess" / "models"
+    models.mkdir(parents=True, exist_ok=True)
+    for name, z in (("classification", LAS_Z), (SHEET_ANALYSIS_FILTER, SHEET_Z)):
+        for suffix in ("", "_smooth"):
+            x = y = np.arange(0.0, 11.0)
+            np.savez(
+                models / f"dtm_{name}{suffix}.npz",
+                x=x,
+                y=y,
+                z=np.full((11, 11), z),
+                valid_mask=np.ones((11, 11), dtype=bool),
+            )
+    return models
+
+
+def _z(models: Path, params: dict) -> float:
+    sampler = build_surface_sampler(
+        models, params["source_filter"], params["method"], params["smooth"]
+    )
+    z, _ = sampler.sample_xy(np.array([[5.0, 5.0]]))
+    return float(z[0])
+
+
+def test_수치지형도_확인_뒤_설계_지형_sheet_라이다로_되돌림(tmp_path, monkeypatch):
+    db = _Db(LAS_STAGE1, _models(las=True))
+    _patch_models(monkeypatch, db)
+    models = _models_dir(tmp_path)
+    assert _z(models, _params(db)) == LAS_Z
+
+    assert _switch(db, "sheet") == (SHEET_ID, "las")
+    params = _params(db)
+    assert design_terrain(params) == "sheet" and db.confirmed() == SHEET_ID
+    assert params[LAS_STASH_KEY] == LAS_STAGE1  # 되돌릴 자리
+    # 뒤 단계 sampler 는 모두 stage 1 params 를 읽음 → 수치지형도 표고
+    assert _z(models, params) == SHEET_Z
+    # 설계 지형이 수치지형도여도 고르기는 둘 다 남음
+    assert analysis_terrain_sources(params, models) == ["las", "sheet"]
+
+    assert _switch(db, "las") == (LAS_ID, "sheet")
+    params = _params(db)
+    assert design_terrain(params) == "las" and db.confirmed() == LAS_ID
+    assert {k: params[k] for k in LAS_STAGE1} == LAS_STAGE1
+    assert _z(models, params) == LAS_Z
+
+
+def test_확인_안_하면_그대로(monkeypatch):
+    db = _Db(LAS_STAGE1, _models(las=True))
+    _patch_models(monkeypatch, db)
+    assert _switch(db, None) is None  # 고르기 안 보임 · 고른 값 안 실음
+    assert _switch(db, "las") is None  # 지금 설계 지형과 같음
+    assert db.writes == 0 and db.confirmed() == LAS_ID
+
+
+def test_라이다_없는_프로젝트_그대로(monkeypatch, tmp_path):
+    db = _Db(SHEET_STAGE1, _models(las=False))
+    _patch_models(monkeypatch, db)
+    assert _switch(db, "las") is None and _switch(db, "sheet") is None
+    assert db.writes == 0 and db.confirmed() == SHEET_ID
+    assert analysis_terrain_sources(_params(db), _models_dir(tmp_path)) == ["sheet"]
+
+
+def _patch_recompute(monkeypatch, tmp_path: Path, db: _Db, failure: str | None):
+    _patch_models(monkeypatch, db)
+    seen: dict = {}
+
+    class _Pool:
+        @asynccontextmanager
+        async def acquire(self):
+            yield db
+
+    async def latest(_connection, _project_id):
+        return {"id": 237, "surface_model_id": LAS_ID}
+
+    async def state(_cursor, _project_id):
+        return {"stages": []}
+
+    async def start(*_args):
+        return None
+
+    async def run_chain(_project_id, model_id, selection, _target=None):
+        seen["model_id"], seen["selection"] = model_id, dict(selection)
+        return failure
+
+    points = [{"x": 0.0, "y": 0.0}, {"x": 5.0, "y": 5.0}]
+    monkeypatch.setattr(replan, "get_db_pool", lambda: _Pool())
+    monkeypatch.setattr(drainage_pipes, "clear_pipe_points", lambda _s: None)
+    monkeypatch.setattr(route_repo, "get_latest_route", latest)
+    monkeypatch.setattr(chain, "_planned_route_points_in_project_crs", lambda *_a: points)
+    monkeypatch.setattr(chain, "run_redesign_chain", run_chain)
+    monkeypatch.setattr(workflow_state, "get_workflow_state", state)
+    monkeypatch.setattr(workflow_state, "start_stage", start)
+    return seen
+
+
+def test_확인_체인이_수치지형도_모델과_params_로_돎(tmp_path, monkeypatch):
+    db = _Db(LAS_STAGE1, _models(las=True))
+    seen = _patch_recompute(monkeypatch, tmp_path, db, None)
+    result = asyncio.run(replan._recompute(PID, tmp_path, "x", None, "sheet"))
+    assert "error" not in result
+    # 옛 노선의 모델 id(라이다)가 아니라 새 설계 지형 모델 · 종 · 횡단 · 배수 · 유토 체인 params
+    assert seen["model_id"] == SHEET_ID
+    assert seen["selection"]["source_filter"] == SHEET_ANALYSIS_FILTER
+    assert design_terrain(_params(db)) == "sheet"
+
+
+def test_체인_실패면_설계_지형도_되돌림(tmp_path, monkeypatch):
+    db = _Db(LAS_STAGE1, _models(las=True))
+    _patch_recompute(monkeypatch, tmp_path, db, "노선 재계산 실패")
+    result = asyncio.run(replan._recompute(PID, tmp_path, "x", None, "sheet"))
+    assert result == {"error": "노선 재계산 실패"}
+    assert design_terrain(_params(db)) == "las" and db.confirmed() == LAS_ID
+
+
+def test_확인_요청_기본은_설계_지형_그대로():
+    request = replan.RouteReplanRequest(vertices=[])
+    assert request.terrain is None
+
+
+def test_B06_구조물_측점은_옛_지형_조건을_버림(tmp_path, monkeypatch):
+    snapshot = tmp_path / "B06_Section" / "sampling.json"
+    snapshot.parent.mkdir(parents=True)
+    snapshot.write_text(
+        json.dumps({"filter_key": "classification", "method": "dtm", "smooth": True}),
+        encoding="utf-8",
+    )
+
+    def stage1(params):
+        async def run(_fn, _project_id):
+            return {**params, "contour_interval_m": 1.0}
+
+        return run
+
+    monkeypatch.setattr(stations, "run_with_connection", stage1(LAS_STAGE1))
+    same = asyncio.run(stations._sampling_conditions(PID, tmp_path))
+    assert same["filter_key"] == "classification" and "source" not in same
+    monkeypatch.setattr(stations, "run_with_connection", stage1(SHEET_STAGE1))
+    moved = asyncio.run(stations._sampling_conditions(PID, tmp_path))
+    assert moved["filter_key"] == SHEET_ANALYSIS_FILTER and moved["source"] == "stage1"
+
+
+def test_뒤_단계는_stage1_params_를_읽음():
+    """종 · 횡단 · 배수 · 3D 가 설계 지형 한 곳(stage 1 params)을 보는지 — 배선 고정."""
+    readers = [
+        "common_util/common_util_drainage_context.py",
+        "B06_Section/B06_Section_Router.py",
+        "B06_Section/B06_Section_Server_Calc_Prebuild.py",
+        "B06_Section/B06_Section_Router_Stations.py",
+        "B05_Profile/B05_Profile_Router.py",
+        "B03_FileInput/B03_FileInput_Service_Chain.py",
+        "B04_PreProcess/B04_PreProcess_Router.py",
+    ]
+    for path in readers:
+        assert "get_surface_confirmation_params" in (ROOT / path).read_text(encoding="utf-8"), path
+
+
+def test_화면_확인이_고른_지형을_실음():
+    page = ROOT / "B05_Profile"
+    apply = (page / "B05_Profile_UI_RouteEdit_Apply.ts").read_text(encoding="utf-8")
+    main = (page / "B05_Profile_UI_RouteEdit.ts").read_text(encoding="utf-8")
+    assert "terrain: params.terrain()," in apply
+    assert "pickedTerrain() !== designTerrain" in main  # 지형만 바꿔도 [확인]이 다시 셈
+    assert "analysisTerrain(projectId, designTerrain)" in main  # 고른 값 없으면 설계 지형