Merge remote-tracking branch 'origin/sub_laptop_5' into main_laptop_1

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"]
             )

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` 기준 상대경로).

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__([])