diff --git a/common_util/common_util_surface_sampler.py b/common_util/common_util_surface_sampler.py index d07c9ca63..d8b9c44a2 100644 --- a/common_util/common_util_surface_sampler.py +++ b/common_util/common_util_surface_sampler.py @@ -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(" 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 diff --git a/common_util/common_util_surface_sampler.ts b/common_util/common_util_surface_sampler.ts new file mode 100644 index 000000000..f064f3815 --- /dev/null +++ b/common_util/common_util_surface_sampler.ts @@ -0,0 +1,82 @@ +/* ============================================================================= + * common_util_surface_sampler.ts + * 확정 지표면(DTM 격자) 표고를 브라우저에서 바로 읽는다 — 노선을 고치는 동안 서버 왕복 없이. + * + * ⚠ 파이썬·TS 짝 — `common_util/common_util_surface_sampler.py` 의 `DtmGridSampler.sample_xy` + * (scipy `RegularGridInterpolator` linear + 네 칸 유효) · 격자 꼴 = `dtm_grid_bytes`. + * 한쪽을 고치면 다른 쪽도 고치고 거울 시험 `resources/tester/test_344_1_surface_sampler_mirror.py` + * 를 돌림. + * + * 칸 폭이 고르지 않음(축이 float32 로 저장돼 0.97~1.09 m) — 「(x − x0) / 1 m」 셈 금지 · 축 배열을 + * 이진 탐색(`searchsorted(side="right") − 1`)해 서버와 같은 칸을 봄. + * ========================================================================== */ + +/** 오름차순 축 · 행렬 · 유효 비트 — 서버 `dtm_grid_bytes` 를 그대로 푼 것. */ +export interface DtmGrid { + x: Float64Array; + y: Float64Array; + /** 행 = y · 열 = x (`z[iy * nx + ix]`). */ + z: Float32Array | Float64Array; + /** 칸마다 한 비트(행 우선 · 낮은 비트 먼저). */ + validBits: Uint8Array; +} + +const GRID_MAGIC = "AGRD"; +const GRID_VERSION = 1; +const HEAD_BYTES = 24; + +/** 서버 `dtm_grid_bytes` 이진을 푼다 — 꼴이 다르면 throw. */ +export function parseDtmGrid(buffer: ArrayBuffer): DtmGrid { + const magic = String.fromCharCode(...new Uint8Array(buffer, 0, 4)); + const [version, nx, ny, zBytes] = new Uint32Array(buffer, 4, 4); + if (magic !== GRID_MAGIC || version !== GRID_VERSION || (zBytes !== 4 && zBytes !== 8)) { + throw new Error("지형 격자 꼴이 다름"); + } + const x = new Float64Array(buffer, HEAD_BYTES, nx); + const y = new Float64Array(buffer, HEAD_BYTES + nx * 8, ny); + const zStart = HEAD_BYTES + (nx + ny) * 8; + const z = + zBytes === 4 + ? new Float32Array(buffer, zStart, nx * ny) + : new Float64Array(buffer, zStart, nx * ny); + const validBits = new Uint8Array(buffer, zStart + nx * ny * zBytes, Math.ceil((nx * ny) / 8)); + return { x, y, z, validBits }; +} + +/** `np.searchsorted(axis, value, side="right") - 1` — value 이하인 마지막 칸 번호 · 앞이면 −1. */ +function cellIndex(axis: Float64Array, value: number): number { + if (!(value >= axis[0])) return -1; // NaN 도 여기(밖) + let low = 0; + let high = axis.length; + while (high - low > 1) { + const mid = (low + high) >> 1; + if (axis[mid] <= value) low = mid; + else high = mid; + } + return low; +} + +function isValid(grid: DtmGrid, index: number): boolean { + return ((grid.validBits[index >> 3] >> (index & 7)) & 1) === 1; +} + +/** (x, y) 표고(m) — 네 칸이 모두 유효할 때만 값 · 밖 · 빈칸 · 마지막 선 위는 NaN(임의 표고로 안 메움). + * 식 · 셈 차례 = scipy `evaluate_linear_2d`(같은 IEEE 연산 → 같은 값). */ +export function heightAt(grid: DtmGrid, x: number, y: number): number { + const nx = grid.x.length; + const ix = cellIndex(grid.x, x); + const iy = cellIndex(grid.y, y); + if (ix < 0 || iy < 0 || ix >= nx - 1 || iy >= grid.y.length - 1) return NaN; + const i00 = iy * nx + ix; + const i10 = i00 + nx; + if (!isValid(grid, i00) || !isValid(grid, i00 + 1)) return NaN; + if (!isValid(grid, i10) || !isValid(grid, i10 + 1)) return NaN; + const ty = (y - grid.y[iy]) / (grid.y[iy + 1] - grid.y[iy]); + const tx = (x - grid.x[ix]) / (grid.x[ix + 1] - grid.x[ix]); + let value = 0.0; + value = value + grid.z[i00] * (1 - ty) * (1 - tx); + value = value + grid.z[i00 + 1] * (1 - ty) * tx; + value = value + grid.z[i10] * ty * (1 - tx); + value = value + grid.z[i10 + 1] * ty * tx; + return Number.isFinite(value) ? value : NaN; +} diff --git a/resources/tester/test_344_1_surface_sampler_mirror.py b/resources/tester/test_344_1_surface_sampler_mirror.py new file mode 100644 index 000000000..d6cefe699 --- /dev/null +++ b/resources/tester/test_344_1_surface_sampler_mirror.py @@ -0,0 +1,206 @@ +"""지형 격자 표본기 거울 — 파이썬 `DtmGridSampler.sample_xy` ↔ TS `common_util_surface_sampler.ts`. + +PLAN 344-1 — 노선 편집이 브라우저에서 지반고를 읽어도 서버 값과 같아야 함. +가장자리 · 빈칸 · 마지막 선 · 칸 폭 고르지 않음 · 내림차순 축 · 무작위 2만 점. +""" + +from __future__ import annotations + +import json +import os +import re +import subprocess +import sys +import time +from pathlib import Path + +import numpy as np +import pytest + +ROOT = Path(__file__).resolve().parents[2] +if str(ROOT) not in sys.path: + sys.path.insert(0, str(ROOT)) + +from common_util import common_util_surface_sampler as sampler_py # noqa: E402 + +TS_FILE = ROOT / "common_util" / "common_util_surface_sampler.ts" +TSC = ROOT / "config" / "node_modules" / "typescript" / "bin" / "tsc" +#: 검증 프로젝트(검증_sub8) 설계 지형 — 없으면 그 시험만 건너뜀. +REAL_GRID = ( + ROOT + / "storage/1/3/1bb0f876-983f-4ca5-9cc7-f59ee1f77e0a/B04_PreProcess/models" + / "dtm_classification_smooth.npz" +) +REAL_LONGITUDINAL = REAL_GRID.parents[2] / "B06_Section" / "longitudinal" / "longitudinal.json" + +_RUNNER = """ +import { readFileSync, writeFileSync } from "node:fs"; +import { parseDtmGrid, heightAt } from "./common_util_surface_sampler.js"; +const bytes = readFileSync(process.argv[2]); +const grid = parseDtmGrid(bytes.buffer.slice(bytes.byteOffset, bytes.byteOffset + bytes.length)); +const points = JSON.parse(readFileSync(process.argv[3], "utf8")); +const z = points.map(([x, y]) => { + const value = heightAt(grid, x, y); + return Number.isNaN(value) ? null : value; +}); +writeFileSync(process.argv[4], JSON.stringify(z)); +""" + + +def _compile(tmp_path: Path) -> Path: + out = tmp_path / "js" + subprocess.run( # noqa: S603 — 고정 실행 파일 + [ + "node", + str(TSC), + str(TS_FILE), + "--outDir", + str(out), + "--module", + "esnext", + "--target", + "es2022", + "--moduleResolution", + "bundler", + "--ignoreConfig", + ], + cwd=str(ROOT), + check=True, + capture_output=True, + ) + for emitted in out.glob("*.js"): + emitted.write_text( + re.sub(r'(from\s+"\./[^"]+)"', r'\1.js"', emitted.read_text(encoding="utf-8")), + encoding="utf-8", + ) + (out / "runner.mjs").write_text(_RUNNER, encoding="utf-8") + return out + + +def _ts_heights(tmp_path: Path, npz: Path, points: np.ndarray) -> list[float | None]: + out = _compile(tmp_path) + grid = tmp_path / "grid.bin" + grid.write_bytes(sampler_py.dtm_grid_bytes(npz)) + payload = tmp_path / "points.json" + result = tmp_path / "z.json" + payload.write_text(json.dumps(points.tolist()), encoding="utf-8") + subprocess.run( # noqa: S603 + ["node", str(out / "runner.mjs"), str(grid), str(payload), str(result)], + cwd=str(ROOT), + check=True, + capture_output=True, + ) + return json.loads(result.read_text(encoding="utf-8")) + + +def _compare(npz: Path, points: np.ndarray, actual: list[float | None]) -> float: + """유효 어긋남 0 을 묶고 최대 표고 차를 돌려줌.""" + expected, valid = sampler_py.DtmGridSampler.from_npz(npz).sample_xy(points) + mismatched = [i for i, ok in enumerate(valid) if bool(ok) != (actual[i] is not None)] + assert not mismatched, ( + f"유효 어긋남 {len(mismatched)}점 — 처음 {points[mismatched[:5]].tolist()}" + ) + diffs = [abs(actual[i] - float(expected[i])) for i in np.flatnonzero(valid)] + return max(diffs) if diffs else 0.0 + + +def _edge_points(x: np.ndarray, y: np.ndarray) -> np.ndarray: + """격자 꼭짓점 · 첫 · 마지막 선 · 그 바로 안팎 · 칸 가운데.""" + xs = np.r_[x, (x[:-1] + x[1:]) / 2, x[0] - 1e-9, x[-1] - 1e-9, x[-1] + 1e-9] + ys = np.r_[y[[0, 1, -2, -1]], y[-1] - 1e-9, y[0] + 1e-9, (y[0] + y[1]) / 2] + edge = np.array([[a, b] for a in xs for b in ys]) + edge2 = np.array([[a, b] for a in x[[0, 1, -2, -1]] for b in y]) + return np.vstack([edge, edge2]) + + +@pytest.fixture +def synthetic_npz(tmp_path: Path) -> Path: + """칸 폭 고르지 않음(float32) · y 내림차순 · 빈칸 · 유효인데 z NaN 한 칸.""" + rng = np.random.default_rng(344) + x = np.cumsum(rng.uniform(0.95, 1.1, 60)).astype(np.float32) + np.float32(208300.0) + y = (np.cumsum(rng.uniform(0.95, 1.1, 40)) + 367400.0).astype(np.float32)[::-1] + z = (800 + rng.normal(0, 5, (40, 60))).astype(np.float32) + mask = np.ones((40, 60), dtype=bool) + mask[10:14, 20:25] = False + mask[0, :] = False + mask[:, -1] = False + z[30, 30] = np.nan + path = tmp_path / "dtm_test.npz" + np.savez(path, x=x, y=y, z=z, valid_mask=mask) + return path + + +@pytest.mark.skipif(not TSC.is_file(), reason="프론트엔드 의존성 미설치") +def test_가짜_격자_가장자리_빈칸_무작위가_서버와_같다(tmp_path: Path, synthetic_npz: Path) -> None: + data = np.load(synthetic_npz) + x, y = np.sort(data["x"].astype(np.float64)), np.sort(data["y"].astype(np.float64)) + rng = np.random.default_rng(1) + random = np.column_stack( + [rng.uniform(x[0] - 5, x[-1] + 5, 20000), rng.uniform(y[0] - 5, y[-1] + 5, 20000)] + ) + points = np.vstack([_edge_points(x, y), random]) + worst = _compare(synthetic_npz, points, _ts_heights(tmp_path, synthetic_npz, points)) + assert worst <= 1e-9, f"표고 차 {worst}" + + +@pytest.mark.skipif(not TSC.is_file(), reason="프론트엔드 의존성 미설치") +@pytest.mark.skipif(not REAL_GRID.is_file(), reason="검증_sub8 지형 없음") +def test_검증_프로젝트_격자와_저장된_종단이_서버와_같다(tmp_path: Path) -> None: + grid = sampler_py.DtmGridSampler.from_npz(REAL_GRID) + rng = np.random.default_rng(2) + random = np.column_stack( + [ + rng.uniform(grid.x[0] - 5, grid.x[-1] + 5, 20000), + rng.uniform(grid.y[0] - 5, grid.y[-1] + 5, 20000), + ] + ) + stored = [] + if REAL_LONGITUDINAL.is_file(): + samples = json.loads(REAL_LONGITUDINAL.read_text(encoding="utf-8"))["samples"] + stored = [[s["x"], s["y"]] for s in samples] + points = np.vstack([_edge_points(grid.x, grid.y), random, np.asarray(stored).reshape(-1, 2)]) + actual = _ts_heights(tmp_path, REAL_GRID, points) + worst = _compare(REAL_GRID, points, actual) + assert worst <= 1e-9, f"표고 차 {worst}" + + +def test_격자_꼴이_축_칸_비트를_그대로_싣는다(synthetic_npz: Path) -> None: + blob = sampler_py.dtm_grid_bytes(synthetic_npz) + assert blob[:4] == b"AGRD" + version, nx, ny, z_bytes, _ = np.frombuffer(blob, " 0) + z = np.frombuffer(blob, " None: + first = sampler_py.DtmGridSampler.open_cached(synthetic_npz) + assert sampler_py.DtmGridSampler.open_cached(synthetic_npz) is first + data = dict(np.load(synthetic_npz)) + data["z"] = data["z"] + np.float32(1.0) + np.savez(synthetic_npz, **data) + later = time.time() + 5 + os.utime(synthetic_npz, (later, later)) + assert sampler_py.DtmGridSampler.open_cached(synthetic_npz) is not first + + +def test_격자_저장본은_npz_보다_옛것이면_다시_쓴다(synthetic_npz: Path) -> None: + target = sampler_py.dtm_grid_file(synthetic_npz) + assert target.name == "dtm_test_grid.bin" + assert target.read_bytes() == sampler_py.dtm_grid_bytes(synthetic_npz) + stamp = target.stat().st_mtime_ns + assert sampler_py.dtm_grid_file(synthetic_npz).stat().st_mtime_ns == stamp # 그대로 씀 + later = time.time() + 10 + os.utime(synthetic_npz, (later, later)) + assert sampler_py.dtm_grid_file(synthetic_npz).stat().st_mtime_ns != stamp + + +def test_짝_머리글이_서로를_가리킨다() -> None: + assert "common_util_surface_sampler.py" in TS_FILE.read_text(encoding="utf-8") + assert "common_util_surface_sampler.ts" in (sampler_py.__doc__ or "")