perf(D 예정노선): 지형 받기 창고 새 도엽 함께 셈 (PLAN 608-1)

- 받은 범위에 닿는 창고 없는 도엽을 스레드 넷까지 함께 셈 · 진행 = 도엽 n/m 매
- 용화 처음 진입 133.4 s → 67.7 s(차례 셈 대비)

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

D03_PlanAnalysis/D03_PlanAnalysis_Terrain.py

@@ -49,6 +49,7 @@ from B04_PreProcess.B04_PreProcess_Engine_SheetSurface import (
 )
 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 import common_util_terrain_store as terrain_store
 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 (
@@ -67,6 +68,8 @@ SITE_RADIUS_M = 1000.0
 # 위치 바꿈이 받은 범위에서 이만큼 넘게 멀면 묻고 새로 · 받을 DEM 칸 상한(요구서_463 3장 4)
 FAR_M = 5000.0
 MAX_CELLS = 5_000_000
+# ponytail: 새 도엽 함께 셀 수 — 장당 메모리 1 GB 안팎 · 넘으면 차례로
+WARM_WORKERS = 4
 STAGE = Path("B04_PreProcess")
 PLAN_MODELS = "models_plan"
 TERRAIN_META = "plan_terrain.json"
@@ -317,8 +320,10 @@ def ensure_plan_terrain(
     models_dir = models_dir_of(project_root)
     models_dir.mkdir(parents=True, exist_ok=True)
 
+    warmed = _warm_store(box, epsg, lambda pct, msg: report(65 + pct * 30 // 100, f"5/5 {msg}"))
+
     def store_report(percent: int, message: str) -> None:
-        report(65 + percent * 34 // 100, f"5/5 {message}")
+        report(95 + percent * 4 // 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)
@@ -343,10 +348,36 @@ def ensure_plan_terrain(
         "satellite": satellite,
         "box_m": list(box),
         "models": models,
-        "terrain": {"computed": grid["computed"], "failed": grid["failed"]},
+        "terrain": {"computed": warmed + grid["computed"], "failed": grid["failed"]},
     }
 
 
+def _warm_store(box: Box, epsg: int, report: Progress) -> list[str]:
+    """창고에 없는 도엽을 함께 셈(스레드 · 장당 약 50 s · 둘이면 약 절반) — 실패는 이어 붙일 때 다시
+    물어 `failed` 에 남음."""
+    from concurrent.futures import ThreadPoolExecutor, as_completed
+
+    cold = [
+        no
+        for no in terrain_store.sheets_for_box(box, epsg)
+        if terrain_store._stored_ok(no, epsg, PLAN_TERRAIN_GRID_M) is None
+    ]
+    if not cold:
+        return []
+    report(0, f"공용 지형 새 도엽 {len(cold)}매 계산 중 — 장당 1 분 안팎")
+    workers = min(WARM_WORKERS, len(cold))
+    with ThreadPoolExecutor(workers, thread_name_prefix="plan-terrain") as pool:
+        jobs = [
+            pool.submit(terrain_store.ensure_sheet_terrain, no, epsg, PLAN_TERRAIN_GRID_M)
+            for no in cold
+        ]
+        for done, job in enumerate(as_completed(jobs), 1):
+            if job.exception() is not None:
+                logger.warning("D 공용 지형 도엽 셈 실패: %s", job.exception())
+            report(done * 100 // len(cold), f"공용 지형 도엽 {done}/{len(cold)}매")
+    return cold
+
+
 def _write_store_model(
     project_root: Path, models_dir: Path, grid: dict[str, Any]
 ) -> list[dict[str, Any]]:

resources/tester/test_530_1_plan_terrain.py

@@ -24,7 +24,7 @@ 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": []}
+    seen = {"download": [], "merge": [], "satellite": [], "dem": [], "warm": []}
 
     def download_sheet(store_dir, sheet_no, version=2):
         seen["download"].append(sheet_no)
@@ -63,6 +63,7 @@ def fake(tmp_path, monkeypatch):
     monkeypatch.setattr(terrain, "parse_and_merge_sheets", merge)
     monkeypatch.setattr(terrain, "download_vworld_satellite_map", satellite)
     monkeypatch.setattr(terrain, "load_terrain", load_terrain)
+    monkeypatch.setattr(terrain, "_warm_store", lambda *a: seen["warm"].append(a[0]) or [])
     return tmp_path, seen
 
 
@@ -91,6 +92,7 @@ def test_one_set_and_extend_keeps_old(fake) -> None:
     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"]
+    assert seen["warm"][-1] == pytest.approx(box)  # 새 도엽 함께 셈 = 같은 상자
     assert terrain.received_box(root) == pytest.approx(tuple(box))
     # 작업 좌표계 .prj — B 좌표계 창구가 읽음
     assert resolve_project_crs(root) == f"EPSG:{EPSG}"