feat(641-1): implicit 굳힌 격자 — 확정 지표면 B05 · B06 화면 일치

- implicit_{필터}_grid.npz = 같은 필터 dtm 축 · 유효 칸 · 유효 칸만 RBF · 밖 = dtm 값 · implicit · dtm 보다 옛것이면 다시 셈(파생 저장본)
- 서버 표본기 implicit = 굳힌 격자 DtmGridSampler(매번 RBF 다시 짓기 없앰) · Node 측점이 source 로 같은 격자
- B05 브라우저 격자 통로 · B06 서버 재계산 격자 = implicit 확정이면 같은 굳힌 격자(새 짝 없음 · 기존 heightAt 짝)
- 짓기 끝(build_implicit)에서 한 번 굳힘 · 못 지으면 처음 읽을 때

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

B04_PreProcess/B04_PreProcess_Engine_ModelBuild.py

@@ -4,6 +4,7 @@ TerrainContext에서 파생 격자·샘플을 받아 각 표현의 모델(npz)
 (GLB/PLY)를 저장하고 메타데이터를 반환한다.
 """
 
+from contextlib import suppress
 from pathlib import Path
 from typing import Any
 
@@ -25,6 +26,7 @@ from B04_PreProcess.B04_PreProcess_Engine_ModelContext import (
     write_binary_ply,
     write_glb,
 )
+from common_util.common_util_surface_sampler import implicit_grid_npz
 
 
 def build_tin(
@@ -209,6 +211,8 @@ def build_implicit(
         smoothing=np.array([float(context.config["implicit_smoothing"])], np.float32),
     )
     write_glb(preview_path, vertices, faces, context.bounds)
+    with suppress(OSError, ValueError):  # 641-1 굳힌 격자 — 못 지으면 처음 읽을 때 지음
+        implicit_grid_npz(output_dir, stem.removeprefix("implicit_"))
     progress(100)
     return with_footprint(
         context,

B05_Profile/B05_Profile_Engine_Sections_Node.py

@@ -7,7 +7,7 @@
 (화면 묶음 stations · ground 단계와 같은 함수). 파이썬은 머리(schema_version · source ·
 coordinate_reference · cad_exchange · options)만 붙임.
 
-DTM 지표면만(격자 이진 = `dtm_grid_file`) — DTM 아닌 지표면(tin · nurbs · implicit · meshfree) ·
+DTM · implicit(굳힌 격자 641-1) 지표면만(격자 이진 = `dtm_grid_file`) — 그 밖(tin · nurbs · meshfree) ·
 Node 못 돌림 = None → 부르는 쪽이 파이썬 `generate_sections`(짝 · 로그 남김).
 측점 열쇠(`key`) = 노선 저장 이월 열쇠(`B05_Profile_Route_Save._carry_cross`).
 """

B05_Profile/B05_Profile_Router_Terrain.py

@@ -40,7 +40,11 @@ from common_util.common_util_surface_confirmation import (
     design_terrain,
     get_surface_confirmation_params,
 )
-from common_util.common_util_surface_sampler import build_surface_sampler, dtm_grid_file
+from common_util.common_util_surface_sampler import (
+    build_surface_sampler,
+    dtm_grid_file,
+    implicit_grid_npz,
+)
 from config.config_db import get_db_pool
 
 logger = logging.getLogger(__name__)
@@ -161,15 +165,17 @@ async def choosable_terrains(
 async def _design_model(
     connection: aiomysql.Connection, project_id: UUID, params: dict
 ) -> tuple[int, str, str] | None:
-    """설계 지형 DTM 모델 (id, 표현, 파일 경로) — 서버 표본기처럼 stage 1 params 의 `dtm_{filter}.npz`
-    (확정 행 먼저). 확정 행이 params 와 어긋나도 서버와 같은 바닥을 받게(578-1) · 없으면 None."""
-    name = f"dtm_{params.get('source_filter') or ''}.npz"
+    """설계 지형 격자 모델 (id, 표현, 파일 경로) — 서버 표본기처럼 stage 1 params 의
+    `{dtm|implicit}_{filter}.npz`(확정 행 먼저 · implicit = 굳힌 격자 641-1). 확정 행이 params 와
+    어긋나도 서버와 같은 바닥을 받게(578-1) · 없으면 None."""
+    method = "implicit" if params.get("method") == "implicit" else "dtm"
+    name = f"{method}_{params.get('source_filter') or ''}.npz"
     async with connection.cursor() as cursor:
         await cursor.execute(
             "SELECT id, model_type, model_file_path FROM surface_models"
-            " WHERE project_id = %s AND model_type = 'dtm' AND status IN ('CONFIRMED', 'COMPLETE')"
+            " WHERE project_id = %s AND model_type = %s AND status IN ('CONFIRMED', 'COMPLETE')"
             " ORDER BY status = 'CONFIRMED' DESC, created_at DESC",
-            (str(project_id),),
+            (str(project_id), method),
         )
         rows = await cursor.fetchall()
     row = next((r for r in rows if Path(str(r[2] or "")).name == name), None)
@@ -181,15 +187,16 @@ def _design_grid(
 ) -> dict | None:
     """설계 지형 격자 받을 자리(PLAN 344-1) — 서버 표본기(`build_surface_sampler`)와 같은 파일만.
 
-    DTM 아닌 표현(TIN 등)은 None → 화면은 서버 표본 길 그대로. `signature` = manifest 구성 ·
-    평활 서명 — 화면이 주소에 붙여 지형이 바뀐 뒤 옛 격자를 먼저 내지 않음.
+    DTM · implicit(굳힌 격자) 밖 표현(TIN 등)은 None → 화면은 서버 표본 길 그대로. `signature` =
+    manifest 구성 · 평활 서명 — 화면이 주소에 붙여 지형이 바뀐 뒤 옛 격자를 먼저 내지 않음.
     """
     source_filter = str(params.get("source_filter") or "")
-    if not confirmed or params.get("method") != "dtm" or confirmed[1] != "dtm":
+    method = params.get("method")
+    if not confirmed or method not in ("dtm", "implicit") or confirmed[1] != method:
         return None
-    if Path(confirmed[2]).name != f"dtm_{source_filter}.npz":
+    if Path(confirmed[2]).name != f"{method}_{source_filter}.npz":
         return None
-    smooth = bool(params.get("smooth"))
+    smooth = bool(params.get("smooth")) and method == "dtm"
     try:
         manifest = json.loads((models_dir / "manifest.json").read_text(encoding="utf-8"))
     except (OSError, ValueError):
@@ -206,7 +213,8 @@ def _design_grid(
 
 
 async def dtm_grid_npz(connection, project_id: UUID, model_id: int, smooth: bool) -> Path:
-    """그 프로젝트 DTM 모델의 npz(평활 고르기 = 서버 표본기와 같음) — 없으면 LookupError(까닭).
+    """그 프로젝트 DTM 모델의 npz(평활 고르기 = 서버 표본기와 같음) · implicit 모델 = 굳힌 격자
+    (641-1 · 표본기와 같은 파일) — 없으면 LookupError(까닭).
 
     경로 = 프로젝트 storage 안 DB 상대 경로만(밖으로 나가면 LookupError · 계산 창구 `/calc` 도 씀).
     """
@@ -218,12 +226,19 @@ async def dtm_grid_npz(connection, project_id: UUID, model_id: int, smooth: bool
             (model_id, str(project_id)),
         )
         row = await cursor.fetchone()
-    if not stored or not row or row[0] != "dtm" or not row[1]:
+    if not stored or not row or row[0] not in ("dtm", "implicit") or not row[1]:
         raise LookupError("DTM 격자가 없는 지표면 모델입니다.")
     root = Path(resolve_stored_project_path(stored))
     model_path = (root / str(row[1])).resolve()
     if not model_path.is_relative_to(root.resolve()):
         raise LookupError("DTM 격자가 없는 지표면 모델입니다.")
+    if row[0] == "implicit":
+        try:
+            return await asyncio.to_thread(
+                implicit_grid_npz, model_path.parent, model_path.stem.removeprefix("implicit_")
+            )
+        except (OSError, ValueError) as exc:
+            raise LookupError("implicit 격자를 굳히지 못했습니다.") from exc
     smooth_path = model_path.with_name(f"{model_path.stem}_smooth.npz")
     npz_path = smooth_path if smooth and smooth_path.is_file() else model_path
     if not npz_path.is_file():

B06_Section/B06_Section_Server_Calc_Prebuild.py

@@ -255,12 +255,13 @@ def _enforce_stored_designs(
 
 
 def design_grid_file(project_root: Path, params: dict[str, Any] | None) -> Path | None:
-    """설계 지형이 DTM 이면 격자 이진(`dtm_grid_file` · 표본기 `build_surface_sampler` 와
-    같은 npz) · DTM 아닌 지표면(tin · nurbs · …) · 못 엶 = None(파이썬 길)."""
-    from common_util.common_util_surface_sampler import dtm_grid_file
+    """설계 지형이 DTM · implicit 면 격자 이진(`dtm_grid_file` · 표본기 `build_surface_sampler` 와
+    같은 npz · implicit = 굳힌 격자 641-1) · 그 밖 지표면(tin · nurbs · …) · 못 엶 = None
+    (파이썬 길)."""
+    from common_util.common_util_surface_sampler import dtm_grid_file, implicit_grid_npz
 
     params = params or {}
-    if params.get("method") != "dtm" or not params.get("source_filter"):
+    if params.get("method") not in ("dtm", "implicit") or not params.get("source_filter"):
         return None
     models = project_root / "B04_PreProcess" / "models"
     smooth = models / f"dtm_{params['source_filter']}_smooth.npz"
@@ -270,6 +271,8 @@ def design_grid_file(project_root: Path, params: dict[str, Any] | None) -> Path
         else smooth.with_name(f"dtm_{params['source_filter']}.npz")
     )
     try:
+        if params["method"] == "implicit":
+            npz = implicit_grid_npz(models, str(params["source_filter"]))
         return dtm_grid_file(npz) if npz.is_file() else None
     except (OSError, ValueError, KeyError) as exc:
         logger.warning("서버 재계산: DTM 격자 이진을 못 만듦 — 파이썬 길 · %s", exc)

common_util/common_util_surface_sampler.py

@@ -177,6 +177,55 @@ def dtm_grid_file(npz_path: Path | str) -> Path:
     return target
 
 
+def implicit_rbf(path: Path | str) -> Callable[[np.ndarray], np.ndarray]:
+    """`implicit_{필터}.npz`(중심점 · 높이 · smoothing) → RBF 표고 함수((N, 2) → (N,))."""
+    from scipy.interpolate import RBFInterpolator
+
+    with np.load(Path(path), allow_pickle=False) as data:
+        centers = np.asarray(data["centers_xy"], dtype=np.float64)
+        center_z = np.asarray(data["center_z"], dtype=np.float64)
+        smoothing = float(data["smoothing"][0]) if "smoothing" in data else 0.0
+    return RBFInterpolator(
+        centers,
+        center_z,
+        neighbors=min(64, len(centers)),
+        smoothing=smoothing,
+        kernel="thin_plate_spline",
+    )
+
+
+def implicit_grid_npz(models_dir: Path | str, source_filter: str) -> Path:
+    """implicit 를 1 m 격자로 굳힌 파생 저장본 `implicit_{필터}_grid.npz`(641-1 · 요구 558-3).
+
+    축 · 유효 칸 = 같은 필터 `dtm_{필터}.npz` 그대로 · 유효 칸만 RBF · 밖은 dtm 값(NaN 없음).
+    없거나 implicit · dtm 보다 옛것이면 다시 셈 — 서버 표본 · 브라우저 · Node 측점 · B06 이 같은 값.
+    """
+    check_surface_keys(source_filter, "implicit")
+    models_dir = Path(models_dir)
+    source = models_dir / f"implicit_{source_filter}.npz"
+    dtm = models_dir / f"dtm_{source_filter}.npz"
+    target = models_dir / f"implicit_{source_filter}_grid.npz"
+    newest = max(source.stat().st_mtime_ns, dtm.stat().st_mtime_ns)
+    if target.is_file() and target.stat().st_mtime_ns >= newest:
+        return target
+    with np.load(dtm, allow_pickle=False) as data:
+        x, y = data["x"], data["y"]
+        z = np.array(data["z"])
+        valid = np.asarray(data["valid_mask"], dtype=bool)
+        resolution = data["resolution"] if "resolution" in data else np.array([1.0], np.float32)
+    rbf = implicit_rbf(source)
+    rows, cols = np.nonzero(valid)
+    query = np.column_stack([x[cols], y[rows]]).astype(np.float64)
+    for start in range(0, len(query), 50_000):
+        end = start + 50_000
+        z[rows[start:end], cols[start:end]] = rbf(query[start:end])
+    temporary = target.with_name(f"{target.stem}.{os.getpid()}.tmp")
+    with open(temporary, "wb") as handle:
+        np.savez(handle, x=x, y=y, z=z, valid_mask=valid, resolution=resolution)
+    os.replace(temporary, target)
+    return target
+
+
 @dataclass
 class CallableSurfaceSampler:
     """테스트와 어댑터에 사용할 함수 기반 sampler."""
@@ -255,21 +304,8 @@ def build_surface_sampler(
         return InterpolatedSurfaceSampler(lambda xy: spline.ev(xy[:, 1], xy[:, 0]), footprint)
 
     if method == "implicit":
-        from scipy.interpolate import RBFInterpolator
-
-        path = models_dir / f"implicit_{source_filter}.npz"
-        with np.load(path, allow_pickle=False) as data:
-            centers = np.asarray(data["centers_xy"], dtype=np.float64)
-            center_z = np.asarray(data["center_z"], dtype=np.float64)
-            smoothing = float(data["smoothing"][0]) if "smoothing" in data else 0.0
-        interpolator = RBFInterpolator(
-            centers,
-            center_z,
-            neighbors=min(64, len(centers)),
-            smoothing=smoothing,
-            kernel="thin_plate_spline",
-        )
-        return InterpolatedSurfaceSampler(lambda xy: interpolator(xy), footprint)
+        # 굳힌 격자(641-1) — 브라우저 · Node 측점 · B06 이 같은 파일을 읽음(`source` 가 붙음)
+        return DtmGridSampler.open_cached(implicit_grid_npz(models_dir, source_filter))
 
     if method == "meshfree":
         from scipy.interpolate import LinearNDInterpolator