feat(641-2): B07 유역 음영 · D03 지형 — implicit 확정이어도 평활 DTM 읽음 · 지킴 시험
- 읽는 자리 한 줄씩(계산 코드 그대로) · dtm 확정은 전과 같음 - test_641_1_implicit_grid: 굳힌 격자 = RBF · 다시 셈 · 서버 · Node · B06 · 브라우저 같은 파일 · B07 · D03 전후 Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
This commit is contained in:
@@ -476,7 +476,7 @@ def _hillshade_for_box(
|
||||
|
||||
params = getattr(context, "surface_params", None) or {}
|
||||
source_filter = str(params.get("source_filter") or "csf")
|
||||
smooth = bool(params.get("smooth", True))
|
||||
smooth = bool(params.get("smooth", True)) or params.get("method") == "implicit" # 641-2
|
||||
models_dir = Path(context.project_root) / "B04_PreProcess" / "models"
|
||||
candidates = [models_dir / f"dtm_{source_filter}_smooth.npz"] if smooth else []
|
||||
candidates.append(models_dir / f"dtm_{source_filter}.npz")
|
||||
|
||||
@@ -105,7 +105,8 @@ def road_dem(models: Path, params: dict[str, Any], source: str) -> tuple[str, Pa
|
||||
return source, None
|
||||
npz = Path(models) / f"dtm_{picked.get('source_filter') or ''}.npz"
|
||||
smooth = npz.with_name(f"{npz.stem}_smooth.npz")
|
||||
npz = smooth if picked.get("smooth") and smooth.is_file() else npz
|
||||
implicit = picked.get("method") == "implicit" # 641-2 — implicit 확정이어도 평활 DTM
|
||||
npz = smooth if (picked.get("smooth") or implicit) and smooth.is_file() else npz
|
||||
return got, npz if npz.is_file() else None
|
||||
|
||||
|
||||
|
||||
@@ -0,0 +1,163 @@
|
||||
"""PLAN 641 — implicit 굳힌 격자(`implicit_{필터}_grid.npz`) · 화면 일치 · B07 · D03 그대로.
|
||||
|
||||
641-1: 굳힌 격자 축 · 유효 = 같은 필터 dtm · 유효 칸 값 = RBF 값 · 밖 = dtm 값 · 옛것이면 다시 셈 ·
|
||||
서버 표본기 · Node 측점(`source`) · B06 재계산 · 브라우저 격자 통로가 같은 파일.
|
||||
641-2: implicit 확정이어도 B07 음영 · D03 지형은 평활 DTM · dtm 확정은 전과 같음.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import asyncio
|
||||
import os
|
||||
from contextlib import asynccontextmanager
|
||||
from pathlib import Path
|
||||
|
||||
import numpy as np
|
||||
|
||||
from B05_Profile import B05_Profile_Router_Terrain as terrain_router
|
||||
from B06_Section import B06_Section_Server_Calc_Prebuild as prebuild
|
||||
from common_util.common_util_surface_sampler import (
|
||||
DtmGridSampler,
|
||||
build_surface_sampler,
|
||||
dtm_grid_bytes,
|
||||
implicit_grid_npz,
|
||||
implicit_rbf,
|
||||
)
|
||||
from D03_PlanAnalysis.D03_PlanAnalysis_TerrainSource import road_dem
|
||||
|
||||
IMPLICIT = {"source_filter": "classification", "method": "implicit", "smooth": False}
|
||||
|
||||
|
||||
def _surface(x, y):
|
||||
return 100.0 + 0.2 * x + 0.1 * y + 2.0 * np.sin(x / 7.0) * np.cos(y / 5.0)
|
||||
|
||||
|
||||
def _project(tmp_path: Path) -> Path:
|
||||
models = tmp_path / "B04_PreProcess" / "models"
|
||||
models.mkdir(parents=True)
|
||||
axis = np.arange(0.0, 40.0, dtype=np.float32)
|
||||
valid = np.zeros((40, 40), dtype=bool)
|
||||
valid[5:35, 3:30] = True
|
||||
for name, z in (("dtm_classification", 50.0), ("dtm_classification_smooth", 60.0)):
|
||||
np.savez(
|
||||
models / f"{name}.npz",
|
||||
x=axis,
|
||||
y=axis,
|
||||
z=np.full((40, 40), z, np.float32),
|
||||
valid_mask=valid,
|
||||
resolution=np.array([1.0], np.float32),
|
||||
)
|
||||
rng = np.random.default_rng(641)
|
||||
centers = rng.uniform(0.0, 39.0, (600, 2)).astype(np.float32)
|
||||
np.savez(
|
||||
models / "implicit_classification.npz",
|
||||
centers_xy=centers,
|
||||
center_z=_surface(centers[:, 0], centers[:, 1]).astype(np.float32),
|
||||
smoothing=np.array([0.1], np.float32),
|
||||
)
|
||||
return models
|
||||
|
||||
|
||||
def test_굳힌_격자_축_유효_dtm_같음_값_RBF(tmp_path):
|
||||
models = _project(tmp_path)
|
||||
path = implicit_grid_npz(models, "classification")
|
||||
assert path.name == "implicit_classification_grid.npz"
|
||||
with np.load(path) as grid, np.load(models / "dtm_classification.npz") as dtm:
|
||||
assert np.array_equal(grid["x"], dtm["x"]) and np.array_equal(grid["y"], dtm["y"])
|
||||
assert np.array_equal(grid["valid_mask"], dtm["valid_mask"])
|
||||
z, valid = grid["z"], grid["valid_mask"]
|
||||
assert not np.isnan(z).any()
|
||||
assert np.array_equal(z[~valid], dtm["z"][~valid]) # 밖 = dtm 값
|
||||
rows, cols = np.nonzero(valid)
|
||||
query = np.column_stack([grid["x"][cols], grid["y"][rows]]).astype(np.float64)
|
||||
rbf = implicit_rbf(models / "implicit_classification.npz")
|
||||
assert np.abs(rbf(query) - z[rows, cols]).max() < 1e-4 # 유효 칸 = RBF(float32)
|
||||
|
||||
|
||||
def test_옛것이면_다시_셈_같으면_그대로(tmp_path):
|
||||
models = _project(tmp_path)
|
||||
path = implicit_grid_npz(models, "classification")
|
||||
stamp = path.stat().st_mtime_ns
|
||||
assert implicit_grid_npz(models, "classification").stat().st_mtime_ns == stamp
|
||||
source = models / "implicit_classification.npz"
|
||||
os.utime(source, ns=(stamp + 10**9, stamp + 10**9))
|
||||
assert implicit_grid_npz(models, "classification").stat().st_mtime_ns > stamp
|
||||
|
||||
|
||||
def test_서버_Node_B06_브라우저_같은_파일(tmp_path, monkeypatch):
|
||||
models = _project(tmp_path)
|
||||
sampler = build_surface_sampler(models, "classification", "implicit", False)
|
||||
grid_npz = models / "implicit_classification_grid.npz"
|
||||
assert isinstance(sampler, DtmGridSampler)
|
||||
assert Path(sampler.source) == grid_npz.resolve() # Node 측점 = dtm_grid_file(source)
|
||||
|
||||
# 서버 표본 = 굳힌 격자 표본(같은 측점 같은 표고)
|
||||
points = np.array([[10.3, 12.7], [20.0, 20.0], [1.0, 1.0]])
|
||||
z, valid = sampler.sample_xy(points)
|
||||
z2, valid2 = DtmGridSampler.from_npz(grid_npz).sample_xy(points)
|
||||
assert np.array_equal(valid, valid2) and np.allclose(z[valid], z2[valid2], atol=0)
|
||||
assert valid.tolist() == [True, True, False]
|
||||
|
||||
# B06 서버 재계산 격자 = 같은 npz 의 `.bin`
|
||||
bin_path = prebuild.design_grid_file(tmp_path, IMPLICIT)
|
||||
assert bin_path is not None and bin_path.read_bytes() == dtm_grid_bytes(grid_npz)
|
||||
|
||||
# 브라우저 격자 통로 — implicit 모델 행 → 굳힌 격자
|
||||
row = ("implicit", "B04_PreProcess/models/implicit_classification.npz")
|
||||
|
||||
class _Cursor:
|
||||
async def execute(self, *_args):
|
||||
return None
|
||||
|
||||
async def fetchone(self):
|
||||
return row
|
||||
|
||||
class _Connection:
|
||||
@asynccontextmanager
|
||||
async def cursor(self):
|
||||
yield _Cursor()
|
||||
|
||||
async def stored(_connection, _project_id):
|
||||
return "x"
|
||||
|
||||
monkeypatch.setattr(terrain_router, "get_project_storage_relative_path", stored)
|
||||
monkeypatch.setattr(terrain_router, "resolve_stored_project_path", lambda _s: tmp_path)
|
||||
served = asyncio.run(terrain_router.dtm_grid_npz(_Connection(), "p", 9, False))
|
||||
assert served.resolve() == grid_npz.resolve()
|
||||
|
||||
info = terrain_router._design_grid(
|
||||
IMPLICIT, models, (9, "implicit", "B04_PreProcess/models/implicit_classification.npz")
|
||||
)
|
||||
assert info is not None and info["model_id"] == 9 and info["smooth"] is False
|
||||
assert (
|
||||
terrain_router._design_grid(IMPLICIT, models, (3, "dtm", "x/dtm_classification.npz"))
|
||||
is None
|
||||
)
|
||||
|
||||
|
||||
def test_D03_B07_implicit_확정이면_평활_DTM(tmp_path):
|
||||
models = _project(tmp_path)
|
||||
assert road_dem(models, IMPLICIT, "design")[1].name == "dtm_classification_smooth.npz"
|
||||
dtm = {**IMPLICIT, "method": "dtm"}
|
||||
assert road_dem(models, dtm, "design")[1].name == "dtm_classification.npz" # 전과 같음
|
||||
assert road_dem(models, {**dtm, "smooth": True}, "design")[1].name.endswith("_smooth.npz")
|
||||
|
||||
|
||||
def test_B07_음영_implicit_확정이면_평활_DTM(tmp_path, monkeypatch):
|
||||
from types import SimpleNamespace
|
||||
|
||||
from B07_DesignDetail import B07_DesignDetail_Router_Support_Basin as basin
|
||||
|
||||
_project(tmp_path)
|
||||
opened: list[str] = []
|
||||
real_load = np.load
|
||||
monkeypatch.setattr(
|
||||
np, "load", lambda path, **kw: opened.append(Path(path).name) or real_load(path, **kw)
|
||||
)
|
||||
box = (0.0, 0.0, 39.0, 39.0)
|
||||
for params, name in (
|
||||
(IMPLICIT, "dtm_classification_smooth.npz"),
|
||||
({**IMPLICIT, "method": "dtm"}, "dtm_classification.npz"), # 전과 같음
|
||||
):
|
||||
basin._hillshade_for_box(SimpleNamespace(surface_params=params, project_root=tmp_path), box)
|
||||
assert opened[-1] == name
|
||||
Reference in New Issue
Block a user