Aislo/resources/tester/test_344_1_surface_sampler_mirror.py

"""지형 격자 표본기 거울 — 파이썬 `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, "<u4", 5, 4)
    assert (version, nx, ny, z_bytes) == (1, 60, 40, 4)
    grid = sampler_py.DtmGridSampler.from_npz(synthetic_npz)
    x = np.frombuffer(blob, "<f8", nx, 24)
    assert np.array_equal(x, grid.x) and np.all(np.diff(x) > 0)
    z = np.frombuffer(blob, "<f4", nx * ny, 24 + 8 * (nx + ny)).reshape(ny, nx)
    assert np.array_equal(z, grid.z.astype(np.float32), equal_nan=True)
    bits = np.frombuffer(blob, np.uint8, offset=24 + 8 * (nx + ny) + 4 * nx * ny)
    mask = np.unpackbits(bits, bitorder="little")[: nx * ny].reshape(ny, nx).astype(bool)
    assert np.array_equal(mask, grid.valid_mask)


def test_표본기_기억은_파일이_바뀌면_새로_읽는다(synthetic_npz: Path) -> 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 "")