diff --git a/B05_Profile/B05_Profile_Router_Lidar.py b/B05_Profile/B05_Profile_Router_Lidar.py index 2d6bed715..97c8f0241 100644 --- a/B05_Profile/B05_Profile_Router_Lidar.py +++ b/B05_Profile/B05_Profile_Router_Lidar.py @@ -18,6 +18,8 @@ DB = `surface_models` 넣기/고치기만(`keep` — 갈음 때 id · 확정 · `POST /route/design-terrain`(갈음 뒤 같은 지형 = `force`) 할 때만. 노선 덮임 = 저장 노선을 2 m 마다 새 지형에서 표본 — 빈칸 있으면 `covered` 거짓 (화면이 라이다 바꿈 막음). +설계 지형이 라이다인 B = 시험 자리(`lidar_try/`)에 따로 지어 덮임 검사 → 다 덮을 때만 본 자리로 +옮김 · 못 덮으면 거절(옛 LAS · 필터 파일 · 행 · 메타 그대로 · 시험 자리 지움 · PLAN 637-5). 일 동안 [저장]과 같은 프로젝트 잠금(`common_util_save_guard`). 계산 자리 = ③ 서버 단독. @@ -27,6 +29,7 @@ from __future__ import annotations import asyncio import json +import shutil from pathlib import Path from typing import Any from uuid import UUID @@ -56,6 +59,7 @@ UPLOAD = PRE / "lidar_upload" LAS_DIR = PRE / "lidar_las" META = PRE / "lidar_meta.json" JOB = PRE / "lidar_job.json" +TRY = PRE / "lidar_try" STEP_M = 2.0 _KEYS = ("source_filter", "method", "smooth") @@ -96,8 +100,10 @@ def line_points(line: list[list[float]], step: float = STEP_M) -> np.ndarray: return np.vstack(out) -def route_coverage(root: Path, selection: dict[str, Any]) -> tuple[bool, float]: - """저장 노선이 새 라이다 지형에 다 덮이나 — (덮임, 덮인 비율).""" +def route_coverage( + root: Path, selection: dict[str, Any], models: Path | None = None +) -> tuple[bool, float]: + """저장 노선이 새 라이다 지형(`models` · 없으면 뿌리)에 다 덮이나 — (덮임, 덮인 비율).""" from B05_Profile.B05_Profile_Router_Replan import _planned_line from common_util.common_util_surface_sampler import build_surface_sampler @@ -105,7 +111,7 @@ def route_coverage(root: Path, selection: dict[str, Any]) -> tuple[bool, float]: if len(line) < 2: return False, 0.0 sampler = build_surface_sampler( - root / MODELS, + models or root / MODELS, str(selection["source_filter"]), str(selection["method"]), bool(selection["smooth"]), @@ -115,19 +121,32 @@ def route_coverage(root: Path, selection: dict[str, Any]) -> tuple[bool, float]: def build_route_lidar( - root: Path, epsg: int, target: dict[str, Any] | None, report: Any + root: Path, epsg: int, target: dict[str, Any] | None, report: Any, guard: bool = False +) -> dict[str, Any]: + """올린 LAS → `lidar_las/` 갈음 → 뿌리에 짓기(있던 라이다 필터 · 없으면 자동) → 덮임 → 메타. + + `guard`(설계 지형이 라이다) = 시험 자리에 지어 노선을 다 덮을 때만 뿌리로 옮김 · + 못 덮으면 ValueError(뿌리 그대로).""" + shutil.rmtree(root / TRY, ignore_errors=True) + try: + return _build(root, root / TRY if guard else root, epsg, target, report) + finally: + shutil.rmtree(root / TRY, ignore_errors=True) + + +def _build( + root: Path, base: Path, epsg: int, target: dict[str, Any] | None, report: Any ) -> dict[str, Any]: - """올린 LAS → `lidar_las/` 갈음 → 뿌리에 짓기(있던 라이다 필터 · 없으면 자동) → 덮임 → 메타.""" from config.config_system import ( SURFACE_AUTO_METHODS, SURFACE_CONFIRM_DEFAULT_METHOD, SURFACE_CONFIRM_DEFAULT_SMOOTH, ) - las_paths = lidar_upload.stage_las(root / UPLOAD, root / LAS_DIR, epsg, report) + las_paths = lidar_upload.stage_las(root / UPLOAD, base / LAS_DIR, epsg, report) method = str((target or {}).get("method") or SURFACE_CONFIRM_DEFAULT_METHOD) result = lidar_upload.build_surface( - root, + base, las_paths, report, [str(target["source_filter"])] if target else None, @@ -139,7 +158,14 @@ def build_route_lidar( "smooth": bool(SURFACE_CONFIRM_DEFAULT_SMOOTH), } report(96, "노선 덮임 보는 중") - covered, ratio = route_coverage(root, selection) + covered, ratio = route_coverage(root, selection, base / MODELS) + if base != root: + if not covered: + raise ValueError( + f"노선을 {ratio * 100:.0f} % 만 덮음 — 설계 지형이 라이다라 바꾸지 않음" + ) + lidar_upload.promote_build(base, root) + las_paths = [root / LAS_DIR / p.name for p in las_paths] meta = { "epsg": int(epsg), "files": [p.name for p in las_paths], @@ -184,7 +210,9 @@ async def upload_route_lidar(project_id: UUID, files: list[UploadFile] = File(.. async def work(report: Any) -> None: async with lock: - got = await asyncio.to_thread(build_route_lidar, root, epsg, target, report) + got = await asyncio.to_thread( + build_route_lidar, root, epsg, target, report, design == "las" + ) await lidar_upload.save_models( project_id, got["models"], epsg, keep=True, drop=got["dropped"] ) diff --git a/common_util/common_util_lidar_upload.py b/common_util/common_util_lidar_upload.py index 52c958af8..dd9f37985 100644 --- a/common_util/common_util_lidar_upload.py +++ b/common_util/common_util_lidar_upload.py @@ -4,6 +4,7 @@ (좌표계 다르면 x · y 옮겨 적음) - 받기 고리 `receive_uploads` · 입력 자리 갈음 `stage_las` · 짓기 `build_surface` (B04 엔진 `run_surface_analysis(maps=False)` — 지도 · GIS 받기 · 도엽 서피스 안 함) +- 시험 자리에 지은 것 본 자리로 `promote_build`(B05 설계 라이다 · 덮임 통과 때만) - 갈음 뒤 안 지은 라이다 필터 지움 `drop_stale_filters` (옛 LAS 산출 · 뒤에 고르면 새 LAS 로 다시 지음) - 뒤 일 `start_job` · 진행 기록 `note_job` · `saved_job`(서버 다시 켜짐 = 「끊김」) @@ -21,6 +22,7 @@ from __future__ import annotations import asyncio import json import logging +import os import shutil from pathlib import Path from typing import Any, Callable @@ -146,6 +148,21 @@ def build_surface( ) +def promote_build(src: Path, dst: Path) -> None: + """`src` 에 지은 라이다 → `dst` — 입력 LAS 갈음 · `processed` · `models` 파일 덮어 옮김 + (이름 같은 것만 · 수치지형도 · 지도 그대로 · 같은 디스크 = 수정시각 살아 지문 그대로).""" + pre_src, pre_dst = src / "B04_PreProcess", dst / "B04_PreProcess" + for sub in ("processed", "models"): + (pre_dst / sub).mkdir(parents=True, exist_ok=True) + for path in sorted((pre_src / sub).iterdir()): + target = pre_dst / sub / path.name + if target.is_dir(): + shutil.rmtree(target) + os.replace(path, target) + shutil.rmtree(pre_dst / "lidar_las", ignore_errors=True) + os.replace(pre_src / "lidar_las", pre_dst / "lidar_las") + + def drop_stale_filters(base: Path, built: list[str]) -> list[str]: """갈음 뒤 — 이번에 안 지은 라이다 필터의 옛 LAS 산출(모델 · 미리보기 · 등고선 · 마스크 · 지면점 · 점 미리보기 · manifest 줄)을 지움 → 지운 모델 파일(`base` 기준 상대경로). diff --git a/resources/tester/test_637_lidar.py b/resources/tester/test_637_lidar.py index fa206d109..8e0d81183 100644 --- a/resources/tester/test_637_lidar.py +++ b/resources/tester/test_637_lidar.py @@ -229,15 +229,15 @@ def test_노선_2m_나눔() -> None: assert len(pts) == 1 + 5 + 2 and np.hypot(*np.diff(pts, axis=0).T).max() <= 2.0 + 1e-9 -def _route_build(tmp_path: Path, monkeypatch, line, target=None) -> dict: # type: ignore[no-untyped-def] +def _route_build(tmp_path: Path, monkeypatch, line, target=None, guard=False, x0=200000.0) -> dict: # type: ignore[no-untyped-def] import B04_PreProcess.B04_PreProcess_Engine as engine from B05_Profile import B05_Profile_Router_Lidar as b05 from B05_Profile import B05_Profile_Router_Replan as replan monkeypatch.setattr(engine, "download_geodata", lambda *a, **k: pytest.fail("지도 받기")) monkeypatch.setattr(replan, "_planned_line", lambda _root: line) - write_las(tmp_path / b05.UPLOAD / "t.las", 200000.0, 500000.0) - return b05.build_route_lidar(tmp_path, EPSG, target, lambda *_: None) + write_las(tmp_path / b05.UPLOAD / "t.las", x0, 500000.0) + return b05.build_route_lidar(tmp_path, EPSG, target, lambda *_: None, guard) def test_B_뿌리_짓기_수치지형도_그대로_덮임(tmp_path: Path, monkeypatch) -> None: # type: ignore[no-untyped-def] @@ -267,6 +267,48 @@ def test_B_뿌리_짓기_수치지형도_그대로_덮임(tmp_path: Path, monkey assert not list((tmp_path / b05.MODELS).glob("*_csf*")) and sheet.read_bytes() == b"sheet" +def _stamps(root: Path) -> dict[str, str]: + """B04_PreProcess 파일 지문(경로 → 크기 · 수정시각 · 내용 앞).""" + pre = root / "B04_PreProcess" + return { + p.relative_to(pre).as_posix(): f"{p.stat().st_size}:{p.stat().st_mtime_ns}" + for p in pre.rglob("*") + if p.is_file() and "lidar_job" not in p.name + } + + +def test_설계_라이다_B_못_덮는_LAS_거절_다_덮으면_갈음(tmp_path: Path, monkeypatch) -> None: # type: ignore[no-untyped-def] + """637-5 — 설계 지형이 라이다인 B: 시험 자리에 지어 덮임 검사 · 못 덮음 = 뿌리 그대로.""" + from B05_Profile import B05_Profile_Router_Lidar as b05 + + line = [[200010, 500010], [200050, 500050]] + target = {"source_filter": "classification", "method": "dtm", "smooth": True} + first = _route_build(tmp_path, monkeypatch, line, target) + assert first["meta"]["covered"] is True + before, meta = _stamps(tmp_path), b05.read_meta(tmp_path) + # 반쪽 LAS(노선 x 200010 ~ 200050 중 200030 부터만) = 거절 · 파일 · 메타 그대로 · 시험 자리 지움 + with pytest.raises( + ValueError, match=r"노선을 \d+ % 만 덮음 — 설계 지형이 라이다라 바꾸지 않음" + ): + _route_build(tmp_path, monkeypatch, line, target, guard=True, x0=200030.0) + assert _stamps(tmp_path) == before and b05.read_meta(tmp_path) == meta + assert not (tmp_path / b05.TRY).exists() and not (tmp_path / b05.UPLOAD).exists() + # 다 덮는 LAS = 갈음 — 본 자리 모델 · 입력 LAS 새것 · 시험 자리 지움 + got = _route_build(tmp_path, monkeypatch, line, target, guard=True, x0=199995.0) + assert got["meta"]["covered"] is True and got["meta"]["selection"] == target + assert not (tmp_path / b05.TRY).exists() + after = _stamps(tmp_path) + assert after["models/dtm_classification.npz"] != before["models/dtm_classification.npz"] + assert after["lidar_las/t.las"] != before["lidar_las/t.las"] + for model in got["models"]: # 행 지문 = 옮긴 파일 지문 + from common_util.common_util_derived_store import file_stamp + + path = tmp_path / model["model_file_path"] + assert path.is_file() + assert model["generation_params"]["file_stamp"] == file_stamp(path) + assert b05.route_coverage(tmp_path, target) == (True, 1.0) + + class _DictCursor(_Cursor): def __init__(self, params: dict) -> None: super().__init__([])