from __future__ import annotations import tempfile import threading import unittest from pathlib import Path from unittest import mock import numpy as np import trimesh import utils.terrain_model_converter as terrain_converter from utils.contour_extractor import _tin_face_coverage_mask from utils.terrain_model_converter import MODEL_METHODS, build_all_terrain_models class TerrainModelConverterTests(unittest.TestCase): def setUp(self) -> None: x, y = np.meshgrid(np.linspace(0, 20, 21), np.linspace(0, 15, 16)) z = 100 + 0.1 * x + 0.2 * y + np.sin(x / 4) self.xyz = np.column_stack([x.ravel(), y.ravel(), z.ravel()]) self.bounds = np.array([[0, 20], [0, 15], [float(z.min()), float(z.max())]]) count = len(self.xyz) self.masks = { "grid_min_z": np.ones(count, dtype=bool), "csf": np.arange(count) % 5 != 0, "pmf": np.arange(count) % 7 != 0, } self.config = { "source_filters": ("grid_min_z", "csf", "pmf"), "precompute": MODEL_METHODS, "max_preview_vertices": 1_000, "tile_size_meters": 10.0, "sync_timeout_seconds": 60, "tin_max_input_points": 1_000, "dtm_grid_resolution_meters": 1.0, "nurbs_degree": 3, "nurbs_patch_size_meters": 10.0, "nurbs_control_points_per_axis": 8, "implicit_max_points_per_tile": 300, "implicit_smoothing": 0.1, "meshfree_max_model_points": 1_000, "meshfree_point_radius_meters": 0.15, } def test_builds_and_reuses_all_fifteen_combinations(self) -> None: with tempfile.TemporaryDirectory() as temporary: output_dir = Path(temporary) structured = {"xyz": self.xyz, "bounds": self.bounds} manifest = build_all_terrain_models(structured, self.masks, output_dir, self.config) self.assertEqual(manifest["status"], "completed") self.assertEqual(manifest["failure_count"], 0) self.assertEqual(set(manifest["source_filters"]), set(self.masks)) for source_filter in self.masks: methods = manifest["source_filters"][source_filter]["methods"] self.assertEqual(set(methods), set(MODEL_METHODS)) self.assertTrue(all(entry["status"] == "completed" for entry in methods.values())) glb_path = output_dir / "dtm_csf_preview.glb" loaded = trimesh.load(glb_path, force="scene") self.assertGreater(len(loaded.geometry), 0) self.assertTrue((output_dir / "meshfree_pmf_preview.ply").read_bytes().startswith(b"ply\n")) with np.load(output_dir / "tin_csf.npz") as original, np.load( output_dir / "tin_csf_smooth.npz" ) as smoothed: np.testing.assert_array_equal(smoothed["vertices"][:, :2], original["vertices"][:, :2]) np.testing.assert_array_equal(smoothed["faces"], original["faces"]) mtimes = {path.name: path.stat().st_mtime_ns for path in output_dir.iterdir() if path.suffix != ".json"} cached = build_all_terrain_models(structured, self.masks, output_dir, self.config) self.assertEqual(cached["failure_count"], 0) self.assertEqual( mtimes, {path.name: path.stat().st_mtime_ns for path in output_dir.iterdir() if path.suffix != ".json"}, ) def test_failed_filter_is_isolated_from_other_results(self) -> None: with tempfile.TemporaryDirectory() as temporary: masks = dict(self.masks) masks["pmf"] = np.zeros(len(self.xyz), dtype=bool) manifest = build_all_terrain_models( {"xyz": self.xyz, "bounds": self.bounds}, masks, Path(temporary), self.config, ) self.assertEqual(manifest["status"], "completed_with_errors") self.assertEqual(manifest["failure_count"], len(MODEL_METHODS)) for source_filter in ("grid_min_z", "csf"): self.assertTrue(all( entry["status"] == "completed" for entry in manifest["source_filters"][source_filter]["methods"].values() )) self.assertTrue(all( entry["status"] == "failed" for entry in manifest["source_filters"]["pmf"]["methods"].values() )) def test_footprint_removes_outlier_and_applies_one_meter_inset(self) -> None: x, y = np.meshgrid(np.arange(0, 11), np.arange(0, 11)) cluster = np.column_stack([x.ravel(), y.ravel(), 100 + 0.1 * x.ravel()]) xyz = np.vstack([cluster, np.array([[20.0, 20.0, 130.0]])]) bounds = np.array([[0, 20], [0, 20], [100, 130]], dtype=float) config = dict(self.config) config.update({ "source_filters": ("csf",), "precompute": ("dtm",), "footprint_resolution_meters": 1.0, "footprint_gap_close_meters": 1.0, "boundary_inset_meters": 1.0, "keep_largest_footprint": True, }) with tempfile.TemporaryDirectory() as temporary: output = Path(temporary) manifest = build_all_terrain_models( {"xyz": xyz, "bounds": bounds}, {"csf": np.ones(len(xyz), dtype=bool)}, output, config, ) with np.load(output / "dtm_csf.npz") as model: valid = model["valid_mask"] self.assertFalse(bool(valid[-1, -1])) self.assertFalse(bool(valid[0, 0])) self.assertTrue(bool(valid[5, 5])) metadata = manifest["source_filters"]["csf"]["methods"]["dtm"] self.assertEqual(metadata["boundary_inset_meters"], 1.0) self.assertLess(metadata["footprint_area_m2"], 121.0) def test_tin_contour_mask_uses_saved_faces_instead_of_vertex_hull(self) -> None: # 두 개의 분리된 TIN 조각 사이를 정점 convex hull은 채우지만, # 저장된 faces 기반 coverage는 가운데 빈 영역을 남겨야 한다. vertices = np.array([ [0.0, 0.0, 0.0], [1.0, 0.0, 0.0], [0.0, 1.0, 0.0], [3.0, 0.0, 0.0], [4.0, 0.0, 0.0], [4.0, 1.0, 0.0], ]) faces = np.array([[0, 1, 2], [3, 4, 5]], dtype=np.uint32) xx, yy = np.meshgrid(np.array([0.25, 2.0, 3.75]), np.array([0.25])) mask = _tin_face_coverage_mask(vertices, faces, xx, yy) self.assertEqual(mask.tolist(), [[True, False, True]]) def test_duplicate_build_request_is_cancelled_without_waiting(self) -> None: started = threading.Event() release = threading.Event() first_result: list[dict] = [] def slow_build(*_args, **_kwargs): started.set() release.wait(timeout=2) return {"status": "completed", "failure_count": 0} with tempfile.TemporaryDirectory() as temporary, mock.patch.object( terrain_converter, "_build_all_terrain_models", side_effect=slow_build ): worker = threading.Thread( target=lambda: first_result.append(terrain_converter.build_all_terrain_models( {}, {}, Path(temporary), {}, force=False )) ) worker.start() self.assertTrue(started.wait(timeout=1)) duplicate = terrain_converter.build_all_terrain_models( {}, {}, Path(temporary), {}, force=False ) self.assertEqual(duplicate["request_status"], "cancelled_already_running") self.assertTrue(worker.is_alive()) release.set() worker.join(timeout=2) self.assertEqual(first_result[0]["status"], "completed") if __name__ == "__main__": unittest.main()