feat(D03): 예정노선 겹 길 임도 프로젝트도 · 지형 원천 인자 · 받은 겹 D03 몫 자리 (PLAN 624-1 · 요구 530 · 531)

- 겹 길(`/plan/layers/*`)이 프로젝트 종류를 안 가림 — 예정노선 = 받은 범위 · 수치지형도 · 임도 = B04 배경 지도 상자 · 설계 지형
- `terrain=sheet|lidar|design` 인자(raster · zero-order) · 응답 `source` · 파생 저장본 열쇠 = 원천 격자 파일 이름
- 라이다 0차곡 = 격자 통째(2 m 솎음 · 파생 저장본) · 수치지형도 = 창고 도엽 선 그대로
- 받은 겹(지적 · 보호구역 · 산림입지 · 단층 · 지질구조선 · 산사태) 자리 = `D03_PlanAnalysis/layers/` — B04 `processed/` 파일 덮음 0 · D 옛 자리 파일은 한 번 복사
- 새 `D03_PlanAnalysis_TerrainSource.py` · 라우터 661 → 658 줄 · 시험 `test_624_1_layers_any_project.py`

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

D03_PlanAnalysis/D03_PlanAnalysis_Layers.py

@@ -3,19 +3,20 @@
 - 도엽 겹(하천중심선 · 도로중심선 …) = 지형 받기 병합본 · 없으면 지난 병합 도엽에서 그 겹만 풀어 냄
 - 지적도 = 브이월드 WFS LX맵(`lt_c_landinfobasemap`) · B04 `request_vworld_wfs` 를 1 km 조각으로 ·
   한 번 1,000 필지 한도 → 꽉 차면 넷으로 다시 쪼갬 · `pnu` 로 겹친 필지 거름 ·
-  저장 = `processed/연속지적도_bounds.geojson`(B04 `/geojson?layer=지적도` 자리) + 곁 열쇠
+  저장 = `<받은 겹 자리>/연속지적도_bounds.geojson` + 곁 열쇠 · 받은 겹 자리 = D03 몫
+  `D03_PlanAnalysis/layers/`(B04 `processed/` 와 이름 겹침 0 · `_TerrainSource` · 624-1)
 - 급경사 = 평활 안 한 사업 DEM(`models_plan/dtm_sheet_<방법>.npz`) · E07 `slope_rad` ·
   기울기 % 값 그림 한 장(1 % 단계 · 회색) + 1 % 칸 수 · 범위 색칠 · 비율 = 화면(PLAN 603) ·
   파생 저장본(열쇠 = 격자 지문)
 - 음영기복 = 같은 DEM · B07 `hillshade_png`(PLAN 581-1)
 - 보호 · 규제 구역 = 브이월드 WFS 여덟 종(겹 이름 = `_분석_440-C` 표 1) ·
-  받은 범위로 잘라 종류 · 이름 · 넓이만 · 저장 = `processed/보호구역_bounds.geojson` + 곁 열쇠 ·
+  받은 범위로 잘라 종류 · 이름 · 넓이만 · 저장 = `layers/보호구역_bounds.geojson` + 곁 열쇠 ·
   종류 하나 실패 = 그 종류만 빠짐
 - 산림입지도 = 브이월드 WFS `lt_c_fsdifrsts`(원자료 산림청 산림입지토양도 · PLAN 581-2) ·
   받은 범위로 잘라 토양형 코드 · 이름 · 넓이만 ·
-  저장 = `processed/산림입지도_bounds.geojson` + 곁 열쇠
+  저장 = `layers/산림입지도_bounds.geojson` + 곁 열쇠
 - 선 겹 = 브이월드 WFS 단층 `lt_l_gimsfault` · 지질구조선 `lt_l_gimslinea`(PLAN 584-1) ·
-  받은 범위로 잘라 가닥 · 길이만 · 저장 = `processed/<이름>_bounds.geojson` + 곁 열쇠
+  받은 범위로 잘라 가닥 · 길이만 · 저장 = `layers/<이름>_bounds.geojson` + 곁 열쇠
 - 0차곡 = `D03_PlanAnalysis_ZeroOrder`
 - 계류 100 m · 도로 300 m 띠 = 도엽 하천중심선(세류 뺌) · 도로중심선(미분류 · 소로 ·
   면리간도로 뺌) 합친 띠 · 받은 범위로 자름 · 파생 저장본(열쇠 = 두 파일 지문)
@@ -253,10 +254,11 @@ def hillshade_layer_png(z: np.ndarray, valid: np.ndarray, cell_m: float) -> dict
 
 
 def slope_layer(
-    project_root: Path, models_dir: Path, epsg: int, kind: str = "slope"
+    project_root: Path, models_dir: Path, epsg: int, kind: str = "slope", npz: Path | None = None
 ) -> dict[str, Any]:
-    """급경사(`slope`) · 음영기복(`hillshade`) — 지문(격자 파일 · 판) 같으면 저장본."""
-    npz = models_dir / f"dtm_sheet_{SHEET_SURFACE_DEFAULT_METHOD}.npz"
+    """급경사(`slope`) · 음영기복(`hillshade`) — 지문(격자 파일 · 판) 같으면 저장본 ·
+    `npz` = 지형 원천 격자(624-1 · 열쇠의 파일 이름 = 원천) · 없으면 D 수치지형도."""
+    npz = npz or models_dir / f"dtm_sheet_{SHEET_SURFACE_DEFAULT_METHOD}.npz"
     stamp = file_stamp(npz)
     if stamp is None:
         raise FileNotFoundError("사업 지형면이 없습니다 — 지형 받기 뒤 보입니다.")

D03_PlanAnalysis/D03_PlanAnalysis_Router.py

@@ -30,7 +30,7 @@ from pathlib import Path
 from typing import Any
 from uuid import UUID, uuid4
 
-from fastapi import APIRouter, File, HTTPException, Request, Response, UploadFile
+from fastapi import APIRouter, File, HTTPException, Query, Request, Response, UploadFile
 from pydantic import BaseModel
 
 from B03_FileInput.B03_FileInput_Repository import get_project_storage_relative_path
@@ -48,6 +48,7 @@ from D03_PlanAnalysis import D03_PlanAnalysis_Network as network
 from D03_PlanAnalysis import D03_PlanAnalysis_SheetTiles as sheet_tiles
 from D03_PlanAnalysis import D03_PlanAnalysis_Store as store
 from D03_PlanAnalysis import D03_PlanAnalysis_Terrain as terrain
+from D03_PlanAnalysis import D03_PlanAnalysis_TerrainSource as terrain_source
 from D03_PlanAnalysis import D03_PlanAnalysis_ZeroOrder as zero_order
 
 logger = logging.getLogger(__name__)
@@ -282,13 +283,9 @@ async def plan_progress(project_id: UUID) -> dict[str, Any]:
     return {"status": "success", **jobs}
 
 
-async def _layer_root(project_id: UUID) -> tuple[Path, int, terrain.Box]:
-    """(실경로, 좌표계, 받은 범위) — 지형 받기 전이면 404."""
-    root, epsg, _, _ = await _plan_ends(project_id)
-    box = terrain.received_box(root, epsg)
-    if box is None:
-        raise HTTPException(status_code=404, detail="지형 받기 뒤 보입니다.")
-    return root, epsg, box
+async def _layer_root(project_id: UUID, source: str | None = None) -> terrain_source.Frame:
+    """겹 틀 — 임도 프로젝트도 · 지형 원천(624-1 · `TerrainSource.layer_frame`)."""
+    return await terrain_source.layer_frame(project_id, source, _plan_ends)
 
 
 SHEET_CLIP_KIND = "plan_sheet"
@@ -335,7 +332,7 @@ async def plan_layer_sheet(project_id: UUID, key: str, request: Request) -> Resp
     받은 범위로 자름(475 · 범위가 늘면 새 사본)."""
     from B04_PreProcess.B04_PreProcess_Router_GIS import _display_copy
 
-    root, epsg, box = await _layer_root(project_id)
+    root, epsg, box, *_ = await _layer_root(project_id)
     try:
         path = await asyncio.to_thread(layers.ensure_sheet_layer, root, key)
     except KeyError as error:
@@ -389,8 +386,7 @@ async def plan_contour_tile(project_id: UUID, z: int, x: int, y: int) -> Respons
 @router.get("/{project_id}/plan/layers/cadastral", response_model=None)
 async def plan_layer_cadastral(project_id: UUID, request: Request) -> Response:
     """지적도 GeoJSON(브이월드 LX맵 · 참고용) — 받은 범위를 덮지 않으면 지금 받음."""
-    root, epsg, box = await _layer_root(project_id)
-    processed = root / terrain.STAGE / "processed"
+    _, epsg, box, processed, *_ = await _layer_root(project_id)
     try:
         got = await asyncio.to_thread(layers.ensure_cadastral, processed, box, epsg)
     except RuntimeError as error:
@@ -399,13 +395,16 @@ async def plan_layer_cadastral(project_id: UUID, request: Request) -> Response:
 
 
 @router.get("/{project_id}/plan/layers/raster", response_model=None)
-async def plan_layer_raster(project_id: UUID, kind: str = "slope") -> dict[str, Any] | Response:
+async def plan_layer_raster(
+    project_id: UUID, kind: str = "slope", source: str | None = Query(None, alias="terrain")
+) -> dict[str, Any] | Response:
     """급경사 기울기 % 값 그림 + 1 % 칸 수(603 · PNG 바이트 + 머리 `X-Plan-Raster` 616-4) ·
     음영기복(`hillshade`) — 격자 지문 같으면 저장본 ·
-    산사태위험지도(`landslide` · 행안부 WMS) — 받은 범위를 덮으면 저장본."""
+    산사태위험지도(`landslide` · 행안부 WMS) — 받은 범위를 덮으면 저장본 ·
+    `terrain` = 지형 원천(sheet · lidar · design · 624-1)."""
+    frame = await _layer_root(project_id, source)
     if kind == "landslide":
-        root, epsg, box = await _layer_root(project_id)
-        processed = root / terrain.STAGE / "processed"
+        _, epsg, box, processed, *_ = frame
         try:
             value = await asyncio.to_thread(landslide.ensure_landslide, processed, box, epsg)
         except landslide.MissingKeyError as error:
@@ -416,25 +415,24 @@ async def plan_layer_raster(project_id: UUID, kind: str = "slope") -> dict[str,
         return {"status": "success", "kind": kind, **value, "box": box_lonlat}
     if kind not in layers.RASTER_KINDS:
         raise HTTPException(status_code=400, detail=f"없는 그림 겹: {kind}")
-    root, epsg, _ = await _layer_root(project_id)
+    root, epsg, _, _, source, _ = frame
+    npz = terrain_source.need_dem(frame)
     try:
-        value = await asyncio.to_thread(
-            layers.slope_layer, root, terrain.models_dir_of(root), epsg, kind
-        )
+        value = await asyncio.to_thread(layers.slope_layer, root, npz.parent, epsg, kind, npz)
     except FileNotFoundError as error:
         raise HTTPException(status_code=404, detail=str(error)) from error
     if kind == "slope":
         png, meta = layers.raster_bytes(value)
-        head = {"X-Plan-Raster": json.dumps({"kind": kind, **meta}), "Content-Encoding": "identity"}
+        meta = {"kind": kind, "source": source, **meta}
+        head = {"X-Plan-Raster": json.dumps(meta), "Content-Encoding": "identity"}
         return Response(png, media_type="image/png", headers=head)  # PNG = 다시 압축 안 함
-    return {"status": "success", "kind": kind, **value}
+    return {"status": "success", "kind": kind, "source": source, **value}
 
 
 @router.get("/{project_id}/plan/layers/zones", response_model=None)
 async def plan_layer_zones(project_id: UUID, request: Request) -> Response:
     """보호 · 규제 구역 GeoJSON(여덟 종 · `summary` = 종류별 개수 · 넓이) — 범위 밖이면 받음."""
-    root, epsg, box = await _layer_root(project_id)
-    processed = root / terrain.STAGE / "processed"
+    _, epsg, box, processed, *_ = await _layer_root(project_id)
     try:
         got = await asyncio.to_thread(layers.ensure_zones, processed, box, epsg)
     except RuntimeError as error:
@@ -445,8 +443,7 @@ async def plan_layer_zones(project_id: UUID, request: Request) -> Response:
 @router.get("/{project_id}/plan/layers/soils", response_model=None)
 async def plan_layer_soils(project_id: UUID, request: Request) -> Response:
     """산림입지도 GeoJSON(토양형 · `summary` = 토양형별 개수 · 넓이) — 범위 밖이면 받음."""
-    root, epsg, box = await _layer_root(project_id)
-    processed = root / terrain.STAGE / "processed"
+    _, epsg, box, processed, *_ = await _layer_root(project_id)
     try:
         got = await asyncio.to_thread(layers.ensure_soils, processed, box, epsg)
     except RuntimeError as error:
@@ -457,7 +454,7 @@ async def plan_layer_soils(project_id: UUID, request: Request) -> Response:
 @router.get("/{project_id}/plan/layers/bands", response_model=None)
 async def plan_layer_bands(project_id: UUID) -> dict[str, Any]:
     """계류 100 m · 도로 300 m 띠(넓이 · 받은 범위 비율) — 두 도엽 겹 지문 같으면 저장본."""
-    root, epsg, box = await _layer_root(project_id)
+    root, epsg, box, *_ = await _layer_root(project_id)
     try:
         value = await asyncio.to_thread(layers.band_layer, root, epsg, box)
     except FileNotFoundError as error:
@@ -470,8 +467,7 @@ async def plan_layer_lines(project_id: UUID, kind: str, request: Request) -> Res
     """선 겹 GeoJSON(단층 · 지질구조선 · `summary` = 가닥 · 길이 m) — 범위 밖이면 받음."""
     if kind not in layers.LINE_KINDS:
         raise HTTPException(status_code=400, detail=f"없는 선 겹: {kind}")
-    root, epsg, box = await _layer_root(project_id)
-    processed = root / terrain.STAGE / "processed"
+    _, epsg, box, processed, *_ = await _layer_root(project_id)
     try:
         got = await asyncio.to_thread(layers.ensure_lines, processed, box, epsg, kind)
     except RuntimeError as error:
@@ -484,22 +480,24 @@ async def plan_layer_zero_order(
     project_id: UUID,
     min_m2: float = zero_order.ZERO_ORDER_MIN_M2,
     max_m2: float = zero_order.ZERO_ORDER_MAX_M2,
+    source: str | None = Query(None, alias="terrain"),
 ) -> Response:
     """0차곡 선(지형 추정 · 위 집수 `min_m2` ~ `max_m2`) — 창고 도엽 선 · 다른 임계는 저장본 ·
     창고에 없어 셈이 필요하면 서버 본체 밖 일꾼(616-3 · 다른 겹 요청을 안 막음)."""
     if not 0 < min_m2 < max_m2:
         raise HTTPException(status_code=400, detail="0 < min_m2 < max_m2 여야 합니다.")
-    root, epsg, _ = await _layer_root(project_id)
-    args = (root, terrain.models_dir_of(root), epsg, min_m2, max_m2)
+    frame = await _layer_root(project_id, source)
+    work, light, full = terrain_source.zero_order_calls(frame, min_m2, max_m2)
     pool = zero_order.heavy_pool()
     try:
-        value = await asyncio.to_thread(zero_order.zero_order_layer, *args, False)
+        value = await asyncio.to_thread(work, *light)
         if value is None:
-            loop = asyncio.get_running_loop()
-            value = await loop.run_in_executor(pool, zero_order.zero_order_layer, *args)
+            value = await asyncio.get_running_loop().run_in_executor(pool, work, *full)
     except FileNotFoundError as error:
         raise HTTPException(status_code=404, detail=str(error)) from error
-    body = json.dumps({"status": "success", **value}, ensure_ascii=False)  # 큰 선 묶음 = 바로 글로
+    body = json.dumps(
+        {"status": "success", "source": frame.source, **value}, ensure_ascii=False
+    )  # 큰 선 묶음 = 바로 글로
     return Response(body, media_type="application/json")
 
 
@@ -591,8 +589,7 @@ async def plan_layer_forest_roads_upload(
 async def plan_layer_landslide_zones(project_id: UUID) -> dict[str, Any]:
     """산사태취약지역(시도 표 · 점 + 지번 맞는 필지 테두리 · 표 없는 시도) — 지적도 못 받으면
     점만."""
-    root, epsg, box = await _layer_root(project_id)
-    processed = root / terrain.STAGE / "processed"
+    root, epsg, box, processed, *_ = await _layer_root(project_id)
     try:
         got = await asyncio.to_thread(layers.ensure_cadastral, processed, box, epsg)
         cadastral = got["path"]

D03_PlanAnalysis/D03_PlanAnalysis_Store.py

@@ -64,7 +64,8 @@ def save_plan(root: str | Path, plan: dict[str, Any]) -> None:
     atomic_write_json(plan_path(root), _clean(plan))
 
 
-async def _project(cursor: Any, project_id: UUID | str) -> dict[str, Any]:
+async def _project(cursor: Any, project_id: UUID | str, plan_only: bool = True) -> dict[str, Any]:
+    """프로젝트 줄 — `plan_only` = 예정노선 아니면 400(지도 겹 길은 종류 안 가림 · 624-1)."""
     await cursor.execute(
         "SELECT project_kind, project_year, storage_path FROM projects"
         " WHERE id = %s AND deleted_at IS NULL",
@@ -73,7 +74,7 @@ async def _project(cursor: Any, project_id: UUID | str) -> dict[str, Any]:
     project = await cursor.fetchone()
     if not project:
         raise HTTPException(status_code=404, detail="프로젝트를 찾을 수 없습니다.")
-    if project["project_kind"] != "PLAN" or project["project_year"] is None:
+    if plan_only and (project["project_kind"] != "PLAN" or project["project_year"] is None):
         raise HTTPException(status_code=400, detail="예정노선 프로젝트가 아닙니다.")
     if not project["storage_path"]:
         raise HTTPException(status_code=404, detail="프로젝트 저장 경로를 찾을 수 없습니다.")
@@ -91,6 +92,15 @@ async def load_plan(project_id: UUID | str) -> tuple[Path, dict[str, Any]]:
     return root, read_plan(root) or empty_plan()
 
 
+async def project_row(project_id: UUID | str) -> dict[str, Any]:
+    """{project_kind, project_year, storage_path} — 종류 안 가림 · 없음 404."""
+    async with (
+        get_db_pool().acquire() as connection,
+        connection.cursor(aiomysql.DictCursor) as cursor,
+    ):
+        return await _project(cursor, project_id, plan_only=False)
+
+
 def plan_units(plan: dict[str, Any]) -> list[dict[str, Any]]:
     """조각마다 {id · line · line_name · year · epsg · from_m · to_m · length_m · planned ·
     stations}

D03_PlanAnalysis/D03_PlanAnalysis_TerrainSource.py

@@ -0,0 +1,179 @@
+"""D03 지도 겹의 프로젝트 틀 · 지형 원천 (요구 530 · 531 · PLAN 624-1).
+
+겹 길(`/plan/layers/*`)은 프로젝트 종류를 안 가림:
+
+- 예정노선(PLAN) = 받은 범위 · 지형 = 수치지형도(`models_plan/dtm_sheet_<방법>.npz`) ·
+  라이다 = D 에 LAS 올리기(624-4) 전이라 없음(404)
+- 그 밖(임도 등) = 범위 = B04 배경 지도 상자(`vworld_white_meta.json` · 없으면 원천 격자 상자) ·
+  지형 = 설계 지형(stage 1 params · 기본) · `lidar` = 라이다 확정 지표면 ·
+  `sheet` = 수치지형도 서피스 ·
+  평활 고르기 = 서버 표본기와 같음(`B05_Profile_Router_Terrain.dtm_grid_npz`)
+- 원천 인자 `terrain=sheet|lidar|design`(빈 값 = design) · 파생 저장본 열쇠 = 격자 파일 이름(원천)
+- 받은 겹 저장 = 종류 상관없이 `D03_PlanAnalysis/layers/`(B04 `processed/` 파일과 이름 겹침 0) ·
+  D 옛 자리(`B04_PreProcess/processed/`) 파일은 새 자리에 없을 때 한 번 복사(다시 받지 않게)
+
+계산 자리 = ③ 서버 단독.
+"""
+
+from __future__ import annotations
+
+import json
+import shutil
+from pathlib import Path
+from typing import Any, Callable, NamedTuple
+
+import numpy as np
+from fastapi import HTTPException
+
+from common_util.common_util_storage import resolve_stored_project_path
+from config.config_system_terrain import SHEET_SURFACE_DEFAULT_METHOD
+from D03_PlanAnalysis import D03_PlanAnalysis_Landslide as landslide
+from D03_PlanAnalysis import D03_PlanAnalysis_Layers as layers
+from D03_PlanAnalysis import D03_PlanAnalysis_Store as store
+from D03_PlanAnalysis import D03_PlanAnalysis_Terrain as terrain
+from D03_PlanAnalysis import D03_PlanAnalysis_ZeroOrder as zero_order
+
+Box = tuple[float, float, float, float]
+
+SOURCES = ("sheet", "lidar", "design")
+LAYER_DIR = Path("D03_PlanAnalysis") / "layers"
+OLD_DIR = Path("B04_PreProcess") / "processed"
+ROAD_MODELS = Path("B04_PreProcess") / "models"
+ROAD_META = ("vworld_white_meta.json", "vworld_satellite_meta.json")
+
+
+def _legacy_names() -> list[str]:
+    names = [layers.CADASTRAL_FILE, layers.CADASTRAL_KEY, layers.ZONE_FILE, layers.ZONE_KEY]
+    names += [layers.SOIL_FILE, layers.SOIL_KEY, landslide.LANDSLIDE_FILE]
+    for label, _ in layers.LINE_KINDS.values():
+        names += [f"{label}_bounds.geojson", f"{label}_bounds.key.json"]
+    return names
+
+
+class Frame(NamedTuple):
+    root: Path
+    epsg: int
+    box: Box
+    store: Path  # 받은 겹 자리(D03 몫)
+    source: str  # 실제 원천 — sheet · lidar
+    dem: Path | None  # 원천 격자(없으면 None)
+
+
+def check_source(choice: str | None) -> str:
+    if choice in (None, ""):
+        return "design"
+    if choice not in SOURCES:
+        raise HTTPException(status_code=400, detail=f"없는 지형 원천: {choice}")
+    return choice
+
+
+def layer_store(root: Path, plan: bool) -> Path:
+    """받은 겹 자리 — D 는 옛 자리 파일을 처음 한 번 복사(덮지 않음)."""
+    store = Path(root) / LAYER_DIR
+    store.mkdir(parents=True, exist_ok=True)
+    if plan:
+        for name in _legacy_names():
+            old, new = Path(root) / OLD_DIR / name, store / name
+            if old.is_file() and not new.exists():
+                shutil.copy2(old, new)
+    return store
+
+
+def plan_frame(root: Path, epsg: int, box: Box, choice: str | None) -> Frame:
+    """예정노선 — 수치지형도 하나(design = sheet) · lidar = 없음."""
+    source = check_source(choice)
+    source = "sheet" if source == "design" else source
+    npz = terrain.models_dir_of(Path(root)) / f"dtm_sheet_{SHEET_SURFACE_DEFAULT_METHOD}.npz"
+    return Frame(
+        Path(root), epsg, box, layer_store(root, True), source, npz if source == "sheet" else None
+    )
+
+
+def road_dem(models: Path, params: dict[str, Any], source: str) -> tuple[str, Path | None]:
+    """(실제 원천, 격자) — 고른 원천 params 의 `dtm_<필터>`(평활이면 `_smooth` 먼저) ·
+    라이다 없는 프로젝트의 lidar = None."""
+    from common_util.common_util_surface_confirmation import (
+        analysis_terrain_params,
+        design_terrain,
+    )
+
+    want = None if source == "design" else ("las" if source == "lidar" else "sheet")
+    picked = analysis_terrain_params(params, want)
+    got = "sheet" if design_terrain(picked) == "sheet" else "lidar"
+    if source != "design" and got != source:
+        return source, None
+    npz = Path(models) / f"dtm_{picked.get('source_filter') or ''}.npz"
+    smooth = npz.with_name(f"{npz.stem}_smooth.npz")
+    npz = smooth if picked.get("smooth") and smooth.is_file() else npz
+    return got, npz if npz.is_file() else None
+
+
+def grid_box(npz: Path) -> Box:
+    with np.load(npz) as dem:
+        x, y = np.asarray(dem["x"], float), np.asarray(dem["y"], float)
+    half = abs(float(x[1] - x[0])) / 2 if len(x) > 1 else 0.5
+    return (x[0] - half, y[0] - half, x[-1] + half, y[-1] + half)
+
+
+def road_box(root: Path, dem: Path | None) -> Box | None:
+    """B04 배경 지도 상자(작업 좌표계 m) · 없으면 격자 상자."""
+    for name in ROAD_META:
+        try:
+            meta = json.loads((Path(root) / OLD_DIR / name).read_text(encoding="utf-8"))
+            return (meta["x_min"], meta["y_min"], meta["x_max"], meta["y_max"])
+        except (OSError, ValueError, KeyError, TypeError):
+            continue
+    return grid_box(dem) if dem is not None else None
+
+
+async def road_frame(row: dict[str, Any], project_id: Any, choice: str | None) -> Frame:
+    """임도 등 — 범위 · 좌표계 · 설계 지형 params 로 원천 격자."""
+    from pyproj import CRS
+
+    from common_util.common_util_crs import identify_epsg, resolve_project_crs
+    from common_util.common_util_surface_confirmation import get_surface_confirmation_params
+    from config.config_db import get_db_pool
+
+    source = check_source(choice)
+    root = Path(resolve_stored_project_path(row["storage_path"]))
+    async with get_db_pool().acquire() as connection:
+        params = await get_surface_confirmation_params(connection, str(project_id))
+    got, dem = road_dem(root / ROAD_MODELS, params, source)
+    box = road_box(root, dem)
+    if box is None:
+        raise HTTPException(status_code=404, detail="지형 전처리 뒤 보입니다.")
+    text = resolve_project_crs(root)
+    epsg = identify_epsg(CRS.from_user_input(text), text)
+    if epsg is None:
+        raise HTTPException(status_code=404, detail="프로젝트 좌표계를 모릅니다.")
+    return Frame(root, epsg, tuple(float(v) for v in box), layer_store(root, False), got, dem)
+
+
+def need_dem(frame: Frame) -> Path:
+    if frame.dem is None:
+        what = "라이다" if frame.source == "lidar" else "수치지형도"
+        raise HTTPException(status_code=404, detail=f"{what} 지형이 없습니다.")
+    return frame.dem
+
+
+async def layer_frame(project_id: Any, source: str | None, plan_ends: Callable) -> Frame:
+    """겹 틀 — 예정노선 = 받은 범위(받기 전 404) · 그 밖 = `road_frame`."""
+    row = await store.project_row(project_id)
+    if row["project_kind"] != "PLAN":
+        return await road_frame(row, project_id, source)
+    root, epsg, _, _ = await plan_ends(project_id)
+    box = terrain.received_box(root, epsg)
+    if box is None:
+        raise HTTPException(status_code=404, detail="지형 받기 뒤 보입니다.")
+    return plan_frame(root, epsg, box, source)
+
+
+def zero_order_calls(frame: Frame, min_m2: float, max_m2: float) -> tuple[Callable, tuple, tuple]:
+    """(셈, 가벼운 인자 = 저장본만, 무거운 인자) — 수치지형도 = 창고 도엽 선(범위 = 원천 격자
+    상자) · 라이다 = 격자 통째."""
+    npz = need_dem(frame)
+    if frame.source == "sheet":
+        args = (frame.root, npz.parent, frame.epsg, min_m2, max_m2)
+        return zero_order.zero_order_layer, (*args, False, npz), (*args, True, npz)
+    args = (frame.root, npz, frame.epsg, min_m2, max_m2)
+    return zero_order.dem_zero_order_layer, (*args, False), args

D03_PlanAnalysis/D03_PlanAnalysis_ZeroOrder.py

@@ -450,10 +450,12 @@ def zero_order_layer(
     min_m2: float = ZERO_ORDER_MIN_M2,
     max_m2: float = ZERO_ORDER_MAX_M2,
     compute: bool = True,
+    npz: Path | None = None,
 ) -> dict[str, Any] | None:
     """0차곡 겹 — 받은 범위의 창고 도엽 선(기본 임계) 또는 누적(다른 임계 · 지문 같으면 저장본) ·
-    경위도 선 + 가닥 · 길이 합 m. `compute=False` = 셈이 필요하면 None(일꾼으로 넘김)."""
-    npz = models_dir / f"dtm_sheet_{SHEET_SURFACE_DEFAULT_METHOD}.npz"
+    경위도 선 + 가닥 · 길이 합 m. `compute=False` = 셈이 필요하면 None(일꾼으로 넘김) ·
+    `npz` = 범위를 줄 수치지형도 격자(624-1 · 임도 프로젝트) · 없으면 D `models_plan`."""
+    npz = npz or models_dir / f"dtm_sheet_{SHEET_SURFACE_DEFAULT_METHOD}.npz"
     if file_stamp(npz) is None:
         raise FileNotFoundError("사업 지형면이 없습니다 — 지형 받기 뒤 보입니다.")
     box = _project_box(npz)
@@ -497,6 +499,58 @@ def zero_order_layer(
     return value
 
 
+DEM_ZERO_ORDER_KIND = "plan_zero_order_dem"
+
+
+def dem_zero_order_layer(
+    project_root: Path,
+    npz: Path,
+    epsg: int,
+    min_m2: float = ZERO_ORDER_MIN_M2,
+    max_m2: float = ZERO_ORDER_MAX_M2,
+    compute: bool = True,
+) -> dict[str, Any] | None:
+    """0차곡 겹 — 라이다 지형 격자 통째(624-1 · 창고 없음) · 2 m 로 솎아 흐름누적 · 선 ·
+    파생 저장본(열쇠 = 격자 파일 이름 · 지문 · 임계). `compute=False` = 저장본 없으면 None."""
+    import shapely
+
+    stamp = file_stamp(npz)
+    if stamp is None:
+        raise FileNotFoundError("지형 격자가 없습니다.")
+    key = derived_key(
+        DEM_ZERO_ORDER_KIND,
+        ZERO_ORDER_VERSION,
+        npz.name,
+        stamp,
+        int(epsg),
+        float(min_m2),
+        float(max_m2),
+        ZERO_ORDER_CELL_M,
+        ZERO_ORDER_MIN_LEN_M,
+        SIMPLIFY_M,
+    )
+    saved = read_derived(project_root, DEM_ZERO_ORDER_KIND, key)
+    if saved is not None or not compute:
+        return saved["value"] if saved is not None else None
+    with np.load(npz) as dem:
+        x, y = np.asarray(dem["x"], float), np.asarray(dem["y"], float)
+        cell = float(dem["resolution"][0]) if "resolution" in dem.files else float(x[1] - x[0])
+        step = max(1, round(ZERO_ORDER_CELL_M / cell))
+        z = np.asarray(dem["z"], float)[::step, ::step]
+        valid = np.asarray(dem["valid_mask"], bool)[::step, ::step]
+    x, y, cell = x[::step], y[::step], cell * step
+    lines = zero_order_lines(z, valid, x, y, cell, min_m2, max_m2)
+    shapes: list[Any] = []
+    if lines:
+        index = np.repeat(np.arange(len(lines)), [len(line) for line in lines])
+        geoms = shapely.linestrings(np.concatenate(lines), indices=index)
+        shapes = list(shapely.simplify(geoms, SIMPLIFY_M))
+    value = _layer_value(shapes, epsg, min_m2, max_m2)
+    value["cell_m"] = cell
+    write_derived(project_root, DEM_ZERO_ORDER_KIND, key, value)
+    return value
+
+
 def _clipped_lines(paths: list[Path], box: tuple[float, ...]) -> list[Any]:
     """도엽 선 이어 붙여 받은 범위로 자름 — 잘려 짧아진 토막(`ZERO_ORDER_MIN_LEN_M` 미만) 거름."""
     import shapely

resources/tester/test_624_1_layers_any_project.py

@@ -0,0 +1,196 @@
+"""D03 지도 겹 길 — 임도 프로젝트도 받음 · 지형 원천 인자 · 받은 겹 자리(요구 530 · PLAN 624-1).
+
+- 예정노선(D) · 임도(B) 둘 다 급경사 · 음영 · 지적 받음 · 원천 머리(`source`)
+- 원천별 파생 저장본 열쇠 다름 · 없는 원천 404 · 틀린 원천 400
+- 받은 겹 = `D03_PlanAnalysis/layers/` — B04 `processed/` 파일 안 덮음 · 새 파일 0
+- D 옛 자리 파일은 한 번 복사(다시 안 받음) · D 급경사 값 = 옛 길 값
+- 라이다 0차곡 = 격자 통째 · 저장본
+
+./venv/Scripts/python.exe -m pytest resources/tester/test_624_1_layers_any_project.py -q
+"""
+
+import json
+from concurrent.futures import ThreadPoolExecutor
+from contextlib import asynccontextmanager
+
+import numpy as np
+import pytest
+from fastapi import FastAPI
+from fastapi.testclient import TestClient
+
+from D03_PlanAnalysis import D03_PlanAnalysis_Layers as layers
+from D03_PlanAnalysis import D03_PlanAnalysis_Router as router_mod
+from D03_PlanAnalysis import D03_PlanAnalysis_TerrainSource as source
+from D03_PlanAnalysis import D03_PlanAnalysis_ZeroOrder as zero_order
+
+PID = "12953f5a-5ba1-42ad-a427-8ac857a58bc2"
+EPSG = 5186
+X0, Y0 = 200000.0, 450000.0
+BOX = (X0, Y0, X0 + 400.0, Y0 + 400.0)
+B04_CADASTRAL = b'{"type": "FeatureCollection", "features": [], "b04": true}'
+
+
+def _dem(path, tilt: float, n: int = 60, cell: float = 2.0) -> None:
+    x = X0 + np.arange(n) * cell
+    y = Y0 + np.arange(n) * cell
+    z = np.add.outer(np.zeros(n), np.arange(n) * cell * tilt)
+    path.parent.mkdir(parents=True, exist_ok=True)
+    np.savez(path, x=x, y=y, z=z, valid_mask=np.ones(z.shape, bool), resolution=[cell])
+
+
+class _Pool:
+    @asynccontextmanager
+    async def acquire(self):
+        yield None
+
+
+class FakeWfs:
+    def __init__(self) -> None:
+        self.calls = 0
+
+    def __call__(self, box, epsg, typename):
+        self.calls += 1
+        lon, lat = 127.0, 37.0
+        ring = [[lon, lat], [lon + 1e-4, lat], [lon + 1e-4, lat + 1e-4], [lon, lat]]
+        geom = {"type": "Polygon", "coordinates": [ring]}
+        return [{"type": "Feature", "geometry": geom, "properties": {"pnu": str(box)}}]
+
+
+@pytest.fixture
+def world(tmp_path, monkeypatch):
+    """D 프로젝트(수치지형도) · B 프로젝트(라이다 csf 확정 + 수치지형도 서피스 · B04 지적도)."""
+    import common_util.common_util_crs as crs
+    import common_util.common_util_surface_confirmation as confirm
+    import config.config_db as config_db
+
+    d_root, b_root = tmp_path / "D", tmp_path / "B"
+    _dem(d_root / "B04_PreProcess" / "models_plan" / "dtm_sheet_laplace.npz", 0.3)
+    b_models = b_root / "B04_PreProcess" / "models"
+    _dem(b_models / "dtm_csf.npz", 0.6)
+    _dem(b_models / "dtm_csf_smooth.npz", 0.5)
+    _dem(b_models / "dtm_sheet_laplace_smooth.npz", 0.2)
+    processed = b_root / "B04_PreProcess" / "processed"
+    processed.mkdir(parents=True)
+    (processed / layers.CADASTRAL_FILE).write_bytes(B04_CADASTRAL)
+    meta = {"x_min": BOX[0], "y_min": BOX[1], "x_max": BOX[2], "y_max": BOX[3]}
+    (processed / "vworld_white_meta.json").write_text(json.dumps(meta), "utf-8")
+    kinds = {"D": "PLAN", "B": "ROAD"}
+    state = {"pid": "D"}
+    params = {"source_filter": "csf", "method": "dtm", "smooth": True}
+
+    async def project_row(project_id):
+        return {"project_kind": kinds[state["pid"]], "storage_path": state["pid"]}
+
+    async def plan_ends(project_id, center=None, reset=False):
+        return d_root, EPSG, [], None
+
+    async def surface_params(connection, project_id):
+        return dict(params)
+
+    monkeypatch.setattr(router_mod.store, "project_row", project_row)
+    monkeypatch.setattr(router_mod, "_plan_ends", plan_ends)
+    monkeypatch.setattr(router_mod.terrain, "received_box", lambda root, epsg: BOX)
+    monkeypatch.setattr(source, "resolve_stored_project_path", lambda p: str(tmp_path / p))
+    monkeypatch.setattr(confirm, "get_surface_confirmation_params", surface_params)
+    monkeypatch.setattr(config_db, "get_db_pool", lambda: _Pool())
+    monkeypatch.setattr(crs, "resolve_project_crs", lambda root: f"EPSG:{EPSG}")
+    wfs = FakeWfs()
+    monkeypatch.setattr(layers, "_wfs", wfs)
+    monkeypatch.setattr(zero_order, "heavy_pool", lambda: ThreadPoolExecutor(1))
+    app = FastAPI()
+    app.include_router(router_mod.router)
+    client = TestClient(app)
+
+    def get(pid: str, path: str, **query):
+        state["pid"] = pid
+        return client.get(f"/api/projects/{PID}/plan/layers/{path}", params=query)
+
+    return {"get": get, "D": d_root, "B": b_root, "wfs": wfs, "params": params}
+
+
+def _raster(res) -> dict:
+    assert res.status_code == 200, res.text
+    return json.loads(res.headers["X-Plan-Raster"])
+
+
+def test_both_kinds_get_slope_with_source(world) -> None:
+    get = world["get"]
+    assert _raster(get("D", "raster", kind="slope"))["source"] == "sheet"
+    assert _raster(get("B", "raster", kind="slope"))["source"] == "lidar"  # 설계 지형 = 라이다
+    assert _raster(get("B", "raster", kind="slope", terrain="sheet"))["source"] == "sheet"
+    shade = get("B", "raster", kind="hillshade", terrain="lidar").json()
+    assert shade["source"] == "lidar" and len(shade["box"]) == 4
+    assert get("D", "raster", kind="slope", terrain="lidar").status_code == 404  # 624-4 전
+    assert get("B", "raster", kind="slope", terrain="las").status_code == 400
+
+
+def test_source_keys_differ(world) -> None:
+    get = world["get"]
+    for terrain in ("lidar", "sheet", "design"):
+        get("B", "raster", kind="slope", terrain=terrain)
+    saved = sorted((world["B"] / "_derived" / layers.SLOPE_KIND).glob("*.json"))
+    assert len(saved) == 2  # design = lidar 같은 격자 → 같은 열쇠
+    lidar = _raster(get("B", "raster", kind="slope", terrain="lidar"))
+    sheet = _raster(get("B", "raster", kind="slope", terrain="sheet"))
+    assert lidar["counts"] != sheet["counts"]
+
+
+def test_road_lidar_missing_when_sheet_only(world) -> None:
+    world["params"].update(source_filter="sheet_laplace")  # 라이다 없는 임도
+    assert get_status(world, "lidar") == 404
+    assert get_status(world, "design") == 200
+
+
+def get_status(world, terrain: str) -> int:
+    return world["get"]("B", "raster", kind="slope", terrain=terrain).status_code
+
+
+def test_road_layers_never_touch_b04_files(world) -> None:
+    processed = world["B"] / "B04_PreProcess" / "processed"
+    before = {p.name: p.read_bytes() for p in processed.iterdir()}
+    res = world["get"]("B", "cadastral")
+    assert res.status_code == 200 and res.json()["features"]
+    after = {p.name: p.read_bytes() for p in processed.iterdir()}
+    assert after == before  # 같은 이름 덮음 0 · 새 파일 0
+    mine = world["B"] / source.LAYER_DIR / layers.CADASTRAL_FILE
+    assert mine.is_file() and mine.read_bytes() != B04_CADASTRAL
+
+
+def test_plan_old_files_copied_once(world) -> None:
+    """D 옛 자리(B04 processed) 지적도 + 열쇠 = 새 자리로 복사 · 다시 받지 않음 · 값 같음."""
+    old = world["D"] / "B04_PreProcess" / "processed"
+    old.mkdir(parents=True)
+    body = {"type": "FeatureCollection", "features": [{"type": "Feature", "properties": {}}]}
+    (old / layers.CADASTRAL_FILE).write_text(json.dumps(body), "utf-8")
+    key = {"box": list(BOX), "epsg": EPSG, "count": 1, "failed": 0, "failed_tiles": []}
+    (old / layers.CADASTRAL_KEY).write_text(json.dumps(key), "utf-8")
+    res = world["get"]("D", "cadastral")
+    assert res.status_code == 200 and res.json() == body and world["wfs"].calls == 0
+    assert (world["D"] / source.LAYER_DIR / layers.CADASTRAL_KEY).is_file()
+
+
+def test_plan_slope_same_as_before(world, tmp_path) -> None:
+    """D 급경사 = 옛 길(`slope_layer(root, models_plan, epsg)`) 값 그대로."""
+    res = world["get"]("D", "raster", kind="slope")
+    meta = _raster(res)
+    old = layers.slope_layer(world["D"], world["D"] / "B04_PreProcess" / "models_plan", EPSG)
+    png, old_meta = layers.raster_bytes(old)
+    assert res.content == png and meta == {"kind": "slope", "source": "sheet", **old_meta}
+
+
+def test_lidar_zero_order_whole_grid(world) -> None:
+    """라이다 0차곡 = 격자 통째(1 m → 2 m 솎음) · 저장본 · 수치지형도 = 창고 길."""
+    n, cell = 200, 1.0
+    x, y = X0 + np.arange(n) * cell, Y0 + np.arange(n) * cell
+    valley = np.abs(x - x.mean())[None, :] * 0.3 + (y - Y0)[:, None] * 0.2
+    npz = world["B"] / "B04_PreProcess" / "models" / "dtm_csf_smooth.npz"
+    np.savez(npz, x=x, y=y, z=valley, valid_mask=np.ones(valley.shape, bool), resolution=[cell])
+    res = world["get"]("B", "zero-order", terrain="lidar")
+    got = res.json()
+    assert res.status_code == 200 and got["source"] == "lidar" and got["cell_m"] == 2.0
+    assert got["count"] >= 1 and got["length_m"] > 0
+    again = zero_order.dem_zero_order_layer(world["B"], npz, EPSG, compute=False)
+    assert again is not None and again["count"] == got["count"]
+    frame = source.plan_frame(world["D"], EPSG, BOX, None)
+    work, light, _ = source.zero_order_calls(frame, 1.0, 2.0)
+    assert work is zero_order.zero_order_layer and light[-1] == frame.dem