diff --git a/B03_FileInput/B03_FileInput_Service_WF1.py b/B03_FileInput/B03_FileInput_Service_WF1.py index 65e3d2185..2957df93d 100644 --- a/B03_FileInput/B03_FileInput_Service_WF1.py +++ b/B03_FileInput/B03_FileInput_Service_WF1.py @@ -24,6 +24,7 @@ from config.config_system import ( SEND_ANALYSIS_COMPLETION_EMAIL, SHEET_SURFACE_DEFAULT_METHOD, SURFACE_AUTO_METHODS, + SURFACE_CONFIRM_DEFAULT_METHOD, SURFACE_CONTOUR_INTERVAL_M, ) @@ -67,7 +68,8 @@ async def trigger_wf1_analysis_and_email( # 자동 전처리는 기본 조합만 만든다 — 필터는 엔진이 입력 LAS를 보고 정하고 # (분류 있으면 classification, 없으면 csf), 표현은 SURFACE_AUTO_METHODS. # 다른 조합은 관리자가 B04 드롭다운에서 요청할 때 만든다 (2026-09-01 사용자 확정). - methods = list(SURFACE_AUTO_METHODS) + # 확정 표현(implicit)도 함께 지음 — 엔진이 기본 필터에만 지음(요구 558). + methods = list(dict.fromkeys([*SURFACE_AUTO_METHODS, SURFACE_CONFIRM_DEFAULT_METHOD])) params = { "input_file_id": str(input_file_id), "source_filters": None, @@ -176,10 +178,10 @@ async def trigger_wf1_analysis_and_email( else: # 자동 확정은 엔진이 정한 기본 필터를 그대로 따른다 — 이제 그 필터 # 하나만 만들어 두므로 config 기본값과 어긋나면 확정할 모델이 없다. + # 표현 · 평활은 config 확정 기본값(implicit · 평활 끔 — 요구 558). selection = surface_confirmation_defaults() if resolved_filters: selection["source_filter"] = resolved_filters[0] - selection["method"] = methods[0] try: model_id = await find_surface_model_for_selection( connection, project_id, selection diff --git a/B04_PreProcess/B04_PreProcess_Engine.py b/B04_PreProcess/B04_PreProcess_Engine.py index 71f1c0583..becc04c5e 100644 --- a/B04_PreProcess/B04_PreProcess_Engine.py +++ b/B04_PreProcess/B04_PreProcess_Engine.py @@ -279,7 +279,7 @@ def run_surface_analysis( _report(40, "ground_filter", "지면 필터 적용 중") # 자동 전처리는 목록을 넘기지 않는다 — 기본 필터는 입력 LAS를 보고 여기서 정한다. if not source_filters: - source_filters = resolve_auto_source_filters(data) + source_filters = resolve_auto_source_filters(data, processed_dir) logger.info("B04 자동 전처리 기본 필터: %s", ", ".join(source_filters)) masks: dict[str, np.ndarray] = {} for filter_key in source_filters: diff --git a/B04_PreProcess/B04_PreProcess_Engine_Filter_Classification.py b/B04_PreProcess/B04_PreProcess_Engine_Filter_Classification.py index b3475e88a..44373f8b3 100644 --- a/B04_PreProcess/B04_PreProcess_Engine_Filter_Classification.py +++ b/B04_PreProcess/B04_PreProcess_Engine_Filter_Classification.py @@ -5,11 +5,18 @@ 쓸 수 있는지는 has_classified_ground()로 먼저 판정한다. """ +import json +from pathlib import Path from typing import Any import numpy as np -from config.config_system import SURFACE_CLASSIFIED_GROUND_MIN_RATIO +from common_util.common_util_json import atomic_write_json +from config.config_system import ( + SURFACE_CLASSIFIED_COVERAGE_CELL_M, + SURFACE_CLASSIFIED_COVERAGE_MIN_RATIO, + SURFACE_CLASSIFIED_GROUND_MIN_RATIO, +) # ASPRS 표준 지면 클래스 코드 GROUND_CLASS_CODE = 2 @@ -38,3 +45,51 @@ def has_classified_ground( return False ground = int(np.count_nonzero(classification == GROUND_CLASS_CODE)) return ground / total >= minimum_ratio + + +def classified_coverage( + structured_data: dict[str, Any] | np.lib.npyio.NpzFile, + cell_meters: float = SURFACE_CLASSIFIED_COVERAGE_CELL_M, +) -> float: + """점 있는 칸 중 class 2 가 있는 칸의 비율 — 일부만 분류된 LAS 를 가려냄.""" + if "classification" not in structured_data: + return 0.0 + classification = np.asarray(structured_data["classification"]) + xyz = structured_data["xyz"] + bounds = np.asarray(structured_data["bounds"], dtype=np.float64) + cell = max(float(cell_meters), 0.1) + cols = int((bounds[0, 1] - bounds[0, 0]) // cell) + 1 + rows = int((bounds[1, 1] - bounds[1, 0]) // cell) + 1 + occupied = np.zeros((rows, cols), dtype=bool) + ground = np.zeros((rows, cols), dtype=bool) + for start in range(0, len(classification), 5_000_000): + points = np.asarray(xyz[start : start + 5_000_000, :2], dtype=np.float64) + gx = np.clip(((points[:, 0] - bounds[0, 0]) // cell).astype(np.int32), 0, cols - 1) + gy = np.clip(((points[:, 1] - bounds[1, 0]) // cell).astype(np.int32), 0, rows - 1) + occupied[gy, gx] = True + is_ground = classification[start : start + len(points)] == GROUND_CLASS_CODE + ground[gy[is_ground], gx[is_ground]] = True + total = int(np.count_nonzero(occupied)) + return float(np.count_nonzero(ground)) / total if total else 0.0 + + +def has_classified_coverage( + structured_data: dict[str, Any] | np.lib.npyio.NpzFile, + processed_dir: Path | None = None, + minimum_ratio: float = SURFACE_CLASSIFIED_COVERAGE_MIN_RATIO, +) -> bool: + """분류 덮임 문턱 판정 — `processed_dir` 면 structured.npz 지문으로 저장본 재사용.""" + if processed_dir is None: + return classified_coverage(structured_data) >= minimum_ratio + stat = (processed_dir / "structured.npz").stat() + stamp = [int(stat.st_size), int(stat.st_mtime_ns)] + path = processed_dir / "classification_coverage.json" + try: + stored = json.loads(path.read_text(encoding="utf-8")) + coverage = float(stored["coverage"]) if stored.get("structured") == stamp else None + except (OSError, ValueError, KeyError, TypeError, AttributeError): + coverage = None + if coverage is None: + coverage = classified_coverage(structured_data) + atomic_write_json(path, {"structured": stamp, "coverage": coverage}) + return coverage >= minimum_ratio diff --git a/B04_PreProcess/B04_PreProcess_Engine_Ground.py b/B04_PreProcess/B04_PreProcess_Engine_Ground.py index f178a6064..a7021f9e8 100644 --- a/B04_PreProcess/B04_PreProcess_Engine_Ground.py +++ b/B04_PreProcess/B04_PreProcess_Engine_Ground.py @@ -5,12 +5,14 @@ source_filters를 따른다. """ +from pathlib import Path from typing import Any import numpy as np from B04_PreProcess.B04_PreProcess_Engine_Filter_Classification import ( filter_classification, + has_classified_coverage, has_classified_ground, ) from B04_PreProcess.B04_PreProcess_Engine_Filter_CSF import filter_csf @@ -45,14 +47,18 @@ def usable_conditional_filters( def resolve_auto_source_filters( structured_data: dict[str, Any] | np.lib.npyio.NpzFile, + processed_dir: Path | None = None, ) -> list[str]: """자동 전처리가 만들 필터를 입력 LAS를 보고 정한다. 첫 항목이 기본 확정값이다. 업체가 이미 지면을 분류해 놨으면 그것을 기본으로 쓴다 — 계산이 없고 품질도 가장 안정적이다. **csf는 분류가 있어도 함께 만든다** — csf 처리 결과를 눈으로 검증할 수 있어야 하기 때문이다 (2026-09-01 사용자 확정). 분류가 없으면 csf 하나만 남는다. + 「분류 있음」 = 비율 문턱 + 덮임(일부만 분류된 LAS 거름) — 자동 고름에만(요구 558). """ usable = usable_conditional_filters(structured_data) + if usable and not has_classified_coverage(structured_data, processed_dir): + usable = [] return [*usable[:1], SURFACE_AUTO_FALLBACK_FILTER] diff --git a/B04_PreProcess/B04_PreProcess_Engine_ModelBuild.py b/B04_PreProcess/B04_PreProcess_Engine_ModelBuild.py index baa33c69f..f80d73b91 100644 --- a/B04_PreProcess/B04_PreProcess_Engine_ModelBuild.py +++ b/B04_PreProcess/B04_PreProcess_Engine_ModelBuild.py @@ -168,11 +168,19 @@ def build_nurbs( ) +def implicit_point_count(ground_count: int, config: dict[str, Any]) -> int: + """implicit 중심점 수 = clip(지면점 × 비율, 하한, 상한) · 지면점이 하한보다 적으면 전부.""" + minimum = int(config["implicit_min_points"]) + if ground_count <= minimum: + return int(ground_count) + ratio = round(ground_count * float(config["implicit_point_ratio"])) + return int(min(max(ratio, minimum), int(config["implicit_max_points"]))) + + def build_implicit( context: TerrainContext, output_dir: Path, stem: str, progress: ProgressCallback ) -> dict[str, Any]: - maximum = max(100, int(context.config["implicit_max_points_per_tile"])) - points = context.sample(maximum) + points = context.grid_sample(implicit_point_count(context.source_count, context.config)) unique_xy, unique_indices = np.unique(points[:, :2], axis=0, return_index=True) points = points[unique_indices] if len(points) < 4: diff --git a/B04_PreProcess/B04_PreProcess_Engine_ModelContext.py b/B04_PreProcess/B04_PreProcess_Engine_ModelContext.py index 13300da3c..a5efb1ca3 100644 --- a/B04_PreProcess/B04_PreProcess_Engine_ModelContext.py +++ b/B04_PreProcess/B04_PreProcess_Engine_ModelContext.py @@ -29,6 +29,8 @@ ProgressCallback = Callable[[int], None] # 대용량 포인트 배치 처리 크기 _BATCH_SIZE = 500_000 +# implicit 격자 칸 뽑기 씨앗 — 고정이라 다시 지어도 같은 중심점 +IMPLICIT_SAMPLE_SEED = 558 # NURBS 평활 계수 — 제어점당 허용 잔차제곱(m²). # 0.01(RMS 0.1m)은 요구가 너무 빡빡해 FITPACK이 제어점 수(122x113)보다 많은 @@ -262,6 +264,30 @@ class TerrainContext: self._samples[maximum] = points return points + def grid_sample(self, count: int, seed: int = IMPLICIT_SAMPLE_SEED) -> np.ndarray: + """implicit 전용 — 격자 칸(√(외곽 넓이 / 점 수))마다 하나 · 모자라면 무작위로 채움. + + 파일 차례 뽑기는 점 많은 트인 곳에 몰려 숲 밑 성긴 곳이 빔(요구 558). 씨앗 고정. + """ + indices = self.indices() + if len(indices) <= count: + return np.asarray(self.xyz[indices], dtype=np.float32) + _, _, resolution, footprint = self.footprint() + cell = max(math.sqrt(float(footprint.sum()) * resolution * resolution / count), 0.01) + xy = np.asarray(self.xyz[indices][:, :2], dtype=np.float64) + rng = np.random.default_rng(seed) + order = rng.permutation(len(indices)) + gx = ((xy[order, 0] - self.bounds[0, 0]) // cell).astype(np.int64) + gy = ((xy[order, 1] - self.bounds[1, 0]) // cell).astype(np.int64) + _, first = np.unique(gy * (int(gx.max()) + 1) + gx, return_index=True) + chosen = order[first] + if len(chosen) > count: + chosen = rng.choice(chosen, count, replace=False) + elif len(chosen) < count: + rest = np.setdiff1d(np.arange(len(indices)), chosen, assume_unique=True) + chosen = np.concatenate([chosen, rng.choice(rest, count - len(chosen), replace=False)]) + return np.asarray(self.xyz[indices[np.sort(chosen)]], dtype=np.float32) + def footprint(self) -> tuple[float, float, float, np.ndarray]: if self._footprint is not None: return self._footprint diff --git a/B04_PreProcess/B04_PreProcess_Engine_Pipeline.py b/B04_PreProcess/B04_PreProcess_Engine_Pipeline.py index 8fb66c937..f2f8766bf 100644 --- a/B04_PreProcess/B04_PreProcess_Engine_Pipeline.py +++ b/B04_PreProcess/B04_PreProcess_Engine_Pipeline.py @@ -151,7 +151,13 @@ _REPRESENTATIONS = { def _cache_is_valid( output_dir: Path, stem: str, method: str, entry: dict[str, Any], config: dict[str, Any] ) -> bool: - """디스크의 결과 파일과 스무딩 메타데이터가 유효한지 검사한다.""" + """디스크의 결과 파일과 스무딩 메타데이터가 유효한지 검사한다. + + manifest 항목이 완료여야 함 — 설정 지문이 바뀌면 manifest 가 비므로 파일만 남은 + 옛 결과(평활 없는 implicit · nurbs · meshfree)를 다시 짓는다(요구 558). + """ + if entry.get("status") != "completed": + return False ext = "ply" if method == "meshfree" else "glb" files_exist = (output_dir / f"{stem}_preview.{ext}").exists() and ( output_dir / f"{stem}.npz" @@ -190,6 +196,8 @@ def _build_all_terrain_models( methods_list.remove("dtm") methods_list.insert(0, "dtm") methods = tuple(methods_list) + # 무거운 implicit 은 기본 필터(첫 필터)에만 — 분류 LAS 의 csf 까지 두 번 짓지 않음(요구 558). + later_methods = tuple(method for method in methods if method != "implicit") signature = config_signature(config) bounds = np.asarray(structured_data["bounds"], dtype=np.float64) xyz = structured_data["xyz"] @@ -211,7 +219,7 @@ def _build_all_terrain_models( } started_at = time.monotonic() timeout = max(0, int(config.get("sync_timeout_seconds", 0))) - total_units = max(1, len(filters) * len(methods)) + total_units = max(1, len(methods) + (len(filters) - 1) * len(later_methods)) done_units = 0 failures = 0 @@ -229,27 +237,11 @@ def _build_all_terrain_models( ) filter_entry["source_point_count"] = context.source_count - for method in methods: + for method in later_methods if filter_index else methods: stem = f"{method}_{filter_key}" entry = filter_entry["methods"].get(method, {}) if not force and _cache_is_valid(output_dir, stem, method, entry, config): - if entry.get("status") != "completed": - entry.update( - { - "status": "completed", - "representation": _REPRESENTATIONS[method], - "model_file": f"{stem}.npz", - "preview_file": f"{stem}_preview." - + ("ply" if method == "meshfree" else "glb"), - "preview_media_type": "application/octet-stream" - if method == "meshfree" - else "model/gltf-binary", - "error": None, - } - ) - filter_entry["methods"][method] = entry - _write_json_file(manifest_path, manifest) 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() diff --git a/B04_PreProcess/B04_PreProcess_UI_Page.ts b/B04_PreProcess/B04_PreProcess_UI_Page.ts index 360e82fc2..6a525a9ef 100644 --- a/B04_PreProcess/B04_PreProcess_UI_Page.ts +++ b/B04_PreProcess/B04_PreProcess_UI_Page.ts @@ -54,7 +54,7 @@ import { currentProjectId, guardProjectLeave } from "../A00_Common/b_project_swi const SOURCE_FILTERS = ["classification", "grid_min_z", "csf", "pmf"] as const; const MODEL_METHODS = ["tin", "dtm", "nurbs", "implicit", "meshfree"] as const; const DEFAULT_FILTER = "csf"; -const DEFAULT_METHOD = "dtm"; +const DEFAULT_METHOD = "implicit"; // 확정 기본 표현(요구 558) — 확정본이 있으면 그것이 먼저 const ROUTE_STAGE = ROUTES.B05_PROFILE; // 도엽 서피스 보간 방식 버튼 순서 — 백엔드 SHEET_SURFACE_METHODS와 같은 차례로 둔다. const SHEET_METHOD_ORDER = ["tin_sheet", "tin", "biharmonic", "anudem", "multires", "laplace"]; @@ -494,7 +494,7 @@ export async function renderB04Surface(root: HTMLElement): Promise { ]); models = modelResponse.models; // 확정본과 같은 조합에서 시작해야 B05와 같은 파일을 보고, 보관함도 한 벌만 쓴다. - // 확정 이력이 없을 때만 개발 기본값(csf·dtm)으로 둔다. + // 확정 이력이 없을 때만 개발 기본값(csf·implicit)으로 둔다. if (confirmed.model_id) { if (confirmed.source_filter) filterGroup.select.value = confirmed.source_filter; if (confirmed.method) methodGroup.select.value = confirmed.method; diff --git a/config/config_system_terrain.py b/config/config_system_terrain.py index 2e83d2fd7..e28ab794b 100644 --- a/config/config_system_terrain.py +++ b/config/config_system_terrain.py @@ -63,6 +63,11 @@ SURFACE_RANSAC_SEED = int(os.getenv("SURFACE_RANSAC_SEED", "42")) SURFACE_CLASSIFIED_GROUND_MIN_RATIO = float( os.getenv("SURFACE_CLASSIFIED_GROUND_MIN_RATIO", "0.002") ) +# 자동 기본 고름만 — 점 있는 칸(한 변 아래 m) 중 class 2 가 있는 칸이 이 비율 이상이어야 분류 있음. +SURFACE_CLASSIFIED_COVERAGE_CELL_M = float(os.getenv("SURFACE_CLASSIFIED_COVERAGE_CELL_M", "10.0")) +SURFACE_CLASSIFIED_COVERAGE_MIN_RATIO = float( + os.getenv("SURFACE_CLASSIFIED_COVERAGE_MIN_RATIO", "0.5") +) # 지면점 비율이 이 값 미만이면 필터가 사실상 실패한 것으로 보고 WARNING을 남긴다. SURFACE_GROUND_RATIO_WARN = float(os.getenv("SURFACE_GROUND_RATIO_WARN", "0.01")) @@ -111,9 +116,10 @@ SURFACE_NURBS_PATCH_SIZE_M = float(os.getenv("SURFACE_NURBS_PATCH_SIZE_M", "50.0 SURFACE_NURBS_CONTROL_POINTS_PER_AXIS = int( os.getenv("SURFACE_NURBS_CONTROL_POINTS_PER_AXIS", "16") ) -SURFACE_IMPLICIT_MAX_POINTS_PER_TILE = int( - os.getenv("SURFACE_IMPLICIT_MAX_POINTS_PER_TILE", "10000") -) +# implicit 중심점 수 = clip(지면점 × 비율, 하한, 상한) · 지면점이 하한보다 적으면 전부(요구 558). +SURFACE_IMPLICIT_POINT_RATIO = float(os.getenv("SURFACE_IMPLICIT_POINT_RATIO", "0.175")) +SURFACE_IMPLICIT_MIN_POINTS = int(os.getenv("SURFACE_IMPLICIT_MIN_POINTS", "50000")) +SURFACE_IMPLICIT_MAX_POINTS = int(os.getenv("SURFACE_IMPLICIT_MAX_POINTS", "2000000")) SURFACE_IMPLICIT_SMOOTHING = float(os.getenv("SURFACE_IMPLICIT_SMOOTHING", "0.1")) SURFACE_MESHFREE_MAX_MODEL_POINTS = int(os.getenv("SURFACE_MESHFREE_MAX_MODEL_POINTS", "500000")) SURFACE_MESHFREE_POINT_RADIUS_M = float(os.getenv("SURFACE_MESHFREE_POINT_RADIUS_M", "0.15")) @@ -166,9 +172,10 @@ SHEET_SURFACE_AUTO_METHODS = [ # 일반 사용자 WF1 자동 확정 기본값 SURFACE_CONFIRM_DEFAULT_FILTER = os.getenv("SURFACE_CONFIRM_DEFAULT_FILTER", "csf") -SURFACE_CONFIRM_DEFAULT_METHOD = os.getenv("SURFACE_CONFIRM_DEFAULT_METHOD", "dtm") +# 확정 표현 = implicit · 평활 끔 — WF1 자동 짓기도 이 표현을 기본 필터에 지음(요구 558). +SURFACE_CONFIRM_DEFAULT_METHOD = os.getenv("SURFACE_CONFIRM_DEFAULT_METHOD", "implicit") SURFACE_CONFIRM_DEFAULT_SMOOTH = ( - os.getenv("SURFACE_CONFIRM_DEFAULT_SMOOTH", "True").lower() == "true" + os.getenv("SURFACE_CONFIRM_DEFAULT_SMOOTH", "False").lower() == "true" ) @@ -190,7 +197,9 @@ def build_surface_model_config() -> dict: "nurbs_degree": SURFACE_NURBS_DEGREE, "nurbs_patch_size_meters": SURFACE_NURBS_PATCH_SIZE_M, "nurbs_control_points_per_axis": SURFACE_NURBS_CONTROL_POINTS_PER_AXIS, - "implicit_max_points_per_tile": SURFACE_IMPLICIT_MAX_POINTS_PER_TILE, + "implicit_point_ratio": SURFACE_IMPLICIT_POINT_RATIO, + "implicit_min_points": SURFACE_IMPLICIT_MIN_POINTS, + "implicit_max_points": SURFACE_IMPLICIT_MAX_POINTS, "implicit_smoothing": SURFACE_IMPLICIT_SMOOTHING, "meshfree_max_model_points": SURFACE_MESHFREE_MAX_MODEL_POINTS, "meshfree_point_radius_meters": SURFACE_MESHFREE_POINT_RADIUS_M, diff --git a/resources/tester/test_640_lidar_default.py b/resources/tester/test_640_lidar_default.py new file mode 100644 index 000000000..f92ee1b75 --- /dev/null +++ b/resources/tester/test_640_lidar_default.py @@ -0,0 +1,136 @@ +"""640 라이다 기본 고름 · implicit 점 수 · 격자 칸 뽑기 · 지문 · 틈 닫기 (요구 558). + +서버 함수만 — 작은 합성 점군 · 임시 폴더. +""" + +from pathlib import Path + +import numpy as np + +from B04_PreProcess.B04_PreProcess_Engine_Filter_Classification import ( + classified_coverage, + has_classified_coverage, +) +from B04_PreProcess.B04_PreProcess_Engine_Ground import resolve_auto_source_filters +from B04_PreProcess.B04_PreProcess_Engine_ModelBuild import implicit_point_count +from B04_PreProcess.B04_PreProcess_Engine_ModelContext import TerrainContext +from B04_PreProcess.B04_PreProcess_Engine_Pipeline import build_all_terrain_models +from config import config_system as cfg + + +def _plane(size_m: float = 60.0, spacing_m: float = 0.5) -> np.ndarray: + axis = np.arange(0.0, size_m, spacing_m) + x, y = np.meshgrid(axis, axis) + z = 100.0 + 0.2 * x + 0.1 * np.sin(y / 5.0) + return np.column_stack([x.ravel(), y.ravel(), z.ravel()]) + + +def _structured(points: np.ndarray, classification: np.ndarray | None = None) -> dict: + data = {"xyz": points, "bounds": np.array([points.min(axis=0), points.max(axis=0)]).T} + if classification is not None: + data["classification"] = classification + return data + + +def test_확정_기본값은_implicit_평활_끔() -> None: + assert cfg.SURFACE_CONFIRM_DEFAULT_METHOD == "implicit" + assert cfg.SURFACE_CONFIRM_DEFAULT_SMOOTH is False + assert cfg.SURFACE_AUTO_METHODS == ("dtm",) # D03 자체 라이다는 dtm 만 그대로 + + +def test_일부만_분류된_LAS_는_자동_기본에서_csf() -> None: + points = _plane(100.0) + partial = np.ones(len(points), dtype=np.uint8) + partial[(points[:, 0] < 30) & (np.arange(len(points)) % 4 == 0)] = 2 # 비율 7 % · 덮임 30 % + assert classified_coverage(_structured(points, partial)) < 0.5 + assert resolve_auto_source_filters(_structured(points, partial)) == ["csf"] + + spread = np.ones(len(points), dtype=np.uint8) + spread[::37] = 2 # 비율 2.7 % · 덮임 100 % + assert classified_coverage(_structured(points, spread)) == 1.0 + assert resolve_auto_source_filters(_structured(points, spread)) == ["classification", "csf"] + + +def test_덮임은_저장본을_다시_씀(tmp_path: Path) -> None: + points = _plane(40.0) + classes = np.full(len(points), 2, dtype=np.uint8) + (tmp_path / "structured.npz").write_bytes(b"x") + assert has_classified_coverage(_structured(points, classes), tmp_path) + assert (tmp_path / "classification_coverage.json").is_file() + + class _NoXyz(dict): + def __getitem__(self, key): # type: ignore[no-untyped-def] + assert key != "xyz", "저장본이 맞으면 점군을 풀지 않아야 함" + return dict.__getitem__(self, key) + + assert has_classified_coverage(_NoXyz(_structured(points, classes)), tmp_path) + + +def _config(**override: object) -> dict: + config = cfg.build_surface_model_config() + config.update({"source_filters": ["a", "b"], "precompute": ["dtm", "implicit"]}) + config.update(override) + return config + + +def test_implicit_은_기본_필터에만(tmp_path: Path) -> None: + points = _plane() + data = _structured(points) + masks = {"a": np.ones(len(points), bool), "b": np.arange(len(points)) % 2 == 0} + manifest = build_all_terrain_models(data, masks, tmp_path, _config()) + assert set(manifest["source_filters"]["a"]["methods"]) == {"dtm", "implicit"} + assert set(manifest["source_filters"]["b"]["methods"]) == {"dtm"} + assert not (tmp_path / "implicit_b.npz").exists() + + +def test_implicit_점_수_비율_하한_상한() -> None: + config = cfg.build_surface_model_config() + assert (config["implicit_point_ratio"], config["implicit_min_points"]) == (0.175, 50_000) + assert config["implicit_max_points"] == 2_000_000 + assert implicit_point_count(1_140_716, config) == 199_625 # 용화 분류 + assert implicit_point_count(768_273, config) == 134_448 # 용화 CSF + assert implicit_point_count(100_000, config) == 50_000 # 하한 + assert implicit_point_count(30_000, config) == 30_000 # 하한보다 적으면 전부 + assert implicit_point_count(50_000_000, config) == 2_000_000 # 상한 + + +def test_격자_칸_뽑기는_고르고_씨앗_고정() -> None: + rng = np.random.default_rng(1) + dense = rng.uniform(0, 20, (200_000, 2)) # 왼쪽 아래 트인 곳 — 점 몰림 + sparse = rng.uniform(0, 100, (30_000, 2)) # 전체 성긴 숲 밑 + xy = np.vstack([dense, sparse]) + points = np.column_stack([xy, np.zeros(len(xy))]) + context = TerrainContext( + xyz=points, + mask=np.ones(len(points), bool), + bounds=np.array([points.min(axis=0), points.max(axis=0)]).T, + config=cfg.build_surface_model_config(), + ) + picked = context.grid_sample(5_000) + assert len(picked) == 5_000 and len(np.unique(picked[:, :2], axis=0)) == 5_000 + assert np.array_equal(picked, context.grid_sample(5_000)) + in_dense = np.count_nonzero((picked[:, 0] < 20) & (picked[:, 1] < 20)) / len(picked) + file_order = context.sample(5_000) + in_dense_file = np.count_nonzero((file_order[:, 0] < 20) & (file_order[:, 1] < 20)) + assert in_dense < 0.1 < 0.5 < in_dense_file / len(file_order) # 파일 차례는 몰림 · 격자는 고름 + assert len(context.grid_sample(300_000)) == len(points) # 모자라면 전부 + + +def test_설정_지문이_바뀌면_implicit_다시_지음(tmp_path: Path) -> None: + points = _plane() + data = _structured(points) + masks = {"a": np.ones(len(points), bool)} + config = _config( + source_filters=["a"], + precompute=["implicit"], + implicit_min_points=500, + implicit_max_points=1_000, + ) + build_all_terrain_models(data, masks, tmp_path, config) + with np.load(tmp_path / "implicit_a.npz") as old: + assert len(old["center_z"]) == 1_000 + changed = dict(config, implicit_max_points=2_000) + manifest = build_all_terrain_models(data, masks, tmp_path, changed) + with np.load(tmp_path / "implicit_a.npz") as new: + assert len(new["center_z"]) == 2_000 + assert manifest["source_filters"]["a"]["methods"]["implicit"]["center_count"] == 2_000