feat(D 예정노선): 지형 받기 = 공용 지형 창고 도엽 이어 붙임 (PLAN 608-1 · 요구 486)

- 병합 등고선 → 받은 범위 통째 DEM 걷음 · load_terrain(받은 범위 · 2 m) 이어 붙여 models_plan/dtm_sheet_laplace.npz · 미리보기 · 평활본 같은 꼴
- 열쇠 = 이어 붙인 값 지문 · 같으면 다시 안 씀(급경사 · 음영 지문 그대로)
- 창고 진행 = 5/5 단계 진행률에 이어 붙임 · 도엽 zip = 보조 겹만(받기 실패는 경고)
- 시험 530_1 = 가짜 창고 · 값 같음 안 다시 씀 · 값 바뀜 다시 씀

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

D03_PlanAnalysis/D03_PlanAnalysis_Terrain.py

@@ -1,5 +1,5 @@
 """D 예정노선 지형 받기 — 사업 전체 한 벌(도엽 · 위성 · DEM) (요구 345 · 389 · 463 · PLAN 530-1 ·
-561-2 · 595).
+561-2 · 595 · 608-1).
 
 지형은 D 프로젝트 storage 안 `B04_PreProcess/` 배치 그대로(지형 코드 주인 = B04 · 고칠 곳 0):
 
@@ -10,7 +10,9 @@
 - 병합 = 지난번 병합 도엽 ∪ 이번 도엽(옛 도엽 유지 · 열쇠 같으면 건너뜀)
 - 위성 = 범위 도엽 도곽 ∪ 옛 위성 범위(덮으면 건너뜀)
 - 지적도 = 지형 일에서 뺌 — 겹을 켤 때 GET 이 받음(`D03_PlanAnalysis_Layers.ensure_cadastral`)
-- DEM = 한 벌 `models_plan/`(연도는 D03 나눔 뒤에 정해짐) · 받은 범위 = `plan_terrain.json`
+- DEM = 공용 지형 창고(`common_util_terrain_store` · 도엽마다 처음 한 번 셈) 도엽을 받은 범위로
+  이어 붙여 한 벌 `models_plan/`(뒤 소비자가 읽는 꼴 그대로 · 열쇠 = 이어 붙인 값 지문) ·
+  도엽 zip = 보조 겹(하천 · 도로 등)만 · 받은 범위 = `plan_terrain.json`
   (상자 · 여유 · 좌표계 · 위치 칸 가운데)
 - 작업 좌표계 = 지점 epsg · `B03_FileInput/input/prj/plan.prj` 한 장(B04 · B05 좌표계 창구가 읽음)
 - 받기 전 범위 보기 = `plan_extent`(내려받기 없음) · 받기는 사용자가 누를 때만
@@ -20,6 +22,7 @@
 
 from __future__ import annotations
 
+import hashlib
 import json
 import logging
 import math
@@ -39,9 +42,15 @@ from B04_PreProcess.B04_PreProcess_Engine_SheetStore import (
     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_SheetSurface import (
+    _key_path,
+    _method_model_entry,
+    _write_method_model,
+)
 from B04_PreProcess.B04_PreProcess_Engine_VWorld import download_vworld_satellite_map
+from B04_PreProcess.B04_PreProcess_Engine_Watershed_Grid import GridSpec
 from common_util.common_util_json import atomic_write_json
+from common_util.common_util_terrain_store import load_terrain
 from config.config_system_terrain import (
     PLAN_TERRAIN_GRID_M,
     SHEET_SURFACE_DEFAULT_METHOD,
@@ -296,35 +305,27 @@ def ensure_plan_terrain(
     before = {s for s in wanted if get_sheet_path(store, s) is not None}
     report(10, f"2/5 수치지형도 도엽 {len(wanted) - len(before)}매 받는 중")
     got = ensure_sheets(store, wanted)
-    if not got["available"]:
-        raise ValueError(f"수치지형도 도엽을 하나도 받지 못했습니다: {got['failed']}")
+    if not got["available"]:  # 도엽 = 보조 겹만 · 지형은 창고가 셈
+        logger.warning("D 수치지형도 도엽 받기 실패: %s", got["failed"])
     report(45, "3/5 도엽 레이어 병합 중")
     merged = _merge_sheets(processed, store, list(got["available"]), keep_old=have is not None)
     report(60, "4/5 위성 지도 받는 중")
     ring = [(p[1], p[0]) for p in box_ring(box, epsg)]
     satellite = _satellite(processed, prj, ring, epsg, keep_old=have is not None)
 
-    report(65, f"5/5 사업 지형면 만드는 중 — 칸 {cells_of(box) / 1e4:.0f}만 · 수 분 걸릴 수 있음")
+    report(65, f"5/5 사업 지형면 — 칸 {cells_of(box) / 1e4:.0f}만")
     models_dir = models_dir_of(project_root)
     models_dir.mkdir(parents=True, exist_ok=True)
-    # B 절취 = 넘긴 꼭짓점 상자 + `SHEET_SURFACE_MARGIN_M` → 그만큼 안쪽으로 넘김(여유 ≥ 그 값)
-    inner = SHEET_SURFACE_MARGIN_M
-    outline = np.array(
-        [[box[0] + inner, box[1] + inner], [box[2] - inner, box[3] - inner]], dtype=np.float64
-    )
-    models = build_sheet_surface_model(
-        project_root,
-        processed,
-        models_dir,
-        outline,
-        f"EPSG:{int(epsg)}",
-        [SHEET_SURFACE_DEFAULT_METHOD],
-        PLAN_TERRAIN_GRID_M,
-    )
+
+    def store_report(percent: int, message: str) -> None:
+        report(65 + percent * 34 // 100, f"5/5 {message}")
+
+    grid = load_terrain(epsg, box=box, cell_m=PLAN_TERRAIN_GRID_M, on_progress=store_report)
+    models = _write_store_model(project_root, models_dir, grid)
     if not models:
         raise ValueError(
-            "사업 지형면을 만들지 못했습니다 — 범위 안 도엽 등고선 높이 확인"
-            f"(병합 도엽 {len(merged)}매)"
+            "사업 지형면을 만들지 못했습니다 — 범위 안 등고선 확인"
+            f"(도엽 {len(grid['sheets'])}매 · 실패 {grid['failed'] or '없음'})"
         )
     meta: dict[str, Any] = {"box": list(box), "margin_m": margin, "epsg": int(epsg)}
     kept = None if reset else received_meta(project_root).get("center")
@@ -342,4 +343,40 @@ def ensure_plan_terrain(
         "satellite": satellite,
         "box_m": list(box),
         "models": models,
+        "terrain": {"computed": grid["computed"], "failed": grid["failed"]},
     }
+
+
+def _write_store_model(
+    project_root: Path, models_dir: Path, grid: dict[str, Any]
+) -> list[dict[str, Any]]:
+    """창고 격자 → `dtm_sheet_<방법>.npz` · 미리보기 · 평활본(B 꼴) — 값 지문 같으면 그대로."""
+    z, x, y = grid["z"], grid["x"], grid["y"]
+    if not z.size or not grid["valid_mask"].any():
+        return []
+    cell = grid["cell_m"]
+    method = SHEET_SURFACE_DEFAULT_METHOD
+    key = {
+        "source": "terrain_store",
+        "origin": [float(x[0]), float(y[0])],
+        "shape": list(z.shape),
+        "z": hashlib.sha1(np.ascontiguousarray(z).tobytes()).hexdigest(),
+    }
+    entry = _method_model_entry(project_root, models_dir, method, cell)
+    key_path = _key_path(models_dir, method)
+    files = [entry["model_file_path"], *(layer["file_path"] for layer in entry["layers"])]
+    if _read_json(key_path) == key and all((project_root / f).is_file() for f in files):
+        return [entry]
+    key_path.unlink(missing_ok=True)  # 쓰는 도중 끊기면 옛 열쇠가 새 파일을 덮지 않게
+    spec = GridSpec(
+        x_min=float(x[0]) - cell / 2,
+        y_max=float(y[-1]) + cell / 2,
+        cell_m=cell,
+        n_rows=z.shape[0],
+        n_cols=z.shape[1],
+    )
+    model = _write_method_model(project_root, models_dir, spec, z[::-1, :], method, cell)
+    if model is None:
+        return []
+    atomic_write_json(key_path, key)
+    return [model]

resources/tester/test_530_1_plan_terrain.py

@@ -1,6 +1,6 @@
 """D 예정노선 지형 받기(PLAN 530-1 · 561-2 · 595) — 사업 전체 한 벌 · 범위 보기 = 내려받기 없음 ·
 옛 도엽 유지 · 새 도엽만 · 범위는 줄지 않음 · 빠짐 없음 · 지적도 뺌 · 새로(reset) = 옛 범위 버림 ·
-받은 범위 좌표계 · 칸 상한 · 거리.
+받은 범위 좌표계 · 칸 상한 · 거리 · DEM = 공용 지형 창고 이어 붙임(608-1 · 값 같으면 안 다시 씀).
 
 ./venv/Scripts/python.exe -m pytest resources/tester/test_530_1_plan_terrain.py -q
 """
@@ -8,6 +8,7 @@
 import json
 from pathlib import Path
 
+import numpy as np
 import pytest
 from pyproj import Transformer
 
@@ -46,15 +47,22 @@ def fake(tmp_path, monkeypatch):
         (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, grid_m=None):
-        seen["dem"].append((models_dir.name, route_xy.tolist(), crs))
-        seen["grid_m"] = grid_m
-        return [{"model_file_path": f"B04_PreProcess/{models_dir.name}/dtm.npz"}]
+    def load_terrain(epsg, *, sheets=None, box=None, cell_m=2.0, on_progress=None):
+        seen["dem"].append(list(box))
+        seen["grid_m"] = cell_m
+        on_progress(50, "지형 계산")
+        x = (np.arange(10) + np.floor(box[0] / cell_m) + 0.5) * cell_m  # 작은 격자 · 원점 칸 배수
+        y = (np.arange(8) + np.floor(box[1] / cell_m) + 0.5) * cell_m
+        z = seen["z"] + 0.1 * np.arange(80, dtype=np.float32).reshape(8, 10)
+        grid = {"x": x, "y": y, "z": z, "valid_mask": np.isfinite(z), "cell_m": cell_m}
+        return grid | {"epsg": epsg, "sheets": ["1"], "computed": [], "failed": {}}
+
+    seen["z"] = np.float32(100.0)
 
     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)
+    monkeypatch.setattr(terrain, "load_terrain", load_terrain)
     return tmp_path, seen
 
 
@@ -63,10 +71,8 @@ def _zips(root):
 
 
 def _dem_box(seen) -> list[float]:
-    """가짜 DEM 이 받은 꼭짓점 상자 + B 절취 여유 = 실제 DEM 상자."""
-    (x0, y0), (x1, y1) = seen["dem"][-1][1]
-    m = SHEET_SURFACE_MARGIN_M
-    return [x0 - m, y0 - m, x1 + m, y1 + m]
+    """창고에 물은 상자 = 받은 범위 그대로."""
+    return seen["dem"][-1]
 
 
 def test_one_set_and_extend_keeps_old(fake) -> None:
@@ -75,7 +81,13 @@ def test_one_set_and_extend_keeps_old(fake) -> None:
     old = _zips(root)
     assert sorted(seen["download"]) == old == sorted(first["sheets"]["downloaded"])
     assert first["satellite"] == "done" and len(first["models"]) == 1
-    assert seen["dem"][-1][0] == "models_plan"  # 연도 없는 한 벌
+    model = first["models"][0]["model_file_path"]
+    assert model == "B04_PreProcess/models_plan/dtm_sheet_laplace.npz"  # 연도 없는 한 벌
+    with np.load(root / model) as dem:  # 뒤 소비자 꼴 — y 오름차순 · 칸 2 m
+        assert dem["z"].shape == (8, 10) and dem["y"][1] > dem["y"][0]
+        assert float(dem["resolution"][0]) == 2.0 and dem["valid_mask"].all()
+        assert dem["z"][1, 0] == pytest.approx(101.0)  # 창고 값 그대로(뒤집힘 없음)
+    assert (root / model.replace(".npz", "_smooth.npz")).is_file()
     m = terrain.DEFAULT_MARGIN_M
     box = [min(A[0], B[0]) - m, min(A[1], B[1]) - m, max(A[0], B[0]) + m, max(A[1], B[1]) + m]
     assert _dem_box(seen) == pytest.approx(box) == first["box_m"]
@@ -103,6 +115,16 @@ def test_one_set_and_extend_keeps_old(fake) -> None:
     assert seen["download"] == [] and third["satellite"] == "skip"
     assert len(seen["merge"]) == merges and third["box_m"] == grown
 
+    # 창고 값 같음 = DEM 안 다시 씀 · 값 바뀜(창고 다시 셈) = 다시 씀
+    npz = root / third["models"][0]["model_file_path"]
+    stamp = npz.stat().st_mtime_ns
+    terrain.ensure_plan_terrain(root, [A, B], EPSG)
+    assert npz.stat().st_mtime_ns == stamp
+    seen["z"] = np.float32(200.0)
+    terrain.ensure_plan_terrain(root, [A, B], EPSG)
+    with np.load(npz) as dem:
+        assert float(dem["z"][0, 0]) == 200.0
+
 
 def test_reset_drops_old_box_sheets_satellite(fake) -> None:
     """595-2 — 새로 = 옛 받은 범위 · 병합 도엽 · 위성 범위 버림 · 좌표계 · 가운데 기록 ·
@@ -174,7 +196,8 @@ def test_box_sheets_leave_no_gap(fake) -> None:
 
 def test_empty_surface_raises_and_keeps_no_box(fake, monkeypatch) -> None:
     root, _ = fake
-    monkeypatch.setattr(terrain, "build_sheet_surface_model", lambda *a, **k: [])
+    seen = fake[1]
+    seen["z"] = np.float32(np.nan)  # 창고에 등고선 없음(빈 도엽)
     with pytest.raises(ValueError, match="사업 지형면을 만들지 못했습니다"):
         terrain.ensure_plan_terrain(root, [A, B], EPSG)
     assert terrain.received_box(root) is None