feat(530-1 · 530-2): D 예정노선 지형 받기 · 2단계 분석 서버 — B05 엔진 인자 셋(models_dir · cache_dir · 복도 반폭) · D03 지형 서비스 · 분석 · API

- B05 지형 로더 models_dir · cache_dir=None · 비용면 cache_dir=None · 탐색 복도 corridor_m(기본 100 m · 기본이면 묶음 열쇠 그대로)
- D03_PlanAnalysis_Terrain — 지점 + 1 km 표본 3×3 도엽 · prune 없음 · 병합 옛 ∪ 새 · 위성 합집합 · 연도별 models_<연도> DEM · plan.prj
- D03_PlanAnalysis_Analyze — 연도 구간마다 갈래 넷 · 법정 기준 · 누가 이어 셈 · 파생 저장본(입력 지문)
- D03_PlanAnalysis_Router — POST plan/terrain · plan/analyze(뒤에서 · 진행률) · GET plan/progress · plan/analysis · PUT [저장]
- 지킴 시험 test_530_1_plan_terrain · test_530_2_plan_analyze · test_530_2_plan_router

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

B05_Profile/B05_Profile_Engine_RouteInitial.py

@@ -132,10 +132,16 @@ _TERRAINS: dict[str, tuple[Terrain, list[list[tuple[float, float]]]]] = {}
 
 
 def load_terrain_and_streams(
-    project_root: Path, selection: dict[str, Any]
+    project_root: Path,
+    selection: dict[str, Any],
+    models_dir: Path | None = None,
+    cache_dir: Path | None = None,
 ) -> tuple[Terrain, list[list[tuple[float, float]]]]:
     """지반 격자(탐색 비용면 캐시 그대로)와 이격을 따질 계류(사업지 좌표계)를 읽는다.
 
+    `models_dir` · `cache_dir` 없으면 B 자리(`B04_PreProcess/models` · `B05_Profile/route`) —
+    D 예정노선은 연도 구간마다 따로 줌(530-2).
+
     같은 입력(비용면 서명 · 계류 파일 지문 · 좌표계)이면 메모리 기억을 그대로 돌려준다 —
     점 놓을 때마다 npz · geojson 을 다시 읽지 않는다. 돌려준 배열은 고치지 않는다.
     """
@@ -156,7 +162,7 @@ def load_terrain_and_streams(
     filter_key = str(selection.get("source_filter"))
     method = str(selection.get("method") or "dtm")
     smooth = bool(selection.get("smooth"))
-    models_dir = project_root / _MODELS_SUBDIR
+    models_dir = project_root / _MODELS_SUBDIR if models_dir is None else Path(models_dir)
     stream_path = project_root / "B04_PreProcess" / "processed" / STREAM_FILE
     crs = resolve_project_crs(project_root)
     try:
@@ -177,7 +183,7 @@ def load_terrain_and_streams(
         return _TERRAINS[memo_key]
 
     x, y, z, valid, *_ = _load_or_build_cost_surface(
-        project_root, models_dir, filter_key, method, smooth
+        project_root, models_dir, filter_key, method, smooth, cache_dir
     )
     to_metric = Transformer.from_crs("EPSG:4326", crs, always_xy=True)
     features = _load_features(stream_path.parent, STREAM_FILE)
@@ -302,6 +308,7 @@ def generate_initial_route(
     pipes: list[tuple[float, float, float]] | None = None,
     with_profile: bool = True,
     standard: dict[str, Any] | None = None,
+    corridor_m: float = CORRIDOR_M,
 ) -> dict[str, Any]:
     """갈래 하나로 초기 계획노선을 만들고 비교값을 붙여 돌려준다. `max_grade` 는 비율(0.14).
 
@@ -313,6 +320,7 @@ def generate_initial_route(
     몇 개에만 균형 종단(`..._Profile` · PLAN 63-4)을 붙여 고른다(419) — 편집 창 추천 점수 · 미리
     보기에만 쓰고 확정 종단과 무관. `pipes`(x, y, 덮개 여유)는 덮개 부족 수만 센다.
     `with_profile` 끄면(종단 반영 끔) 종단 없이 계획고 지반고 횡단 토공으로 고른다.
+    `corridor_m` = 탐색 복도 반폭(기본 B 값 · D 예정노선만 넓힘 · 530-2).
     """
     if mode not in MODES:
         raise ValueError(f"모르는 갈래: {mode}")
@@ -431,7 +439,15 @@ def generate_initial_route(
     pool: Pool | None = None
     target = _weighted_end(used) if mode == "weighted" else mode
     key = _pool_key(
-        expected, terrain, streams, max_grade, min_radius_m, stream_offset, grade_smooth, goal_grade
+        expected,
+        terrain,
+        streams,
+        max_grade,
+        min_radius_m,
+        stream_offset,
+        grade_smooth,
+        goal_grade,
+        *([] if corridor_m == CORRIDOR_M else [corridor_m]),  # 기본 복도 = B 열쇠 그대로
     )
     if target == "follow":  # 비중 예상 100 = 추적 그 선(PLAN 23-3e) · 묶음은 짜지 않음(빠른 길)
         outline, metrics = follow_line()
@@ -445,6 +461,7 @@ def generate_initial_route(
                 stream_offset_m=stream_offset,
                 grade_smooth_m=grade_smooth,
                 target_grade=goal_grade,
+                corridor_m=corridor_m,
             )
             graph = RouteGraph(grid, expected, limit=max_grade, min_radius_m=min_radius_m)
             while len(_POOLS) >= POOL_CACHE_SIZE:
@@ -513,7 +530,7 @@ def generate_initial_route(
             "min_radius_m": round(min_radius_m, 2),
             "stream_offset_m": round(stream_offset, 2),
             "road_width_m": ROAD_WIDTH_M,
-            "corridor_m": CORRIDOR_M,
+            "corridor_m": corridor_m,
             "grade_smooth_m": round(grade_smooth, 2),
             "target_grade_pct": round(goal_grade * 100.0, 2),
             # 기준값 확정(PLAN 20장, 2026-09-29) — 더는 후보가 아니다. 항상 빈 목록.

B05_Profile/B05_Profile_Engine_RouteInitial_Search.py

@@ -111,8 +111,10 @@ class SearchGrid:
         stream_offset_m: float,
         grade_smooth_m: float = GRADE_SMOOTH_M,
         target_grade: float | None = None,
+        corridor_m: float = CORRIDOR_M,
     ) -> None:
         step = max(1, int(round(SEARCH_CELL_M / _cell_size(terrain.x))))
+        self.corridor_m = corridor_m  # 복도 반폭 — 기본 B 값 · D 예정노선만 넓힘(530-2)
         self.target_grade = target_grade  # 영선 노선 목표 기울기(비율) · 없으면 링크 상한
         self.xs, self.ys = terrain.x[::step], terrain.y[::step]
         # 기울기 · 옆경사는 비교값과 같은 고른 지반에서 잰다(`smoothed_ground`)
@@ -122,7 +124,7 @@ class SearchGrid:
         self.d_expected = _distance_grid([expected], self.xs, self.ys, self.cell)
         self.d_stream = _distance_grid(streams, self.xs, self.ys, self.cell)
         valid = np.asarray(terrain.valid[::step, ::step], dtype=bool)
-        self.corridor = (self.d_expected <= CORRIDOR_M) & valid
+        self.corridor = (self.d_expected <= corridor_m) & valid
         self.grad_y, self.grad_x = np.gradient(self.z, self.cell)
         finite = np.where(np.isfinite(self.d_stream), self.d_stream, 0.0)
         self.away_y, self.away_x = np.gradient(finite, self.cell)  # 계류에서 멀어지는 쪽
@@ -214,7 +216,7 @@ class SearchGrid:
         along = (sample(self.d_stream, mid) < self.stream_offset_m) & ~across & (size > 1e-6)
         terms = np.stack(
             [
-                np.minimum(sample(self.d_expected, mid), CORRIDOR_M),
+                np.minimum(sample(self.d_expected, mid), self.corridor_m),
                 earth,
                 (grade / limit) ** 2,
                 over / limit,

B05_Profile/B05_Profile_Engine_Solver.py

@@ -212,11 +212,18 @@ def _build_cost_surface(models_dir: Path, filter_key: str, method: str, smooth:
 
 
 def _load_or_build_cost_surface(
-    project_root: Path, models_dir: Path, filter_key: str, method: str, smooth: bool
+    project_root: Path,
+    models_dir: Path,
+    filter_key: str,
+    method: str,
+    smooth: bool,
+    cache_dir: Path | None = None,
 ):
-    """비용면을 반환한다(서명 일치 시 캐시 재사용, 아니면 재빌드·재캐시)."""
+    """비용면을 반환한다(서명 일치 시 캐시 재사용, 아니면 재빌드·재캐시).
+
+    `cache_dir` 없으면 `B05_Profile/route` — D 예정노선은 연도 구간마다 따로 줌(530-2)."""
     check_surface_keys(filter_key, method)
-    cache_dir = project_root / _ROUTE_CACHE_SUBDIR
+    cache_dir = project_root / _ROUTE_CACHE_SUBDIR if cache_dir is None else Path(cache_dir)
     suffix = "_smooth" if smooth else ""
     cache_path = cache_dir / f"cost_surface_{filter_key}_{method}{suffix}.npz"
     signature = _cost_surface_signature(models_dir, filter_key, method, smooth)

D03_PlanAnalysis/D03_PlanAnalysis_Analyze.py

@@ -0,0 +1,208 @@
+"""D 예정노선 2단계 분석 — 연도 구간마다 B05 노선 추천 엔진으로 지점을 이음 (요구 345 · PLAN 530-2).
+
+- 구간 = [P(i-1), P(i)] 한 번씩 `generate_initial_route` — 시종점을 옮기지 않아 연도 경계가 맞물림
+- 갈래 넷(`PLAN_MODES` · follow 뺌 — 두 점이면 직선) · 연도마다 고름
+- 추천 기준 = 프로젝트 임도 종류 · 설계속도 → B 법정 함수 · 지형 구분은 B 폴백(`TERRAIN_TYPE`)
+- 계류 이격 = 도엽 하천선(B 로더 그대로) · 복도 반폭 = `PLAN_CORRIDOR_M`(D 만)
+- 연도 경계는 꺾인 채(경계점 = 측점) · 누가거리 = 등록 시작 누가부터 고른 갈래 길이로 이어 셈
+- 파생 저장본 = `D03_PlanAnalysis/_derived/route_<연도>.json`(입력 지문 같으면 다시 안 셈)
+
+계산 자리 = ③ 서버 단독(B 추천과 같음).
+"""
+
+from __future__ import annotations
+
+import hashlib
+import json
+import logging
+from pathlib import Path
+from typing import Any, Callable
+
+from B05_Profile.B05_Profile_Engine_Grade import (
+    legal_grade_limit_pct,
+    legal_plan_radius_min_m,
+    resolve_design_speed,
+)
+from B05_Profile.B05_Profile_Engine_RouteInitial import (
+    generate_initial_route,
+    load_terrain_and_streams,
+)
+from B05_Profile.B05_Profile_Engine_RouteInitial_Store import request_fingerprint
+from common_util.common_util_json import atomic_write_json
+from common_util.common_util_surface_confirmation import SHEET_ANALYSIS_FILTER
+from D03_PlanAnalysis.D03_PlanAnalysis_Terrain import PLAN_CORRIDOR_M, models_dir_of
+
+logger = logging.getLogger(__name__)
+
+PLAN_MODES = ("gradeline", "earthwork", "gentle", "weighted")
+# 도엽 DEM(스무딩) — B 「수치지형도」 분석 지형과 같은 고르기
+SELECTION = {"source_filter": SHEET_ANALYSIS_FILTER, "method": "dtm", "smooth": True}
+# D 등록에 지형 구분 칸 없음 — B `_road_settings` 못 읽을 때 폴백 그대로(가장 완화)
+TERRAIN_TYPE = "special"
+ALGO = 1
+
+Progress = Callable[[int, str], None]
+Loader = Callable[..., tuple[Any, list]]
+
+
+def derived_dir(project_root: Path) -> Path:
+    return project_root / "D03_PlanAnalysis" / "_derived"
+
+
+def plan_criteria(road_type: str | None, design_speed_kph: int | None) -> dict[str, Any]:
+    """추천 기준 — 종단 상한(비율) · 최소 R · 설계속도(B 법정 함수)."""
+    grade_class = str(road_type or "work")
+    speed = resolve_design_speed(grade_class, design_speed_kph)
+    return {
+        "max_grade": legal_grade_limit_pct(grade_class, TERRAIN_TYPE, False, speed) / 100.0,
+        "min_radius_m": legal_plan_radius_min_m(speed, TERRAIN_TYPE),
+        "design_speed": speed,
+    }
+
+
+def _same_point(a: Any, b: Any) -> bool:
+    return abs(float(a[0]) - float(b[0])) < 1e-6 and abs(float(a[1]) - float(b[1])) < 1e-6
+
+
+def read_segment(project_root: Path, seg: dict[str, Any]) -> dict[str, Any] | None:
+    """저장본 — 지점이 지금과 같을 때만(입력 지문은 분석 때 다시 봄)."""
+    try:
+        stored = json.loads(
+            (derived_dir(project_root) / f"route_{int(seg['year'])}.json").read_text("utf-8")
+        )
+    except (OSError, ValueError):
+        return None
+    if _same_point(stored["start"], seg["start"]) and _same_point(stored["end"], seg["end"]):
+        return stored
+    return None
+
+
+def _route_view(result: dict[str, Any]) -> dict[str, Any]:
+    keep = ("planned", "nodes", "curves", "metrics", "relaxed", "terrain_bound_pct")
+    return {key: result.get(key) for key in keep}
+
+
+def analyze_segment(
+    project_root: Path,
+    seg: dict[str, Any],
+    terrain: Any,
+    streams: list,
+    criteria: dict[str, Any],
+    corridor_m: float = PLAN_CORRIDOR_M,
+) -> dict[str, Any]:
+    """연도 구간 하나 · 갈래 넷 — 입력 지문이 저장본과 같으면 그대로(`stored` True)."""
+    expected = [tuple(map(float, seg["start"])), tuple(map(float, seg["end"]))]
+    inputs = {
+        "weights": None,
+        "max_grade": criteria["max_grade"],
+        "min_radius_m": criteria["min_radius_m"],
+        "stream_offset_m": None,
+        "grade_smooth_m": None,
+        "target_grade": None,
+        "with_profile": False,
+        "design_speed": criteria["design_speed"],
+    }
+    keys = [
+        request_fingerprint(expected, terrain, streams, mode=mode, **inputs)[0]
+        for mode in PLAN_MODES
+    ]
+    fingerprint = hashlib.blake2b(
+        repr((keys, float(corridor_m), ALGO)).encode(), digest_size=16
+    ).hexdigest()
+    stored = read_segment(project_root, seg)
+    if stored is not None and stored.get("fingerprint") == fingerprint:
+        return {**stored, "stored": True}
+    routes: dict[str, Any] = {}
+    for mode in PLAN_MODES:
+        try:
+            routes[mode] = _route_view(
+                generate_initial_route(
+                    expected, terrain, streams, mode=mode, corridor_m=corridor_m, **inputs
+                )
+            )
+        except ValueError as error:  # NoRoute 포함 — 그 갈래만 빔
+            routes[mode] = {"error": str(error)}
+    result = {
+        "year": int(seg["year"]),
+        "start": list(expected[0]),
+        "end": list(expected[1]),
+        "fingerprint": fingerprint,
+        "corridor_m": float(corridor_m),
+        "criteria": {
+            "max_grade_pct": round(criteria["max_grade"] * 100.0, 2),
+            "min_radius_m": criteria["min_radius_m"],
+            "design_speed_kph": criteria["design_speed"],
+            "terrain_type": TERRAIN_TYPE,
+        },
+        "routes": routes,
+    }
+    atomic_write_json(derived_dir(project_root) / f"route_{int(seg['year'])}.json", result)
+    return {**result, "stored": False}
+
+
+def analyze_plan(
+    project_root: Path,
+    segments: list[dict[str, Any]],
+    criteria: dict[str, Any],
+    *,
+    loader: Loader = load_terrain_and_streams,
+    corridor_m: float = PLAN_CORRIDOR_M,
+    on_progress: Progress | None = None,
+) -> list[dict[str, Any]]:
+    """연도 구간마다 지형(연도 DEM · 연도 비용면 캐시)을 읽어 `analyze_segment`."""
+    results = []
+    for i, seg in enumerate(segments):
+        year = int(seg["year"])
+        if on_progress is not None:
+            on_progress(100 * i // len(segments), f"{year}년 구간 노선 추천 중")
+        terrain, streams = loader(
+            project_root,
+            SELECTION,
+            models_dir_of(project_root, year),
+            derived_dir(project_root) / f"cost_{year}",
+        )
+        results.append(analyze_segment(project_root, seg, terrain, streams, criteria, corridor_m))
+    if on_progress is not None:
+        on_progress(100, "분석 끝")
+    return results
+
+
+def pick_mode(result: dict[str, Any], wanted: str | None) -> str | None:
+    """고른 갈래 — 없거나 빈 갈래면 차례대로 첫 성공 갈래."""
+    routes = result.get("routes") or {}
+    if wanted in routes and "error" not in routes[wanted]:
+        return wanted
+    return next((m for m in PLAN_MODES if m in routes and "error" not in routes[m]), None)
+
+
+def chain(
+    results: list[dict[str, Any]], choices: dict[int, str | None], start_chainage_m: float
+) -> list[dict[str, Any]]:
+    """연도별 고른 갈래 · 시작 · 끝 누가(이어 셈) — 앞 연도 끝 누가 = 다음 연도 시작 누가."""
+    at = float(start_chainage_m or 0.0)
+    rows = []
+    for result in results:
+        mode = pick_mode(result, choices.get(int(result["year"])))
+        length = float(result["routes"][mode]["metrics"]["length_m"]) if mode else 0.0
+        rows.append(
+            {
+                "year": int(result["year"]),
+                "mode": mode,
+                "length_m": round(length, 3),
+                "start_chainage_m": round(at, 3),
+                "end_chainage_m": round(at + length, 3),
+            }
+        )
+        at += length
+    return rows
+
+
+def joined_line(results: list[dict[str, Any]], rows: list[dict[str, Any]]) -> list[list[float]]:
+    """고른 갈래를 이은 중심선 — 경계점 한 번만(꺾인 채)."""
+    line: list[list[float]] = []
+    for result, row in zip(results, rows):
+        if row["mode"] is None:
+            continue
+        planned = result["routes"][row["mode"]]["planned"]
+        line.extend(planned[1:] if line and _same_point(line[-1], planned[0]) else planned)
+    return line

D03_PlanAnalysis/D03_PlanAnalysis_Router.py

@@ -0,0 +1,250 @@
+"""D 예정노선 2단계 분석 API (요구 345 · PLAN 530-1 · 530-2).
+
+- `POST /api/projects/{id}/plan/terrain` — 지형 받기(뒤에서) · 끝나면 연도별 `surface_models` 행
+- `POST …/plan/analyze` — 연도 구간마다 노선 추천 갈래 넷(뒤에서 · 파생 저장본)
+- `GET …/plan/progress` — 두 일의 진행률
+- `GET …/plan/analysis` — 저장본 + 고른 갈래 · 이은 누가 · 이은 중심선
+- `PUT …/plan/analysis` — [저장] 연도별 고른 갈래 → `plan_route_years.route_json` · `fingerprint`
+
+지점 정본은 `plan_route_years` 시점 · 종점 칸(529-3 몫)을 읽기만 함. 계산 자리 = ③ 서버 단독 —
+[저장]은 브라우저 값을 받지 않고 갈래 이름만 받아 서버 저장본으로 다시 셈.
+"""
+
+from __future__ import annotations
+
+import asyncio
+import json
+import logging
+from pathlib import Path
+from typing import Any
+from uuid import UUID
+
+import aiomysql
+from fastapi import APIRouter, HTTPException
+from pydantic import BaseModel
+
+from B03_FileInput.B03_FileInput_Repository import get_project_storage_relative_path
+from common_util.common_util_storage import resolve_stored_project_path
+from config.config_db import get_db_pool
+from D03_PlanAnalysis import D03_PlanAnalysis_Analyze as analyze
+from D03_PlanAnalysis import D03_PlanAnalysis_Terrain as terrain
+
+logger = logging.getLogger(__name__)
+router = APIRouter(prefix="/api/projects", tags=["D03 Plan Analysis"])
+
+# 진행 — (프로젝트, 일) → {running, percent, message, error} · 프로세스 메모리
+_JOBS: dict[tuple[str, str], dict[str, Any]] = {}
+_TASKS: set[asyncio.Task] = set()
+KINDS = ("terrain", "analyze")
+
+
+class ChoiceRequest(BaseModel):
+    choices: dict[int, str]  # 연도 → 갈래
+
+
+async def _plan_inputs(project_id: UUID) -> tuple[Path, dict[str, Any], list[dict[str, Any]]]:
+    """(프로젝트 실경로, 프로젝트 칸, 연도 구간) — 지점이 빈 연도 · 좌표계 섞임은 거절."""
+    async with get_db_pool().acquire() as connection:
+        stored = await get_project_storage_relative_path(connection, project_id)
+        async with connection.cursor(aiomysql.DictCursor) as cursor:
+            await cursor.execute(
+                "SELECT road_type, design_speed_kph, route_start_m FROM projects WHERE id = %s",
+                (str(project_id),),
+            )
+            project = await cursor.fetchone()
+            await cursor.execute(
+                "SELECT year, start_x, start_y, end_x, end_y, epsg, route_json, fingerprint "
+                "FROM plan_route_years WHERE project_id = %s ORDER BY year",
+                (str(project_id),),
+            )
+            rows = await cursor.fetchall()
+    if not stored or not project:
+        raise HTTPException(status_code=404, detail="프로젝트 저장 경로를 찾을 수 없습니다.")
+    if not rows:
+        raise HTTPException(
+            status_code=400, detail="연도별 지점이 없습니다 — 1단계에서 지정하세요."
+        )
+    for row in rows:
+        if any(row[k] is None for k in ("start_x", "start_y", "end_x", "end_y", "epsg")):
+            raise HTTPException(status_code=400, detail=f"{row['year']}년 지점이 비었습니다.")
+    if len({int(row["epsg"]) for row in rows}) != 1:
+        raise HTTPException(status_code=400, detail="연도마다 좌표계가 다릅니다.")
+    segments = [
+        {
+            "year": int(row["year"]),
+            "start": (float(row["start_x"]), float(row["start_y"])),
+            "end": (float(row["end_x"]), float(row["end_y"])),
+            "epsg": int(row["epsg"]),
+            "route_json": row["route_json"],
+        }
+        for row in rows
+    ]
+    return Path(resolve_stored_project_path(stored)), project, segments
+
+
+def _start_job(project_id: UUID, kind: str, work: Any) -> dict[str, Any]:
+    key = (str(project_id), kind)
+    if (_JOBS.get(key) or {}).get("running"):
+        raise HTTPException(status_code=409, detail="이미 돌고 있습니다.")
+    job = {"running": True, "percent": 0, "message": "시작", "error": None}
+    _JOBS[key] = job
+
+    def report(percent: int, message: str) -> None:
+        job.update(percent=int(percent), message=message)
+
+    async def run() -> None:
+        try:
+            await work(report)
+            job.update(percent=100)
+        except Exception as error:  # noqa: BLE001 — 진행률에 까닭을 그대로 올림
+            logger.exception("D %s 실패: project_id=%s", kind, project_id)
+            job["error"] = str(getattr(error, "detail", None) or error)
+        finally:
+            job["running"] = False
+
+    task = asyncio.create_task(run(), name=f"D {kind} {project_id}")
+    _TASKS.add(task)
+    task.add_done_callback(_TASKS.discard)
+    return {"status": "accepted", **job}
+
+
+async def _save_models(project_id: UUID, years: dict[int, list[dict[str, Any]]], epsg: int) -> None:
+    """연도별 지형면 → `surface_models` 행(같은 파일 · 같은 지문이면 id 살림 · 다른 행 그대로)."""
+    from B04_PreProcess.B04_PreProcess_Repository import _existing_models, _upsert_surface_model
+
+    async with get_db_pool().acquire() as connection:
+        await connection.begin()
+        try:
+            existing = await _existing_models(connection, project_id)
+            for models in years.values():
+                for model in models:
+                    await _upsert_surface_model(
+                        connection,
+                        existing,
+                        project_id=project_id,
+                        model=model,
+                        source_file_id=None,
+                        processed_cloud_id=None,
+                        crs_epsg=epsg,
+                    )
+            await connection.commit()
+        except Exception:
+            await connection.rollback()
+            raise
+
+
+@router.post("/{project_id}/plan/terrain", response_model=None)
+async def plan_terrain(project_id: UUID) -> dict[str, Any]:
+    """지점 기준 수치지형도 · 위성 · 연도별 DEM 받기(뒤에서) — 진행은 `GET …/plan/progress`."""
+    root, _, segments = await _plan_inputs(project_id)
+    epsg = segments[0]["epsg"]
+
+    async def work(report: Any) -> None:
+        got = await asyncio.to_thread(terrain.ensure_plan_terrain, root, segments, epsg, report)
+        await _save_models(project_id, got["years"], epsg)
+        report(100, f"지형 받기 끝 — 새 도엽 {len(got['sheets']['downloaded'])}매")
+
+    return _start_job(project_id, "terrain", work)
+
+
+@router.post("/{project_id}/plan/analyze", response_model=None)
+async def plan_analyze(project_id: UUID) -> dict[str, Any]:
+    """연도 구간마다 노선 추천 갈래 넷(뒤에서 · 같은 입력이면 저장본)."""
+    root, project, segments = await _plan_inputs(project_id)
+    criteria = analyze.plan_criteria(project["road_type"], project["design_speed_kph"])
+    missing = [s["year"] for s in segments if not terrain.models_dir_of(root, s["year"]).is_dir()]
+    if missing:
+        raise HTTPException(status_code=409, detail=f"지형을 먼저 받으세요: {missing}년")
+
+    async def work(report: Any) -> None:
+        await asyncio.to_thread(
+            lambda: analyze.analyze_plan(root, segments, criteria, on_progress=report)
+        )
+
+    return _start_job(project_id, "analyze", work)
+
+
+@router.get("/{project_id}/plan/progress", response_model=None)
+async def plan_progress(project_id: UUID) -> dict[str, Any]:
+    idle = {"running": False, "percent": 0, "message": None, "error": None}
+    return {
+        "status": "success",
+        **{kind: dict(_JOBS.get((str(project_id), kind)) or idle) for kind in KINDS},
+    }
+
+
+def _saved_mode(route_json: Any) -> str | None:
+    try:
+        value = json.loads(route_json) if isinstance(route_json, str) else route_json
+    except ValueError:
+        return None
+    return value.get("mode") if isinstance(value, dict) else None
+
+
+def _assemble(
+    root: Path, project: dict[str, Any], segments: list[dict[str, Any]], choices: dict[int, Any]
+) -> tuple[list[dict[str, Any] | None], list[dict[str, Any]] | None, list[list[float]]]:
+    """(연도 저장본 · 이은 누가 · 이은 중심선) — 저장본이 하나라도 없으면 None · 빈 선."""
+    results = [analyze.read_segment(root, seg) for seg in segments]
+    if any(result is None for result in results):
+        return results, None, []
+    rows = analyze.chain(results, choices, float(project["route_start_m"] or 0.0))
+    return results, rows, analyze.joined_line(results, rows)
+
+
+@router.get("/{project_id}/plan/analysis", response_model=None)
+async def plan_analysis(project_id: UUID) -> dict[str, Any]:
+    """분석 저장본(지점이 지금과 같은 것만) + 고른 갈래(저장값 · 없으면 첫 성공 갈래)."""
+    root, project, segments = await _plan_inputs(project_id)
+    choices = {seg["year"]: _saved_mode(seg["route_json"]) for seg in segments}
+    results, rows, line = await asyncio.to_thread(_assemble, root, project, segments, choices)
+    return {
+        "status": "success",
+        "years": [
+            {"year": seg["year"], "start": seg["start"], "end": seg["end"], "analysis": result}
+            for seg, result in zip(segments, results)
+        ],
+        "chain": rows,
+        "line": line,
+    }
+
+
+@router.put("/{project_id}/plan/analysis", response_model=None)
+async def save_plan_analysis(project_id: UUID, request: ChoiceRequest) -> dict[str, Any]:
+    """[저장] — 연도별 고른 갈래를 서버 저장본으로 다시 이어 `plan_route_years` 에 씀."""
+    root, project, segments = await _plan_inputs(project_id)
+    bad = [m for m in request.choices.values() if m not in analyze.PLAN_MODES]
+    if bad:
+        raise HTTPException(status_code=400, detail=f"모르는 갈래: {bad}")
+    choices = {seg["year"]: _saved_mode(seg["route_json"]) for seg in segments}
+    choices.update(request.choices)
+    results, rows, _ = await asyncio.to_thread(_assemble, root, project, segments, choices)
+    if rows is None:
+        raise HTTPException(
+            status_code=409, detail="분석 저장본이 없습니다 — [분석]을 먼저 하세요."
+        )
+    lost = [r["year"] for r in rows if choices.get(r["year"]) not in (None, r["mode"])]
+    if lost:
+        raise HTTPException(status_code=409, detail=f"고른 갈래에 노선이 없습니다: {lost}년")
+    async with get_db_pool().acquire() as connection:
+        await connection.begin()
+        try:
+            async with connection.cursor() as cursor:
+                for result, row in zip(results, rows):
+                    route = result["routes"][row["mode"]]
+                    body = {**row, **{k: route[k] for k in ("planned", "nodes", "curves")}}
+                    await cursor.execute(
+                        "UPDATE plan_route_years SET route_json = %s, fingerprint = %s "
+                        "WHERE project_id = %s AND year = %s",
+                        (
+                            json.dumps(body, ensure_ascii=False),
+                            result["fingerprint"],
+                            str(project_id),
+                            row["year"],
+                        ),
+                    )
+            await connection.commit()
+        except Exception:
+            await connection.rollback()
+            raise
+    return {"status": "success", "chain": rows}

D03_PlanAnalysis/D03_PlanAnalysis_Terrain.py

@@ -0,0 +1,221 @@
+"""D 예정노선 지형 받기 — 지점 기준 수치지형도 · 위성 · 연도 구간별 DEM (요구 345 · PLAN 530-1).
+
+지형은 D 프로젝트 storage 안 `B04_PreProcess/` 배치 그대로(지형 코드 주인 = B04 · 고칠 곳 0):
+
+- 도엽 = 모든 지점 + 구간 표본(`SAMPLE_STEP_M`) 각 3×3 → `ensure_sheets`(없는 것만 · prune 없음)
+- 병합 = 지난번 병합 도엽 ∪ 이번 도엽(옛 도엽 유지 · 열쇠 같으면 건너뜀)
+- 위성 = 기준 도엽 도곽 ∪ 옛 위성 범위(덮으면 건너뜀)
+- DEM = 연도 구간마다 `models_<연도>/`(긴 노선 메모리 · 비용면 한도) · 범위 = 구간 상자 + 탐색 복도
+- 작업 좌표계 = 지점 epsg · `B03_FileInput/input/prj/plan.prj` 한 장(B04 · B05 좌표계 창구가 읽음)
+
+계산 자리 = ③ 서버 단독. DB 행(`surface_models`)은 라우터가 씀.
+"""
+
+from __future__ import annotations
+
+import json
+import logging
+import math
+from pathlib import Path
+from typing import Any, Callable
+
+import numpy as np
+
+from B04_PreProcess.B04_PreProcess_Engine_MapSheet import (
+    neighbors_for_points,
+    sheet5k_to_bounds,
+    sheets_for_points,
+)
+from B04_PreProcess.B04_PreProcess_Engine_SheetParser import parse_and_merge_sheets
+from B04_PreProcess.B04_PreProcess_Engine_SheetStore import (
+    ensure_sheets,
+    get_project_map_sheets_dir,
+    get_sheet_path,
+)
+from B04_PreProcess.B04_PreProcess_Engine_SheetSurface import build_sheet_surface_model
+from B04_PreProcess.B04_PreProcess_Engine_VWorld import download_vworld_satellite_map
+from common_util.common_util_json import atomic_write_json
+from config.config_system_terrain import SHEET_SURFACE_DEFAULT_METHOD, SHEET_SURFACE_MARGIN_M
+
+logger = logging.getLogger(__name__)
+
+# 구간 표본 간격 — 도엽(≈2.8 × 2.3 km) 3×3 묶음이 끊기지 않게
+SAMPLE_STEP_M = 1000.0
+# 탐색 복도 반폭(D 만 · B 기본 100 m) — 실측(530-2): 노선이 복도 끝에 닿지 않는 폭
+PLAN_CORRIDOR_M = 300.0
+# DEM 이 복도 밖으로 더 덮는 폭(복도 끝 칸 기울기 · 횡단 표본 ±20 m)
+DEM_EXTRA_M = 50.0
+STAGE = Path("B04_PreProcess")
+PRJ_NAME = "plan.prj"
+
+Progress = Callable[[int, str], None]
+
+
+def models_dir_of(project_root: Path, year: int) -> Path:
+    """연도 구간 DEM 자리."""
+    return project_root / STAGE / f"models_{int(year)}"
+
+
+def segment_samples(
+    start: tuple[float, float], end: tuple[float, float]
+) -> list[tuple[float, float]]:
+    """구간 위 표본(끝점 둘 포함 · `SAMPLE_STEP_M` 간격 이하)."""
+    length = math.hypot(end[0] - start[0], end[1] - start[1])
+    count = max(1, math.ceil(length / SAMPLE_STEP_M))
+    return [
+        (start[0] + (end[0] - start[0]) * i / count, start[1] + (end[1] - start[1]) * i / count)
+        for i in range(count + 1)
+    ]
+
+
+def reference_points(segments: list[dict[str, Any]], epsg: int) -> list[tuple[float, float]]:
+    """도엽 기준 좌표(위도, 경도) — 모든 지점 + 구간 표본."""
+    from pyproj import Transformer
+
+    to_wgs = Transformer.from_crs(f"EPSG:{epsg}", "EPSG:4326", always_xy=True)
+    points: list[tuple[float, float]] = []
+    for seg in segments:
+        for x, y in segment_samples(seg["start"], seg["end"]):
+            lon, lat = to_wgs.transform(x, y)
+            points.append((lat, lon))
+    return points
+
+
+def write_plan_prj(project_root: Path, epsg: int) -> Path:
+    """작업 좌표계 .prj — 내용이 같으면 그대로."""
+    from pyproj import CRS
+
+    path = project_root / "B03_FileInput" / "input" / "prj" / PRJ_NAME
+    text = CRS.from_epsg(int(epsg)).to_wkt("WKT1_ESRI")
+    if not path.is_file() or path.read_text(encoding="utf-8", errors="ignore") != text:
+        path.parent.mkdir(parents=True, exist_ok=True)
+        path.write_text(text, encoding="utf-8")
+    return path
+
+
+def dem_outline(start: tuple[float, float], end: tuple[float, float]) -> np.ndarray:
+    """DEM 상자 꼭짓점 둘 — B 여유(`SHEET_SURFACE_MARGIN_M`)를 더하면 복도 + `DEM_EXTRA_M`."""
+    pad = max(0.0, PLAN_CORRIDOR_M + DEM_EXTRA_M - SHEET_SURFACE_MARGIN_M)
+    xs, ys = (start[0], end[0]), (start[1], end[1])
+    return np.array(
+        [[min(xs) - pad, min(ys) - pad], [max(xs) + pad, max(ys) + pad]], dtype=np.float64
+    )
+
+
+def _read_json(path: Path) -> dict[str, Any] | None:
+    try:
+        return json.loads(path.read_text(encoding="utf-8"))
+    except (OSError, ValueError):
+        return None
+
+
+def _merge_sheets(processed: Path, store: Path, available: list[str]) -> list[str]:
+    """병합 = 지난번 병합 도엽(zip 남은 것) ∪ 이번 도엽 · 열쇠 같으면 건너뜀."""
+    key_path = processed / "sheet_merge.key.json"
+    old = (_read_json(key_path) or {}).get("sheets") or []
+    sheets = sorted({*available, *(s for s in old if get_sheet_path(store, s) is not None)})
+    key = {"sheets": sheets}
+    if sheets and (_read_json(key_path) != key or not any(processed.glob("도엽_*.geojson"))):
+        parse_and_merge_sheets(store, sheets, processed)
+        atomic_write_json(key_path, key)
+    return sheets
+
+
+def _satellite(processed: Path, prj: Path, points: list[tuple[float, float]], epsg: int) -> str:
+    """위성 = 기준 도엽 도곽 ∪ 옛 범위 · 이미 덮으면 건너뜀. 돌려줌 = skip · done · failed."""
+    from pyproj import Transformer
+
+    to_metric = Transformer.from_crs("EPSG:4326", f"EPSG:{epsg}", always_xy=True)
+    xs: list[float] = []
+    ys: list[float] = []
+    for sheet in sheets_for_points(points):
+        lon0, lat0, lon1, lat1 = sheet5k_to_bounds(sheet)
+        for lon, lat in ((lon0, lat0), (lon0, lat1), (lon1, lat0), (lon1, lat1)):
+            x, y = to_metric.transform(lon, lat)
+            xs.append(x)
+            ys.append(y)
+    need = {"x": [min(xs), max(xs)], "y": [min(ys), max(ys)]}
+    meta = _read_json(processed / "vworld_satellite_meta.json")
+    try:
+        old = [float(meta[k]) for k in ("x_min", "x_max", "y_min", "y_max")] if meta else None
+    except (KeyError, TypeError, ValueError):
+        old = None
+    if old and old[0] <= need["x"][0] and old[1] >= need["x"][1]:
+        if old[2] <= need["y"][0] and old[3] >= need["y"][1]:
+            return "skip"
+    if old:
+        need = {
+            "x": [min(old[0], need["x"][0]), max(old[1], need["x"][1])],
+            "y": [min(old[2], need["y"][0]), max(old[3], need["y"][1])],
+        }
+    try:
+        done = download_vworld_satellite_map(prj, need, processed, layer_name="Satellite")
+    except Exception as exc:  # noqa: BLE001 — 위성은 배경 · 지형 받기를 막지 않음
+        logger.warning("D 위성 지도 받기 실패: %s", exc)
+        return "failed"
+    return "done" if done else "failed"
+
+
+def ensure_plan_terrain(
+    project_root: Path,
+    segments: list[dict[str, Any]],
+    epsg: int,
+    on_progress: Progress | None = None,
+) -> dict[str, Any]:
+    """지점 기준 지형 받기 — `segments` = [{year, start: (x, y), end: (x, y)}](작업 좌표계 m).
+
+    돌려줌 = 도엽(받은 · 실패 · 병합) · 위성 상태 · 연도별 모델 등록용 dict 목록.
+    """
+
+    def report(percent: int, message: str) -> None:
+        if on_progress is not None:
+            on_progress(percent, message)
+
+    if not segments:
+        raise ValueError("연도 구간이 없습니다.")
+    processed = project_root / STAGE / "processed"
+    processed.mkdir(parents=True, exist_ok=True)
+    prj = write_plan_prj(project_root, epsg)
+    crs = f"EPSG:{int(epsg)}"
+
+    report(5, "수치지형도 도엽 고르는 중")
+    points = reference_points(segments, epsg)
+    wanted = neighbors_for_points(points)
+    store = get_project_map_sheets_dir(project_root)
+    before = {s for s in wanted if get_sheet_path(store, s) is not None}
+    report(10, f"수치지형도 도엽 {len(wanted) - len(before)}매 받는 중")
+    got = ensure_sheets(store, wanted)
+    if not got["available"]:
+        raise ValueError(f"수치지형도 도엽을 하나도 받지 못했습니다: {got['failed']}")
+    report(45, "도엽 레이어 병합 중")
+    merged = _merge_sheets(processed, store, list(got["available"]))
+    report(60, "위성 지도 받는 중")
+    satellite = _satellite(processed, prj, points, epsg)
+
+    years: dict[int, list[dict[str, Any]]] = {}
+    for i, seg in enumerate(segments):
+        year = int(seg["year"])
+        report(65 + 30 * i // len(segments), f"{year}년 구간 지형면 만드는 중")
+        models_dir = models_dir_of(project_root, year)
+        models_dir.mkdir(parents=True, exist_ok=True)
+        years[year] = build_sheet_surface_model(
+            project_root,
+            processed,
+            models_dir,
+            dem_outline(seg["start"], seg["end"]),
+            crs,
+            [SHEET_SURFACE_DEFAULT_METHOD],
+        )
+        if not years[year]:
+            raise ValueError(f"{year}년 구간 지형면을 만들지 못했습니다 — 도엽 등고선 확인")
+    report(100, "지형 받기 끝")
+    return {
+        "sheets": {
+            "wanted": len(wanted),
+            "downloaded": sorted(set(got["available"]) - before),
+            "failed": got["failed"],
+            "merged": merged,
+        },
+        "satellite": satellite,
+        "years": years,
+    }

main.py

@@ -72,6 +72,7 @@ from B08_Quantity.B08_Quantity_Router_Summary import router as b08_summary_route
 from B08_Quantity.B08_Quantity_Router_Material import router as b08_material_router
 from B08_Quantity.B08_Quantity_Router_Milestone import router as b08_milestone_router
 from B09_Estimation.B09_Estimation_Router import router as b09_estimation_router
+from D03_PlanAnalysis.D03_PlanAnalysis_Router import router as d03_plan_analysis_router
 from M01_MasterData.M01_MasterData_Router import router as m01_master_data_router
 from M02_MasterTemplete.M02_Labor_Router import router as m02_labor_router
 from M02_MasterTemplete.M02_MasterTemplete_Router import router as m02_master_template_router
@@ -468,6 +469,7 @@ app.include_router(b04_basins_router, dependencies=protected_with_company)
 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(d03_plan_analysis_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_route_terrain_router, dependencies=protected_with_company)

resources/tester/test_530_1_plan_terrain.py

@@ -0,0 +1,115 @@
+"""D 예정노선 지형 받기(PLAN 530-1) — 가짜 내려받기 · 옛 도엽 유지 · 새 도엽만 · 빠짐 없음.
+
+./venv/Scripts/python.exe -m pytest resources/tester/test_530_1_plan_terrain.py -q
+"""
+
+import json
+import math
+from pathlib import Path
+
+import pytest
+from pyproj import Transformer
+
+from B04_PreProcess import B04_PreProcess_Engine_SheetStore as store_mod
+from B04_PreProcess.B04_PreProcess_Engine_MapSheet import sheets_for_points
+from common_util.common_util_crs import resolve_project_crs
+from config.config_system_terrain import SHEET_SURFACE_MARGIN_M
+from D03_PlanAnalysis import D03_PlanAnalysis_Terrain as terrain
+
+EPSG = 5186
+A, B, C = (230000.0, 450000.0), (231500.0, 451200.0), (238000.0, 455000.0)
+
+
+@pytest.fixture
+def fake(tmp_path, monkeypatch):
+    seen = {"download": [], "merge": [], "satellite": [], "dem": []}
+
+    def download_sheet(store_dir, sheet_no, version=2):
+        seen["download"].append(sheet_no)
+        store_dir = Path(store_dir)
+        store_dir.mkdir(parents=True, exist_ok=True)
+        (store_dir / f"{sheet_no}.zip").write_bytes(b"PK")
+        index = store_mod._load_index(store_dir)
+        index["sheets"][sheet_no] = {"file": f"{sheet_no}.zip"}
+        store_mod._save_index(store_dir, index)
+
+    def merge(store_dir, sheets, output_dir, layers=None):
+        seen["merge"].append(list(sheets))
+        (output_dir / "도엽_등고선.geojson").write_text("{}", encoding="utf-8")
+        return {}
+
+    def satellite(prj, bounds, output_dir, layer_name="Satellite", ext="jpeg"):
+        seen["satellite"].append(bounds)
+        meta = {"x_min": bounds["x"][0], "x_max": bounds["x"][1]}
+        meta.update({"y_min": bounds["y"][0], "y_max": bounds["y"][1]})
+        (output_dir / "vworld_satellite_meta.json").write_text(json.dumps(meta), "utf-8")
+        return output_dir / "vworld_satellite.png"
+
+    def dem(root, processed, models_dir, route_xy, crs, methods=None):
+        seen["dem"].append((models_dir.name, route_xy.tolist(), crs))
+        return [{"model_file_path": f"B04_PreProcess/{models_dir.name}/dtm.npz"}]
+
+    monkeypatch.setattr(store_mod, "download_sheet", download_sheet)
+    monkeypatch.setattr(terrain, "parse_and_merge_sheets", merge)
+    monkeypatch.setattr(terrain, "download_vworld_satellite_map", satellite)
+    monkeypatch.setattr(terrain, "build_sheet_surface_model", dem)
+    return tmp_path, seen
+
+
+def _zips(root):
+    return sorted(p.stem for p in store_mod.get_project_map_sheets_dir(root).glob("*.zip"))
+
+
+def test_more_points_keep_old_sheets_and_fetch_only_new(fake) -> None:
+    root, seen = fake
+    first = terrain.ensure_plan_terrain(root, [{"year": 2026, "start": A, "end": B}], EPSG)
+    old = _zips(root)
+    assert sorted(seen["download"]) == old == sorted(first["sheets"]["downloaded"])
+    assert first["satellite"] == "done" and list(first["years"]) == [2026]
+    # 작업 좌표계 .prj — B 좌표계 창구가 읽음
+    assert resolve_project_crs(root) == f"EPSG:{EPSG}"
+
+    seen["download"].clear()
+    segs = [{"year": 2026, "start": A, "end": B}, {"year": 2027, "start": B, "end": C}]
+    second = terrain.ensure_plan_terrain(root, segs, EPSG)
+    now = _zips(root)
+    assert set(old) <= set(now)  # 옛 도엽 그대로(prune 없음)
+    assert sorted(seen["download"]) == sorted(set(now) - set(old))  # 새 도엽만 받음
+    assert seen["merge"][-1] == now  # 병합 = 옛 ∪ 새
+    # 위성 = 옛 범위 ∪ 새 범위
+    first_box, union = seen["satellite"][0], seen["satellite"][-1]
+    assert union["x"][0] <= first_box["x"][0] and union["x"][1] >= first_box["x"][1]
+    assert union["y"][0] <= first_box["y"][0] and union["y"][1] >= first_box["y"][1]
+    assert [d[0] for d in seen["dem"][-2:]] == ["models_2026", "models_2027"]
+    assert second["sheets"]["failed"] == {}
+
+    # 같은 지점 다시 — 받기 0 · 위성 건너뜀 · 병합 열쇠 같음
+    seen["download"].clear()
+    merges = len(seen["merge"])
+    third = terrain.ensure_plan_terrain(root, segs, EPSG)
+    assert seen["download"] == [] and third["satellite"] == "skip"
+    assert len(seen["merge"]) == merges
+
+
+def test_far_points_leave_no_gap(fake) -> None:
+    root, _ = fake
+    far = (A[0] + 9000.0, A[1] + 7000.0)  # 11 km — 3×3 두 벌이 안 닿는 거리
+    got = terrain.ensure_plan_terrain(root, [{"year": 2026, "start": A, "end": far}], EPSG)
+    to_wgs = Transformer.from_crs(f"EPSG:{EPSG}", "EPSG:4326", always_xy=True)
+    on_line = []
+    for i in range(201):
+        lon, lat = to_wgs.transform(
+            A[0] + (far[0] - A[0]) * i / 200, A[1] + (far[1] - A[1]) * i / 200
+        )
+        on_line.append((lat, lon))
+    assert set(sheets_for_points(on_line)) <= set(got["sheets"]["merged"])
+
+
+def test_dem_outline_covers_corridor() -> None:
+    box = terrain.dem_outline(A, B)
+    reach = terrain.PLAN_CORRIDOR_M + terrain.DEM_EXTRA_M
+    assert box[0][0] - SHEET_SURFACE_MARGIN_M <= min(A[0], B[0]) - reach + 1e-9
+    assert box[1][1] + SHEET_SURFACE_MARGIN_M >= max(A[1], B[1]) + reach - 1e-9
+    samples = terrain.segment_samples(A, C)
+    steps = [math.dist(p, q) for p, q in zip(samples, samples[1:])]
+    assert samples[0] == A and samples[-1] == C and max(steps) <= terrain.SAMPLE_STEP_M

resources/tester/test_530_2_plan_analyze.py

@@ -0,0 +1,111 @@
+"""D 예정노선 2단계 분석(PLAN 530-2) — 합성 지형 두 구간 · 경계 · 누가 · 저장본 · 복도 인자.
+
+./venv/Scripts/python.exe -m pytest resources/tester/test_530_2_plan_analyze.py -q
+"""
+
+import numpy as np
+import pytest
+
+from B05_Profile.B05_Profile_Engine_RouteInitial import _POOLS, Terrain
+from B05_Profile.B05_Profile_Engine_RouteInitial_Search import CORRIDOR_M, SearchGrid
+from D03_PlanAnalysis import D03_PlanAnalysis_Analyze as analyze
+
+CRITERIA = {"max_grade": 0.14, "min_radius_m": 12.0, "design_speed": 20}
+SEGMENTS = [
+    {"year": 2026, "start": (60.0, 200.0), "end": (560.0, 230.0)},
+    {"year": 2027, "start": (560.0, 230.0), "end": (1040.0, 160.0)},
+]
+
+
+@pytest.fixture(scope="module")
+def hill() -> Terrain:
+    x = np.arange(0.0, 1102.0, 2.0)
+    y = np.arange(0.0, 402.0, 2.0)
+    xx, yy = np.meshgrid(x, y)
+    z = 30.0 * np.exp(-((xx - 300.0) ** 2 + (yy - 210.0) ** 2) / 80.0**2) + 0.02 * xx
+    return Terrain(x, y, z, np.ones(xx.shape, dtype=bool))
+
+
+@pytest.fixture
+def run(tmp_path, hill):
+    _POOLS.clear()
+    calls = []
+
+    def loader(root, selection, models_dir, cache_dir):
+        calls.append((models_dir.name, cache_dir.name))
+        return hill, []
+
+    def go(criteria=CRITERIA, corridor_m=150.0):
+        return analyze.analyze_plan(
+            tmp_path, SEGMENTS, criteria, loader=loader, corridor_m=corridor_m
+        )
+
+    go.calls = calls
+    yield go
+    _POOLS.clear()
+
+
+def test_boundary_chainage_and_line(run) -> None:
+    results = run()
+    # 연도 구간마다 제 DEM · 제 비용면 캐시
+    assert run.calls == [("models_2026", "cost_2026"), ("models_2027", "cost_2027")]
+    for seg, result in zip(SEGMENTS, results):
+        assert set(result["routes"]) == set(analyze.PLAN_MODES)
+        for route in result["routes"].values():
+            assert "error" not in route
+            # 시종점 그대로 — 연도 경계가 정확히 맞물림
+            assert route["planned"][0] == list(seg["start"])
+            assert route["planned"][-1] == list(seg["end"])
+    assert results[0]["routes"]["gradeline"]["planned"][-1] == results[1]["start"]
+
+    rows = analyze.chain(results, {2026: "earthwork", 2027: None}, 1234.5)
+    assert [r["mode"] for r in rows] == ["earthwork", "gradeline"]
+    assert rows[0]["start_chainage_m"] == 1234.5
+    assert rows[1]["start_chainage_m"] == rows[0]["end_chainage_m"]
+    for result, row in zip(results, rows):
+        length = result["routes"][row["mode"]]["metrics"]["length_m"]
+        assert row["end_chainage_m"] - row["start_chainage_m"] == pytest.approx(length, abs=1e-3)
+
+    line = analyze.joined_line(results, rows)
+    first = results[0]["routes"]["earthwork"]["planned"]
+    second = results[1]["routes"]["gradeline"]["planned"]
+    assert len(line) == len(first) + len(second) - 1  # 경계점 한 번만(꺾인 채)
+    assert line[len(first) - 1] == list(SEGMENTS[0]["end"])
+
+
+def test_same_input_reads_store(run, monkeypatch) -> None:
+    first = run()
+    assert [r["stored"] for r in first] == [False, False]
+
+    def boom(*args, **kwargs):
+        raise AssertionError("저장본이 있는데 다시 셈")
+
+    monkeypatch.setattr(analyze, "generate_initial_route", boom)
+    again = run()
+    assert [r["stored"] for r in again] == [True, True]
+    assert again[1]["routes"] == first[1]["routes"]
+    # 입력(복도 · 기준)이 바뀌면 다시 셈
+    with pytest.raises(AssertionError, match="다시 셈"):
+        run(corridor_m=120.0)
+    with pytest.raises(AssertionError, match="다시 셈"):
+        run(criteria={**CRITERIA, "max_grade": 0.12})
+
+
+def test_corridor_argument_keeps_b_default(hill) -> None:
+    line = [SEGMENTS[0]["start"], SEGMENTS[0]["end"]]
+    base = SearchGrid(line, hill, [], stream_offset_m=10.0)
+    wide = SearchGrid(line, hill, [], stream_offset_m=10.0, corridor_m=180.0)
+    assert base.corridor_m == CORRIDOR_M == 100.0
+    assert wide.cells > base.cells
+
+
+def test_criteria_from_project() -> None:
+    from B05_Profile.B05_Profile_Engine_Grade import (
+        legal_grade_limit_pct,
+        legal_plan_radius_min_m,
+    )
+
+    got = analyze.plan_criteria("work", 99)  # 못 고르는 속도 → 종류 기본
+    assert got["design_speed"] == 20
+    assert got["max_grade"] == legal_grade_limit_pct("work", "special", False, 20) / 100.0
+    assert got["min_radius_m"] == legal_plan_radius_min_m(20, "special")

resources/tester/test_530_2_plan_router.py

@@ -0,0 +1,127 @@
+"""D 분석 API(PLAN 530-2) — GET 이은 누가 · PUT [저장] = 서버 저장본으로 다시 셈 · 모르는 갈래 거절.
+
+./venv/Scripts/python.exe -m pytest resources/tester/test_530_2_plan_router.py -q
+"""
+
+import json
+
+import pytest
+from fastapi import FastAPI
+from fastapi.testclient import TestClient
+
+from common_util.common_util_json import atomic_write_json
+from D03_PlanAnalysis import D03_PlanAnalysis_Analyze as analyze
+from D03_PlanAnalysis import D03_PlanAnalysis_Router as router_mod
+
+PID = "12953f5a-5ba1-42ad-a427-8ac857a58bc2"
+P0, P1, P2 = (0.0, 0.0), (300.0, 400.0), (900.0, 400.0)
+
+
+def _route(points: list, length: float) -> dict:
+    return {"planned": points, "nodes": [], "curves": [], "metrics": {"length_m": length}}
+
+
+class _Cursor:
+    def __init__(self, log):
+        self.log = log
+
+    async def __aenter__(self):
+        return self
+
+    async def __aexit__(self, *exc):
+        return False
+
+    async def execute(self, sql, args=None):
+        self.log.append((sql, args))
+
+
+class _Connection(_Cursor):
+    async def begin(self):
+        pass
+
+    async def commit(self):
+        self.log.append(("COMMIT", None))
+
+    async def rollback(self):
+        pass
+
+    def cursor(self, *args):
+        return _Cursor(self.log)
+
+
+class _Pool:
+    def __init__(self, log):
+        self.log = log
+
+    def acquire(self):
+        return _Connection(self.log)
+
+
+@pytest.fixture
+def client(tmp_path, monkeypatch):
+    segments = [
+        {"year": 2026, "start": P0, "end": P1, "epsg": 5186, "route_json": None},
+        {"year": 2027, "start": P1, "end": P2, "epsg": 5186, "route_json": '{"mode": "gentle"}'},
+    ]
+    for seg, routes in (
+        (segments[0], {"gradeline": _route([[0, 0], [100, 200], [300, 400]], 530.0)}),
+        (
+            segments[1],
+            {
+                "gradeline": _route([[300, 400], [900, 400]], 600.0),
+                "gentle": _route([[300, 400], [600, 500], [900, 400]], 640.0),
+                "earthwork": {"error": "길 없음"},
+            },
+        ),
+    ):
+        atomic_write_json(
+            analyze.derived_dir(tmp_path) / f"route_{seg['year']}.json",
+            {
+                "year": seg["year"],
+                "start": list(seg["start"]),
+                "end": list(seg["end"]),
+                "fingerprint": "f",
+                "routes": routes,
+            },
+        )
+    project = {"road_type": "work", "design_speed_kph": None, "route_start_m": 1000.0}
+
+    async def inputs(project_id):
+        return tmp_path, project, segments
+
+    log: list = []
+    monkeypatch.setattr(router_mod, "_plan_inputs", inputs)
+    monkeypatch.setattr(router_mod, "get_db_pool", lambda: _Pool(log))
+    app = FastAPI()
+    app.include_router(router_mod.router)
+    test = TestClient(app)
+    test.log = log
+    return test
+
+
+def test_get_chains_saved_choice(client) -> None:
+    body = client.get(f"/api/projects/{PID}/plan/analysis").json()
+    chain = body["chain"]
+    assert [r["mode"] for r in chain] == ["gradeline", "gentle"]  # 저장값 · 없으면 첫 성공
+    assert chain[0]["start_chainage_m"] == 1000.0
+    assert chain[1]["start_chainage_m"] == chain[0]["end_chainage_m"] == 1530.0
+    assert chain[1]["end_chainage_m"] == 2170.0
+    assert body["line"] == [[0, 0], [100, 200], [300, 400], [600, 500], [900, 400]]
+
+
+def test_put_saves_server_routes(client) -> None:
+    got = client.put(f"/api/projects/{PID}/plan/analysis", json={"choices": {"2027": "gradeline"}})
+    assert got.status_code == 200
+    assert got.json()["chain"][1]["end_chainage_m"] == 2130.0
+    updates = [args for sql, args in client.log if sql.startswith("UPDATE plan_route_years")]
+    saved = {args[3]: json.loads(args[0]) for args in updates}
+    assert saved[2027]["mode"] == "gradeline" and saved[2027]["planned"] == [[300, 400], [900, 400]]
+    assert saved[2026]["start_chainage_m"] == 1000.0 and updates[0][1] == "f"
+    assert client.log[-1][0] == "COMMIT"
+
+
+def test_put_rejects_unknown_or_empty_mode(client) -> None:
+    url = f"/api/projects/{PID}/plan/analysis"
+    assert client.put(url, json={"choices": {"2027": "follow"}}).status_code == 400
+    assert client.put(url, json={"choices": {"2027": "earthwork"}}).status_code == 409
+    assert not [sql for sql, _ in client.log if sql.startswith("UPDATE")]