"""343-1 같은 파일 다시 올림 = 전처리 결과 살림 · 점군 안 풂 · 도엽 병합 · 타일 기억 열쇠. 서버 함수만 시험 — 실제 재올림은 하지 않음(완료 메일). """ from __future__ import annotations import asyncio import json import os from pathlib import Path import numpy as np from B03_FileInput.B03_FileInput_Router_Helpers import ( OutputsWouldBeDiscarded, _confirm_replace_response, describe_existing_outputs, preprocess_input_unchanged, ) from B04_PreProcess.B04_PreProcess_Engine import ( _key_matches, _LazyStructured, _LazyXyz, _source_identity, _structured_stats, ) from B04_PreProcess.B04_PreProcess_Engine_Ground import summarize_masks from common_util.common_util_project_reset import purge_output_folders, purge_output_records def _project(tmp_path: Path) -> tuple[Path, Path]: las = tmp_path / "B03_FileInput" / "input" / "a.las" las.parent.mkdir(parents=True) las.write_bytes(b"LASF" + b"0" * 100) processed = tmp_path / "B04_PreProcess" / "processed" processed.mkdir(parents=True) (processed / "source_identity.json").write_text( json.dumps(_source_identity([las])), encoding="utf-8" ) return tmp_path, las def test_같은_지형_파일이면_전처리_살림(tmp_path: Path) -> None: root, las = _project(tmp_path) assert preprocess_input_unchanged(root, [las]) is True def test_지형_파일이_바뀌면_살리지_않음(tmp_path: Path) -> None: root, las = _project(tmp_path) stat = las.stat() os.utime(las, ns=(stat.st_atime_ns, stat.st_mtime_ns + 5_000_000_000)) assert preprocess_input_unchanged(root, [las]) is False assert preprocess_input_unchanged(root, []) is False def test_전처리_기록이_없으면_살리지_않음(tmp_path: Path) -> None: root, las = _project(tmp_path) (root / "B04_PreProcess" / "processed" / "source_identity.json").unlink() assert preprocess_input_unchanged(root, [las]) is False def test_살리면_확인_창_목록에서_전처리_빠짐(tmp_path: Path) -> None: root, _ = _project(tmp_path) (root / "B05_Profile").mkdir() (root / "B05_Profile" / "route.json").write_text("{}", encoding="utf-8") assert "지표면·전처리 결과" in describe_existing_outputs(root) kept = describe_existing_outputs(root, keep_preprocess=True) assert "지표면·전처리 결과" not in kept assert "노선·종단 설계" in kept def test_확인_응답에_전처리_살림_표시() -> None: body = json.loads(_confirm_replace_response(OutputsWouldBeDiscarded(["횡단 설계"], True)).body) assert body["preprocess_kept"] is True body = json.loads(_confirm_replace_response(OutputsWouldBeDiscarded(["횡단 설계"])).body) assert body["preprocess_kept"] is False def test_폴더_지우기는_전처리만_남김(tmp_path: Path) -> None: root, _ = _project(tmp_path) (root / "B05_Profile").mkdir() (root / "B05_Profile" / "route.json").write_text("{}", encoding="utf-8") removed = purge_output_folders(root, keep_preprocess=True) assert "B04_PreProcess" not in removed assert (root / "B04_PreProcess" / "processed" / "source_identity.json").is_file() assert not (root / "B05_Profile" / "route.json").exists() assert "B04_PreProcess" in purge_output_folders(root) assert not (root / "B04_PreProcess" / "processed").exists() class _Cursor: def __init__(self, log: list[str]) -> None: self.log = log self.rowcount = 0 async def __aenter__(self) -> "_Cursor": return self async def __aexit__(self, *_: object) -> None: return None async def execute(self, sql: str, _params: object = None) -> None: self.log.append(sql) class _Connection: def __init__(self) -> None: self.log: list[str] = [] def cursor(self) -> _Cursor: return _Cursor(self.log) def test_DB_지우기는_지표면_행을_남김() -> None: connection = _Connection() asyncio.run(purge_output_records(connection, "p", keep_preprocess=True)) # type: ignore[arg-type] joined = "\n".join(connection.log) assert "surface_models" not in joined assert "processed_point_cloud" not in joined assert "DELETE FROM routes" in joined connection = _Connection() asyncio.run(purge_output_records(connection, "p")) # type: ignore[arg-type] assert "surface_models" in "\n".join(connection.log) def test_점_수_평균Z_저장값_두번째는_안_풂(tmp_path: Path) -> None: path = tmp_path / "structured.npz" xyz = np.array([[0, 0, 1], [1, 1, 3]], dtype=np.float32) np.savez_compressed(path, xyz=xyz, bounds=np.zeros((3, 2))) with np.load(path) as structured: stats, loaded = _structured_stats(structured, path) assert stats["point_count"] == 2 and stats["mean_z"] == 2.0 assert loaded is not None with np.load(path) as structured: stats, loaded = _structured_stats(structured, path) assert loaded is None and stats["point_count"] == 2 # 구조화가 다시 돌면(파일이 바뀌면) 다시 셈 np.savez_compressed(path, xyz=np.zeros((5, 3), np.float32), bounds=np.zeros((3, 2))) with np.load(path) as structured: stats, loaded = _structured_stats(structured, path) assert stats["point_count"] == 5 and loaded is not None class _NoXyz(dict): files = ["xyz", "bounds", "classification"] def __getitem__(self, key: str) -> object: if key == "xyz": raise AssertionError("점군을 풀면 안 됨") return dict.__getitem__(self, key) def test_저장본이_맞으면_점군을_풀지_않음() -> None: source = _NoXyz(classification=np.zeros(3, np.uint8), bounds=np.zeros((3, 2))) data = _LazyStructured(source, _LazyXyz(source, 3)) summary = summarize_masks(data, {"csf": np.array([True, False, True])}) assert summary["csf"]["total_point_count"] == 3 assert "classification" in data and "rgb" not in data assert len(data["classification"]) == 3 def test_점군은_쓸_때_풂() -> None: xyz = np.arange(9, dtype=np.float32).reshape(3, 3) lazy = _LazyXyz({"xyz": xyz}, 3) assert lazy.shape == (3, 3) and lazy.ndim == 2 assert np.asarray(lazy, dtype=np.float64)[2, 2] == 8.0 assert lazy[1, 0] == 3.0 def test_결과_옆_열쇠_대조(tmp_path: Path) -> None: key_path = tmp_path / "sheet_merge.key.json" key = {"sheets": ["37612001", "37612002"]} assert _key_matches(key_path, key) is False key_path.write_text(json.dumps(key), encoding="utf-8") assert _key_matches(key_path, key) is True assert _key_matches(key_path, {"sheets": ["37612001"]}) is False def test_지도_타일_기억은_파일이_바뀌면_다시_읽음(tmp_path: Path) -> None: from B04_PreProcess.B04_PreProcess_Engine_MvtHelper import load_geojson_cached path = tmp_path / "등고선_bounds.geojson" def write(count: int) -> None: features = [ { "type": "Feature", "properties": {"h": index}, "geometry": {"type": "Point", "coordinates": [127.0 + index * 0.01, 37.0]}, } for index in range(count) ] path.write_text( json.dumps({"type": "FeatureCollection", "features": features}), encoding="utf-8" ) write(1) assert len(load_geojson_cached(path, "t343_contour")) == 1 assert len(load_geojson_cached(path, "t343_contour")) == 1 write(3) stat = path.stat() os.utime(path, ns=(stat.st_atime_ns, stat.st_mtime_ns + 5_000_000_000)) assert len(load_geojson_cached(path, "t343_contour")) == 3 def test_다시_돌면_같은_결과_점군_안_풂(tmp_path: Path, monkeypatch) -> None: # type: ignore[no-untyped-def] """작은 LAS 로 전처리 두 번 — 두 번째는 저장본만 읽고 점군 xyz 를 풀지 않음.""" import laspy import B04_PreProcess.B04_PreProcess_Engine as engine import B04_PreProcess.B04_PreProcess_Engine_SheetSurface as sheet rng = np.random.default_rng(1) count = 4000 x = rng.uniform(200000, 200060, count) y = rng.uniform(500000, 500060, count) header = laspy.LasHeader(point_format=3, version="1.2") header.offsets = [200000, 500000, 0] header.scales = [0.01, 0.01, 0.01] las_data = laspy.LasData(header) las_data.x, las_data.y, las_data.z = x, y, 100 + 0.1 * x - 0.1 * 200000 las_data.classification = np.full(count, 2, np.uint8) las_path = tmp_path / "B03_FileInput" / "input" / "t.las" las_path.parent.mkdir(parents=True) las_data.write(las_path) monkeypatch.setattr(engine, "download_geodata", lambda *a, **k: None) monkeypatch.setattr(sheet, "build_sheet_surface_from_route", lambda *a, **k: []) first = engine.run_surface_analysis(tmp_path, las_path, source_filters=None, methods=["dtm"]) real_lazy_load = engine._LazyXyz._load calls: list[int] = [] def counting_load(self): # type: ignore[no-untyped-def] calls.append(1) return real_lazy_load(self) monkeypatch.setattr(engine._LazyXyz, "_load", counting_load) second = engine.run_surface_analysis(tmp_path, las_path, source_filters=None, methods=["dtm"]) assert first["models"] and calls == [] assert second["processed"] == first["processed"] assert second["ground_summary"] == first["ground_summary"] assert [m["model_file_path"] for m in second["models"]] == [ m["model_file_path"] for m in first["models"] ]