Merge remote-tracking branch 'origin/dev' into sub_laptop_5

common_util/common_util_surface_sampler.py

@@ -178,6 +178,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."""
@@ -256,21 +305,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