fix(754-1): B04 등고선 · 격자 외곽 = 지금 지표면 외곽 — 어느 표현을 지어도 낡은 dtm 같이 다시 지음 · 등고선 캐시 판단에 dtm · 굳힌 격자 때
- 지표면 짓기 차례에 dtm 을 늘 맨 앞 — manifest 항목 없음 · 지문 다름이면 같이 다시 지음(화면 [서피스 생성] = 표현 하나만 짓던 길) - 등고선 캐시 낡음 = 모델 · 같은 필터 dtm · implicit 굳힌 격자 중 하나라도 더 새것(짓기 차례 · 등고선 API 한 함수) - 등고선 추출이 모델 · dtm 파일을 닫음 — 실패 때 열린 채 남아 Windows 에서 다시 짓기(교체)가 막히던 흠 - ptd 처럼 dtm 없던 필터도 dtm · 굳힌 격자가 생겨 ptd · implicit 확정 길 B05 · B06 열림 - 새 시험 test_754_1_contour_outline.py Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
This commit is contained in:
Claude Opus 5.5
@@ -122,6 +122,23 @@ def extract_contours_from_grid(
|
||||
return contours_geojson_list
|
||||
|
||||
|
||||
def contour_cache_is_stale(contour_path: Path, model_path: Path) -> bool:
|
||||
"""등고선 캐시가 없거나 그 면을 낳은 파일보다 옛것이면 True(754-1).
|
||||
|
||||
낳은 파일 = 모델 npz · 같은 필터 `dtm_{필터}.npz`(외곽 자르기) · implicit 굳힌 격자.
|
||||
"""
|
||||
stem = Path(model_path).stem.removesuffix("_smooth")
|
||||
folder = Path(model_path).parent
|
||||
sources = [Path(model_path), folder / f"dtm_{stem.split('_', 1)[-1]}.npz"]
|
||||
if stem.startswith("implicit_"):
|
||||
sources.append(folder / f"{stem}_grid.npz")
|
||||
try:
|
||||
built = Path(contour_path).stat().st_mtime_ns
|
||||
except OSError:
|
||||
return True
|
||||
return any(path.is_file() and path.stat().st_mtime_ns > built for path in sources)
|
||||
|
||||
|
||||
def _load_footprint_mask(
|
||||
model_npz_path: Path, x_coords: np.ndarray, y_coords: np.ndarray
|
||||
) -> np.ndarray | None:
|
||||
@@ -137,10 +154,10 @@ def _load_footprint_mask(
|
||||
if not dtm_path.exists():
|
||||
return None
|
||||
try:
|
||||
d = np.load(dtm_path)
|
||||
dtm_x = np.asarray(d["x"]).ravel()
|
||||
dtm_y = np.asarray(d["y"]).ravel()
|
||||
dtm_mask = np.asarray(d["valid_mask"], dtype=bool)
|
||||
with np.load(dtm_path) as d:
|
||||
dtm_x = np.asarray(d["x"]).ravel()
|
||||
dtm_y = np.asarray(d["y"]).ravel()
|
||||
dtm_mask = np.asarray(d["valid_mask"], dtype=bool)
|
||||
except Exception:
|
||||
return None
|
||||
if len(dtm_x) < 2 or len(dtm_y) < 2:
|
||||
@@ -360,7 +377,21 @@ def extract_contours(
|
||||
model_npz_path = Path(model_npz_path)
|
||||
if not model_npz_path.exists():
|
||||
raise FileNotFoundError(f"모델 파일이 존재하지 않습니다: {model_npz_path}")
|
||||
data = np.load(model_npz_path)
|
||||
# 실패해도 모델 파일을 닫음 — 열린 채 남으면 Windows 에서 다시 짓기(교체)가 막힘(754-1)
|
||||
with np.load(model_npz_path) as data:
|
||||
return _extract_loaded(
|
||||
data, model_npz_path, representation, interval, target_grid_m, scene_center
|
||||
)
|
||||
|
||||
|
||||
def _extract_loaded(
|
||||
data: Any,
|
||||
model_npz_path: Path,
|
||||
representation: str,
|
||||
interval: float,
|
||||
target_grid_m: float,
|
||||
scene_center: tuple[float, float, float] | None,
|
||||
) -> list[dict[str, Any]]:
|
||||
|
||||
if representation == "regular_grid":
|
||||
x_coords, y_coords, z_grid, valid_mask = (
|
||||
|
||||
@@ -16,6 +16,7 @@ import numpy as np
|
||||
|
||||
from B04_PreProcess.B04_PreProcess_Engine_Contour import (
|
||||
CONTOUR_EXTRACTOR_VERSION,
|
||||
contour_cache_is_stale,
|
||||
extract_contours,
|
||||
)
|
||||
from B04_PreProcess.B04_PreProcess_Engine_ModelBuild import BUILDERS
|
||||
@@ -191,11 +192,9 @@ def _build_all_terrain_models(
|
||||
output_dir.mkdir(parents=True, exist_ok=True)
|
||||
manifest_path = output_dir / "manifest.json"
|
||||
filters = tuple(key for key in config["source_filters"] if key in ground_masks)
|
||||
methods_list = [key for key in config["precompute"] if key in BUILDERS]
|
||||
if "dtm" in methods_list:
|
||||
methods_list.remove("dtm")
|
||||
methods_list.insert(0, "dtm")
|
||||
methods = tuple(methods_list)
|
||||
# dtm 은 늘 맨 앞 — 같은 필터 외곽 · implicit 굳힌 격자 · 등고선 자르기의 바탕이라
|
||||
# 어느 표현을 지어도 낡았으면(manifest 항목 없음 · 지문 다름) 같이 다시 지음(754-1).
|
||||
methods = ("dtm", *(key for key in config["precompute"] if key in BUILDERS and key != "dtm"))
|
||||
# 무거운 implicit 은 기본 필터(첫 필터)에만 — 분류 LAS 의 csf 까지 두 번 짓지 않음(요구 558).
|
||||
later_methods = tuple(method for method in methods if method != "implicit")
|
||||
signature = config_signature(config)
|
||||
@@ -244,12 +243,14 @@ def _build_all_terrain_models(
|
||||
if not force and _cache_is_valid(output_dir, stem, method, entry, config):
|
||||
interval = float(config.get("contour_interval_meters", 1.0))
|
||||
contour_path = output_dir / f"contour_{filter_key}_{method}_{interval}m.json"
|
||||
needs_contours = not contour_path.exists()
|
||||
needs_contours = contour_cache_is_stale(contour_path, output_dir / f"{stem}.npz")
|
||||
if method in config.get("smoothing_methods", ("dtm", "tin")):
|
||||
smooth_contour_path = (
|
||||
output_dir / f"contour_{filter_key}_{method}_smooth_{interval}m.json"
|
||||
)
|
||||
needs_contours = needs_contours or not smooth_contour_path.exists()
|
||||
needs_contours = needs_contours or contour_cache_is_stale(
|
||||
smooth_contour_path, output_dir / f"{stem}_smooth.npz"
|
||||
)
|
||||
if needs_contours:
|
||||
_cache_contours_safely(
|
||||
output_dir,
|
||||
|
||||
@@ -16,6 +16,7 @@ from fastapi.responses import JSONResponse, Response
|
||||
from B03_FileInput.B03_FileInput_Repository import get_project_storage_relative_path
|
||||
from B04_PreProcess.B04_PreProcess_Engine_Contour import (
|
||||
CONTOUR_EXTRACTOR_VERSION,
|
||||
contour_cache_is_stale,
|
||||
extract_contours,
|
||||
)
|
||||
from common_util.common_util_atomic import atomic_write_bytes
|
||||
@@ -37,10 +38,10 @@ MODEL_REPRESENTATIONS = {
|
||||
|
||||
|
||||
def _is_contour_cache_current(contour_path: Path, model_path: Path) -> bool:
|
||||
"""캐시가 현재 추출기 버전이고 기반 모델 npz보다 최신인지 검사한다."""
|
||||
"""캐시가 현재 추출기 버전이고 모델 · 같은 필터 dtm · 굳힌 격자보다 최신인지 검사한다."""
|
||||
if contour_cache_is_stale(contour_path, model_path):
|
||||
return False
|
||||
try:
|
||||
if contour_path.stat().st_mtime < model_path.stat().st_mtime:
|
||||
return False
|
||||
with contour_path.open("rb") as cache_file:
|
||||
head = cache_file.read(256).decode("utf-8", errors="ignore")
|
||||
except OSError:
|
||||
|
||||
@@ -0,0 +1,102 @@
|
||||
"""754-1 등고선 · 격자 외곽 = 지금 지표면 외곽 (요구 589-11).
|
||||
|
||||
implicit 하나만 지어도 같은 필터 dtm 이 낡았으면 같이 다시 지음 · 등고선 캐시는 dtm · 굳힌 격자가
|
||||
바뀌면 다시 셈 · ptd 처럼 dtm 없던 필터도 굳힌 격자가 생김. 서버 함수만 — 합성 점군 · 임시 폴더.
|
||||
"""
|
||||
|
||||
import json
|
||||
import os
|
||||
from pathlib import Path
|
||||
|
||||
import numpy as np
|
||||
|
||||
from B04_PreProcess.B04_PreProcess_Engine_Contour import contour_cache_is_stale
|
||||
from B04_PreProcess.B04_PreProcess_Engine_Pipeline import build_all_terrain_models
|
||||
from common_util.common_util_surface_sampler import implicit_grid_npz
|
||||
from config import config_system as cfg
|
||||
|
||||
|
||||
def _banded_plane() -> tuple[dict, np.ndarray]:
|
||||
axis = np.arange(0.0, 60.0, 0.5)
|
||||
x, y = np.meshgrid(axis, axis)
|
||||
points = np.column_stack([x.ravel(), y.ravel(), (100.0 + 0.2 * x).ravel()])
|
||||
points = points[~((points[:, 0] > 28) & (points[:, 0] < 33))] # 5 m 빈 띠(z 106 = x 30)
|
||||
data = {"xyz": points, "bounds": np.array([points.min(axis=0), points.max(axis=0)]).T}
|
||||
return data, np.ones(len(points), bool)
|
||||
|
||||
|
||||
def _config(filter_key: str, precompute: list[str], close_m: float) -> dict:
|
||||
config = cfg.build_surface_model_config()
|
||||
config.update(
|
||||
{
|
||||
"source_filters": [filter_key],
|
||||
"precompute": precompute,
|
||||
"footprint_gap_close_meters": close_m,
|
||||
"implicit_min_points": 500,
|
||||
"implicit_max_points": 2_000,
|
||||
}
|
||||
)
|
||||
return config
|
||||
|
||||
|
||||
def _valid_area(path: Path) -> float:
|
||||
with np.load(path) as grid:
|
||||
resolution = float(np.asarray(grid["resolution"]).ravel()[0])
|
||||
return float(np.count_nonzero(grid["valid_mask"])) * resolution**2
|
||||
|
||||
|
||||
def _band_contour(path: Path) -> bool:
|
||||
lines = json.loads(path.read_text(encoding="utf-8"))["contours"]
|
||||
return any(abs(line["level"] - 106.0) < 1e-6 for line in lines)
|
||||
|
||||
|
||||
def test_implicit_만_지어도_낡은_dtm_같이_다시(tmp_path: Path) -> None:
|
||||
data, mask = _banded_plane()
|
||||
masks = {"csf": mask}
|
||||
build_all_terrain_models(data, masks, tmp_path, _config("csf", ["dtm", "implicit"], 1.0))
|
||||
narrow = _valid_area(tmp_path / "dtm_csf.npz")
|
||||
contour = tmp_path / "contour_csf_implicit_1.0m.json"
|
||||
assert not _band_contour(contour) # 1 m 닫기 = 띠 빈칸 · 106 m 줄 잘림
|
||||
|
||||
# 화면 [서피스 생성] = 고른 표현 하나만 — 외곽 설정(지문)이 바뀌었으니 dtm 도 다시
|
||||
manifest = build_all_terrain_models(data, masks, tmp_path, _config("csf", ["implicit"], 3.0))
|
||||
methods = manifest["source_filters"]["csf"]["methods"]
|
||||
assert methods["dtm"]["status"] == "completed"
|
||||
wide = _valid_area(tmp_path / "dtm_csf.npz")
|
||||
grid_area = _valid_area(implicit_grid_npz(tmp_path, "csf"))
|
||||
glb_area = methods["implicit"]["footprint_area_m2"]
|
||||
assert wide > 1.05 * narrow
|
||||
assert grid_area == wide # 격자 외곽 = dtm 외곽
|
||||
assert abs(glb_area - wide) / wide < 0.05 # 3D 외곽 = 격자 외곽
|
||||
assert _band_contour(contour) # 등고선이 넓힌 띠까지
|
||||
assert not contour_cache_is_stale(contour, tmp_path / "implicit_csf.npz")
|
||||
|
||||
|
||||
def test_dtm_이_바뀌면_등고선_캐시_낡음(tmp_path: Path) -> None:
|
||||
model = tmp_path / "implicit_csf.npz"
|
||||
dtm = tmp_path / "dtm_csf.npz"
|
||||
grid = tmp_path / "implicit_csf_grid.npz"
|
||||
contour = tmp_path / "contour_csf_implicit_1.0m.json"
|
||||
assert contour_cache_is_stale(contour, model) # 없음
|
||||
for index, path in enumerate((model, dtm, grid, contour)):
|
||||
path.write_bytes(b"x")
|
||||
os.utime(path, ns=(10**18 + index, 10**18 + index))
|
||||
assert not contour_cache_is_stale(contour, model)
|
||||
for newer in (dtm, grid):
|
||||
os.utime(newer, ns=(2 * 10**18, 2 * 10**18))
|
||||
assert contour_cache_is_stale(contour, model)
|
||||
os.utime(newer, ns=(10**18, 10**18))
|
||||
tin_smooth = tmp_path / "tin_csf_smooth.npz"
|
||||
tin_smooth.write_bytes(b"x")
|
||||
os.utime(tin_smooth, ns=(10**18, 10**18))
|
||||
os.utime(dtm, ns=(2 * 10**18, 2 * 10**18))
|
||||
assert contour_cache_is_stale(contour, tin_smooth) # 스무딩 면도 같은 필터 dtm 을 봄
|
||||
|
||||
|
||||
def test_ptd_implicit_만_지어도_굳힌_격자(tmp_path: Path) -> None:
|
||||
data, mask = _banded_plane()
|
||||
build_all_terrain_models(data, {"ptd": mask}, tmp_path, _config("ptd", ["implicit"], 3.0))
|
||||
assert (tmp_path / "dtm_ptd.npz").is_file()
|
||||
grid = implicit_grid_npz(tmp_path, "ptd") # B05 · B06 표본이 읽는 파일
|
||||
with np.load(grid) as loaded:
|
||||
assert np.isfinite(loaded["z"][loaded["valid_mask"]]).all()
|
||||
Reference in New Issue
Block a user