feat(지형표본): 344-1 지형 격자 표본기 짝 — TS heightAt(쌍선형 · 축 이진 탐색 · 네 칸 유효) · 서버 표본기 메모리 기억(경로 · mtime · 크기) · 격자 이진 꼴 · 거울 시험(서버와 차 0)

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01Se7ytqbohmBQnfYpwYFnYf
This commit is contained in:
2026-10-16 11:03:34 +09:00
co-authored by Claude Opus 5.5
parent 2879f9bea9
commit 5d37c851f3
3 changed files with 348 additions and 1 deletions
+60 -1
View File
@@ -8,10 +8,16 @@ DTM valid_mask를 footprint로 결합해 데이터가 없는 영역을 임의
B05 종횡단 전용이었으나 B04 배수유역 세부 설계도 같은 표고면을 써야 해서
common_util로 옮겼다 — 두 화면의 종단 Z가 갈라지면 세부유역 경계가 달라진다
(2026-08-01 구조 개편).
⚠ 파이썬·TS 짝 — `DtmGridSampler.sample_xy` ↔ `common_util_surface_sampler.ts` `heightAt`
(격자 = `dtm_grid_bytes` 꼴 · 쌍선형 · 네 칸 유효). 한쪽을 고치면 다른 쪽도 고치고 거울 시험
`resources/tester/test_344_1_surface_sampler_mirror.py` 를 돌림.
"""
import os
from collections.abc import Callable
from dataclasses import dataclass
from functools import lru_cache
from pathlib import Path
from typing import Literal, Protocol, get_args
@@ -89,6 +95,12 @@ class DtmGridSampler:
with np.load(Path(path), allow_pickle=False) as data:
return cls(data["x"], data["y"], data["z"], data["valid_mask"])
@classmethod
def open_cached(cls, path: Path | str) -> "DtmGridSampler":
"""`from_npz` 를 프로세스 메모리에 기억 — 열쇠 = 경로 · 수정시각 · 크기."""
stat = os.stat(path)
return _open_dtm(str(Path(path).resolve()), stat.st_mtime_ns, stat.st_size)
def sample_xy(self, xy: np.ndarray) -> tuple[np.ndarray, np.ndarray]:
xy = np.asarray(xy, dtype=np.float64)
if xy.ndim != 2 or xy.shape[1] != 2:
@@ -116,6 +128,53 @@ class DtmGridSampler:
return z, valid
# ponytail: 지표면 4벌(sheet 격자 한 벌 약 17 MB) — 서버 프로세스 하나 전제 · 늘면 maxsize 를 셈.
@lru_cache(maxsize=4)
def _open_dtm(path: str, _mtime_ns: int, _size: int) -> DtmGridSampler:
return DtmGridSampler.from_npz(path)
#: 브라우저로 보내는 격자 꼴(`dtm_grid_bytes`) 머리 — TS `parseDtmGrid` 가 같은 값을 봄.
GRID_MAGIC = b"AGRD"
GRID_VERSION = 1
def dtm_grid_bytes(path: Path | str) -> bytes:
"""DTM 격자를 브라우저 표본기(TS 짝)용 원시 이진으로 — 값을 하나도 바꾸지 않음.
꼴(little endian): `AGRD` · u32 판 · u32 nx · u32 ny · u32 z 바이트(4 | 8) · u32 0 ·
f64 x[nx] · f64 y[ny] · z[ny × nx](행 = y) · 유효 비트(행 우선 · 낮은 비트 먼저).
축 · 행렬은 `DtmGridSampler` 가 오름차순으로 맞춘 그대로 — 서버 표본과 같은 칸을 봄.
"""
grid = DtmGridSampler.open_cached(path)
with np.load(Path(path), allow_pickle=False) as data:
z_dtype = np.dtype("<f4") if data["z"].dtype == np.float32 else np.dtype("<f8")
head = np.array(
[GRID_VERSION, len(grid.x), len(grid.y), z_dtype.itemsize, 0], dtype="<u4"
).tobytes()
return b"".join(
[
GRID_MAGIC,
head,
grid.x.astype("<f8").tobytes(),
grid.y.astype("<f8").tobytes(),
grid.z.astype(z_dtype).tobytes(),
np.packbits(grid.valid_mask.reshape(-1), bitorder="little").tobytes(),
]
)
def dtm_grid_file(npz_path: Path | str) -> Path:
"""`dtm_grid_bytes` 저장본(npz 곁 `<이름>_grid.bin`) — 없거나 npz 보다 옛것이면 새로 씀."""
npz_path = Path(npz_path)
target = npz_path.with_name(f"{npz_path.stem}_grid.bin")
if not target.is_file() or target.stat().st_mtime_ns < npz_path.stat().st_mtime_ns:
temporary = target.with_name(f"{target.name}.{os.getpid()}.tmp")
temporary.write_bytes(dtm_grid_bytes(npz_path))
os.replace(temporary, target)
return target
@dataclass
class CallableSurfaceSampler:
"""테스트와 어댑터에 사용할 함수 기반 sampler."""
@@ -160,7 +219,7 @@ def build_surface_sampler(
dtm_path = dtm_smooth if smooth and dtm_smooth.exists() else dtm_original
if not dtm_path.exists():
raise FileNotFoundError(f"기준 DTM이 없습니다: {dtm_path.name}")
footprint = DtmGridSampler.from_npz(dtm_path)
footprint = DtmGridSampler.open_cached(dtm_path)
if method == "dtm":
return footprint