feat(637-2): B05 라이다 나중 올리기 창구 — POST · GET /route/lidar(뿌리 B04_PreProcess/models 짓기 · 입력 lidar_las · 갈음 · 넣기/고치기만 keep · 수치지형도 B = las_selection 적기 · 노선 2 m 덮임 covered · 저장 잠금) · 지형 바꿈 force(같은 지형 다시 셈) · main 등록 · 시험(요구 546)

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

B05_Profile/B05_Profile_Router_Lidar.py

@@ -0,0 +1,225 @@
+"""B05 라이다 나중 올리기(PLAN 637-2 · 요구 546) — 프로젝트 만든 뒤 LAS 를 올려 지형 칩에 라이다.
+
+- `POST /api/projects/{id}/route/lidar` — LAS/LAZ(+ .prj) multipart `files` → 뒤에서 짓기
+  (저장 · 설계 지형 바꿈 중 = 409 · 이미 돎 = 409)
+- `GET …/route/lidar` — 진행 · 결과 `{ready · running · percent · message · error · interrupted ·
+  covered · covered_ratio · epsg · files · design · sources}`
+
+짓는 자리 = 프로젝트 뿌리 `B04_PreProcess/models`(B05 표본이 읽는 자리) ·
+입력 LAS = `B04_PreProcess/lidar_las/`(B03 원본 · 입력 행은 안 건드림) · 다시 올림 = 갈음.
+길 = 공용 `common_util_lidar_upload`(예정노선과 같음 · `maps=False` = 지도 · GIS · 도엽 서피스
+안 받음 → 수치지형도 모델 · 배수유역 그대로).
+DB = `surface_models` 넣기/고치기만(`keep` — 갈음 때 id · 확정 · 노선 연결 살림 ·
+`save_surface_analysis_to_db` 안 씀).
+라이다 확정값 — 지금 설계가 라이다면 그 필터로 갈음 · 수치지형도면 `las_selection` 에 적음
+(칩 라이다 열림 · 설계 지형은 그대로).
+노선 · 노드 · 초기값 · 종단 · 횡단 안 바뀜 — 설계 값은 사용자가 지형 칩 확인창에서
+`POST /route/design-terrain`(갈음 뒤 같은 지형 = `force`) 할 때만.
+노선 덮임 = 저장 노선을 2 m 마다 새 지형에서 표본 — 빈칸 있으면 `covered` 거짓
+(화면이 라이다 바꿈 막음).
+일 동안 [저장]과 같은 프로젝트 잠금(`common_util_save_guard`).
+
+계산 자리 = ③ 서버 단독.
+"""
+
+from __future__ import annotations
+
+import asyncio
+import json
+from pathlib import Path
+from typing import Any
+from uuid import UUID
+
+import numpy as np
+from fastapi import APIRouter, File, HTTPException, UploadFile
+
+from common_util import common_util_lidar_upload as lidar_upload
+from common_util.common_util_json import atomic_write_json
+from common_util.common_util_surface_confirmation import (
+    LAS_STASH_KEY,
+    analysis_terrain_sources,
+    design_terrain,
+    get_surface_confirmation_params,
+    stash_las_selection,
+)
+from config.config_db import get_db_pool
+
+router = APIRouter(prefix="/api/projects", tags=["B05 Route Lidar"])
+
+_JOBS: dict[tuple[str, str], dict[str, Any]] = {}
+_TASKS: set[asyncio.Task] = set()
+KIND = "lidar"
+PRE = Path("B04_PreProcess")
+MODELS = PRE / "models"
+UPLOAD = PRE / "lidar_upload"
+LAS_DIR = PRE / "lidar_las"
+META = PRE / "lidar_meta.json"
+JOB = PRE / "lidar_job.json"
+STEP_M = 2.0
+_KEYS = ("source_filter", "method", "smooth")
+
+
+def las_target(params: dict[str, Any]) -> dict[str, Any] | None:
+    """이미 있는 라이다 확정값 — 지금 설계(라이다) · 적어 둔 것 · 없으면 None(새로 덧붙임)."""
+    if design_terrain(params) == "las":
+        return {key: params[key] for key in _KEYS}
+    stash = params.get(LAS_STASH_KEY)
+    if isinstance(stash, dict) and stash.get("source_filter"):
+        return {key: stash[key] for key in _KEYS if key in stash}
+    return None
+
+
+def project_epsg(root: Path) -> int:
+    """작업 좌표계 — 설계 노선 · 지형 PRJ 사다리(`resolve_project_crs`) · EPSG 로 못 읽으면 400."""
+    from pyproj import CRS
+
+    from common_util.common_util_crs import identify_epsg, resolve_project_crs
+    from common_util.common_util_route_geometry import load_design_route
+
+    planned = load_design_route(root)
+    text = resolve_project_crs(root, route_crs_input=planned.crs_input if planned else None)
+    epsg = identify_epsg(CRS.from_user_input(text), text)
+    if epsg is None:
+        raise HTTPException(status_code=400, detail="프로젝트 좌표계를 EPSG 로 읽지 못했습니다.")
+    return int(epsg)
+
+
+def line_points(line: list[list[float]], step: float = STEP_M) -> np.ndarray:
+    """꺾은선을 `step` m 마다(꼭짓점 포함) 나눈 점."""
+    xy = np.asarray(line, dtype=np.float64).reshape(-1, 2)
+    out = [xy[:1]]
+    for a, b in zip(xy[:-1], xy[1:]):
+        count = max(1, int(np.ceil(np.hypot(*(b - a)) / step)))
+        t = np.arange(1, count + 1)[:, None] / count
+        out.append(a + (b - a) * t)
+    return np.vstack(out)
+
+
+def route_coverage(root: Path, selection: dict[str, Any]) -> tuple[bool, float]:
+    """저장 노선이 새 라이다 지형에 다 덮이나 — (덮임, 덮인 비율)."""
+    from B05_Profile.B05_Profile_Router_Replan import _planned_line
+    from common_util.common_util_surface_sampler import build_surface_sampler
+
+    line = _planned_line(root)
+    if len(line) < 2:
+        return False, 0.0
+    sampler = build_surface_sampler(
+        root / MODELS,
+        str(selection["source_filter"]),
+        str(selection["method"]),
+        bool(selection["smooth"]),
+    )
+    _, valid = sampler.sample_xy(line_points(line))
+    return bool(valid.all()), round(float(np.mean(valid)), 4)
+
+
+def build_route_lidar(
+    root: 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)
+    method = str((target or {}).get("method") or SURFACE_CONFIRM_DEFAULT_METHOD)
+    result = lidar_upload.build_surface(
+        root,
+        las_paths,
+        report,
+        [str(target["source_filter"])] if target else None,
+        sorted({*SURFACE_AUTO_METHODS, method}),
+    )
+    selection = target or {
+        "source_filter": result["source_filters"][0],
+        "method": method,
+        "smooth": bool(SURFACE_CONFIRM_DEFAULT_SMOOTH),
+    }
+    report(96, "노선 덮임 보는 중")
+    covered, ratio = route_coverage(root, selection)
+    meta = {
+        "epsg": int(epsg),
+        "files": [p.name for p in las_paths],
+        "selection": selection,
+        "point_count": result["processed"]["point_count"],
+        "bounds": result["processed"]["bounds"],
+        "covered": covered,
+        "covered_ratio": ratio,
+    }
+    atomic_write_json(root / META, meta)
+    models = [model for model in result["models"] if model.get("model_file_path")]
+    return {"meta": meta, "models": models}
+
+
+async def _root_and_params(project_id: UUID) -> tuple[Path, dict[str, Any]]:
+    from B05_Profile.B05_Profile_Router_Replan import _project_paths
+
+    paths = await _project_paths(project_id)
+    if paths is None:
+        raise HTTPException(status_code=404, detail="프로젝트 저장 경로를 찾을 수 없습니다.")
+    async with get_db_pool().acquire() as connection:
+        params = await get_surface_confirmation_params(connection, str(project_id))
+    return paths[0], params
+
+
+@router.post("/{project_id}/route/lidar", response_model=None)
+async def upload_route_lidar(project_id: UUID, files: list[UploadFile] = File(...)) -> dict:
+    """B 에 LAS 올리기 → 뒤에서 라이다 지형(진행 = `GET …/route/lidar`)."""
+    from common_util import common_util_save_guard as guard
+
+    lidar_upload.check_names([upload.filename or "" for upload in files])
+    root, params = await _root_and_params(project_id)
+    lock = guard.project_lock(str(project_id))
+    if lock.locked():
+        raise HTTPException(status_code=409, detail="저장 중입니다 — 끝난 뒤 다시 올리세요.")
+    if (_JOBS.get((str(project_id), KIND)) or {}).get("running"):
+        raise HTTPException(status_code=409, detail="이미 돌고 있습니다.")
+    target, design = las_target(params), design_terrain(params)
+    epsg = await asyncio.to_thread(project_epsg, root)
+    await lidar_upload.receive_uploads(files, root / UPLOAD)
+
+    async def work(report: Any) -> None:
+        async with lock:
+            got = await asyncio.to_thread(build_route_lidar, root, epsg, target, report)
+            await lidar_upload.save_models(project_id, got["models"], epsg, keep=True)
+            if design == "sheet":  # 칩 라이다 열림 · 설계 지형은 그대로
+                async with get_db_pool().acquire() as connection:
+                    await connection.begin()
+                    await stash_las_selection(connection, str(project_id), got["meta"]["selection"])
+                    await connection.commit()
+        report(100, f"라이다 지형 끝 — 점 {got['meta']['point_count']:,}개")
+
+    return lidar_upload.start_job(_JOBS, _TASKS, project_id, KIND, work, root / JOB, "B05")
+
+
+def read_meta(root: Path) -> dict[str, Any]:
+    try:
+        return json.loads((root / META).read_text(encoding="utf-8"))
+    except (OSError, ValueError):
+        return {}
+
+
+@router.get("/{project_id}/route/lidar", response_model=None)
+async def route_lidar(project_id: UUID) -> dict[str, Any]:
+    """올린 라이다 — 진행 · 메타(덮임 · 좌표계 · 파일) · 지금 설계 지형 · 고를 수 있는 지형."""
+    root, params = await _root_and_params(project_id)
+    job = _JOBS.get((str(project_id), KIND))
+    job = dict(job) if job else await asyncio.to_thread(lidar_upload.saved_job, root / JOB)
+    meta = await asyncio.to_thread(read_meta, root)
+    picked = meta.get("selection") or {}
+    dem = root / MODELS / f"dtm_{picked.get('source_filter')}.npz"
+    ready = bool(picked) and await asyncio.to_thread(dem.is_file)
+    return {
+        "status": "success",
+        **job,
+        "ready": ready,
+        "covered": bool(meta.get("covered")),
+        "covered_ratio": meta.get("covered_ratio"),
+        "epsg": meta.get("epsg"),
+        "files": meta.get("files", []),
+        "point_count": meta.get("point_count"),
+        "design": design_terrain(params),
+        "sources": analysis_terrain_sources(params, root / MODELS),
+    }

B05_Profile/B05_Profile_Router_TerrainSwitch.py

@@ -5,7 +5,8 @@
 (종단 · 횡단 · 배수 · 관 · 구조물 따라감 ·
 계획고 옮김 없음 = 옛 · 새 계획 폴리라인 같음). 노선 · 노드 · 곡선표 = 지금 저장본 그대로(안 씀).
 초기값(`initial_snapshot/`) · 추천 저장본 · 예정노선 안 씀(옛 추천 [적용] 길 `as_initial` 대신).
-[저장]과 같은 프로젝트 잠금 · 떠 두고 실패면 되돌림(`common_util_save_guard`) · 같은 지형 = 바꿀 것 없음.
+[저장]과 같은 프로젝트 잠금 · 떠 두고 실패면 되돌림(`common_util_save_guard`) ·
+같은 지형 = 바꿀 것 없음(`force` = 같은 지형도 체인 다시 탐 — LAS 갈음 뒤 · PLAN 637-2).
 """
 
 import asyncio
@@ -28,6 +29,8 @@ logger = logging.getLogger(__name__)
 
 class DesignTerrainRequest(BaseModel):
     terrain: Literal["las", "sheet"]
+    #: 같은 지형도 다시 셈 — LAS 갈음 뒤 설계 지형이 라이다인 B(PLAN 637-2)
+    force: bool = False
 
 
 @router.post("/{project_id}/route/design-terrain", response_model=None)
@@ -53,7 +56,7 @@ async def switch_route_design_terrain(
         pool = get_db_pool()
         async with pool.acquire() as connection:
             now = design_terrain(await get_surface_confirmation_params(connection, str(project_id)))
-            if now == body.terrain:
+            if now == body.terrain and not body.force:
                 return {"status": "success", "design": now, "changed": False}
             rows = await guard.snapshot_rows(connection, str(project_id), surface=True)
         files = await asyncio.to_thread(

common_util/common_util_surface_confirmation.py

@@ -185,3 +185,39 @@ async def merge_surface_confirmation_params(
             (json.dumps(params, ensure_ascii=False), project_id),
         )
     return normalized
+
+
+async def stash_las_selection(
+    connection: aiomysql.Connection,
+    project_id: str,
+    selection: dict[str, Any],
+) -> None:
+    """stage 1 params 에 라이다 확정값(`LAS_STASH_KEY`)만 적음 — 설계 지형은 그대로(B05 LAS 나중
+    올리기 · PLAN 637-2 · 지형 칩 라이다 열림). 트랜잭션은 부르는 쪽."""
+    async with connection.cursor(aiomysql.DictCursor) as cursor:
+        await cursor.execute(
+            """
+            SELECT params
+            FROM project_workflow_stages
+            WHERE project_id = %s AND stage_no = 1
+            FOR UPDATE
+            """,
+            (project_id,),
+        )
+        row = await cursor.fetchone()
+        if row is None:
+            raise LookupError("WF1 단계 상태를 찾을 수 없습니다.")
+        params = _decode_params(row.get("params"))
+        params[LAS_STASH_KEY] = {
+            "source_filter": str(selection["source_filter"]),
+            "method": str(selection["method"]),
+            "smooth": bool(selection["smooth"]),
+        }
+        await cursor.execute(
+            """
+            UPDATE project_workflow_stages
+            SET params = %s
+            WHERE project_id = %s AND stage_no = 1
+            """,
+            (json.dumps(params, ensure_ascii=False), project_id),
+        )

main.py

@@ -46,6 +46,7 @@ from B04_PreProcess.B04_PreProcess_Router_Inflow import router as b04_inflow_rou
 from B04_PreProcess.B04_PreProcess_Router_Watershed import router as b04_watershed_router
 from B05_Profile.B05_Profile_Router import router as b05_route_router
 from B05_Profile.B05_Profile_Router_Corridor import router as b05_corridor_router
+from B05_Profile.B05_Profile_Router_Lidar import router as b05_route_lidar_router
 from B05_Profile.B05_Profile_Router_Lifecycle import router as b05_route_lifecycle_router
 from B05_Profile.B05_Profile_Router_Replan import router as b05_route_replan_router
 from B05_Profile.B05_Profile_Router_Terrain import router as b05_route_terrain_router
@@ -512,6 +513,7 @@ app.include_router(e06_stab_rock_router, dependencies=protected_with_company)
 app.include_router(b05_corridor_router, dependencies=protected_with_company)
 app.include_router(b05_route_replan_router, dependencies=protected_with_company)
 app.include_router(b05_route_terrain_router, dependencies=protected_with_company)
+app.include_router(b05_route_lidar_router, dependencies=protected_with_company)
 app.include_router(b05_structures_router, dependencies=protected_with_company)
 app.include_router(b06_section_router, dependencies=protected_with_company)
 app.include_router(b06_section_stations_router, dependencies=protected_with_company)

resources/tester/test_637_lidar.py

@@ -155,3 +155,107 @@ def test_DB_넣기_고치기만_갈음은_id_확정_살림(monkeypatch) -> None:
     assert calls[0]["have"] == (11, "new")
     assert (calls[0]["source_file_id"], calls[0]["processed_cloud_id"]) == (1320, 7)
     assert confirmed == [11]
+
+
+# ── 637-2 B05 창구 ──────────────────────────────────────────────────────────
+
+
+def test_라이다_확정값_고르기() -> None:
+    from B05_Profile import B05_Profile_Router_Lidar as b05
+
+    las = {"source_filter": "classification", "method": "dtm", "smooth": True}
+    assert b05.las_target(las) == las  # 설계 = 라이다 → 그 필터로 갈음
+    sheet = {"source_filter": "sheet_laplace", "method": "dtm", "smooth": True}
+    assert b05.las_target(sheet) is None  # 수치지형도만 = 새로 덧붙임(자동 필터)
+    assert b05.las_target({**sheet, "las_selection": las}) == las  # 적어 둔 것
+
+
+def test_노선_2m_나눔() -> None:
+    from B05_Profile import B05_Profile_Router_Lidar as b05
+
+    pts = b05.line_points([[0, 0], [10, 0], [10, 3]])
+    assert pts[0].tolist() == [0, 0] and pts[-1].tolist() == [10, 3]
+    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]
+    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)
+
+
+def test_B_뿌리_짓기_수치지형도_그대로_덮임(tmp_path: Path, monkeypatch) -> None:  # type: ignore[no-untyped-def]
+    from B05_Profile import B05_Profile_Router_Lidar as b05
+
+    sheet = tmp_path / b05.MODELS / "dtm_sheet_laplace.npz"
+    sheet.parent.mkdir(parents=True)
+    sheet.write_bytes(b"sheet")
+    got = _route_build(tmp_path, monkeypatch, [[200010, 500010], [200050, 500050]])
+    meta = got["meta"]
+    assert sheet.read_bytes() == b"sheet"  # 수치지형도 모델 파일 그대로
+    assert (tmp_path / b05.MODELS / f"dtm_{meta['selection']['source_filter']}.npz").is_file()
+    assert meta["covered"] is True and meta["covered_ratio"] == 1.0
+    assert meta["files"] == ["t.las"] and not (tmp_path / b05.UPLOAD).exists()
+    assert all(m["model_file_path"].startswith("B04_PreProcess/models/") for m in got["models"])
+    assert b05.read_meta(tmp_path)["selection"] == meta["selection"]
+    # 노선 반이 라이다 밖 = 덮임 거짓
+    again = _route_build(tmp_path, monkeypatch, [[200010, 500030], [200110, 500030]])["meta"]
+    assert again["covered"] is False and 0.3 < again["covered_ratio"] < 0.7
+    # 있던 라이다 확정값 = 그 필터로 짓기
+    target = {"source_filter": "classification", "method": "dtm", "smooth": True}
+    kept = _route_build(tmp_path, monkeypatch, [[200010, 500010], [200020, 500020]], target)
+    assert kept["meta"]["selection"] == target
+    assert {m["source_filter"] for m in kept["models"]} == {"classification"}
+
+
+class _DictCursor(_Cursor):
+    def __init__(self, params: dict) -> None:
+        super().__init__([])
+        self.params, self.sql = params, []
+
+    async def execute(self, sql: str, args: tuple = ()) -> None:
+        self.sql.append((sql, args))
+
+    async def fetchone(self) -> dict:
+        import json
+
+        return {"params": json.dumps(self.params)}
+
+
+def test_las_selection_적기_칩_열림(tmp_path: Path) -> None:
+    import asyncio
+    import json
+
+    from common_util import common_util_surface_confirmation as confirm
+
+    params = {"source_filter": "sheet_laplace", "method": "dtm", "smooth": True}
+    assert confirm.analysis_terrain_sources(params, tmp_path) == ["sheet"]  # 칩 라이다 막힘
+    cursor = _DictCursor(params)
+    connection = type("C", (), {"cursor": lambda self, *_: cursor})()
+    pick = {"source_filter": "csf", "method": "dtm", "smooth": True}
+    asyncio.run(confirm.stash_las_selection(connection, str(PID), pick))
+    written = json.loads(cursor.sql[-1][1][0])
+    assert written["las_selection"] == pick
+    assert written["source_filter"] == "sheet_laplace"  # 설계 지형 그대로
+    assert confirm.analysis_terrain_sources(written, tmp_path) == ["las", "sheet"]
+    assert confirm.analysis_terrain_params(written, "las")["source_filter"] == "csf"
+
+
+def test_지형_바꿈_force_같은_지형도_다시_셈(monkeypatch) -> None:  # type: ignore[no-untyped-def]
+    import test_626_1_recommend_preview as base
+
+    from B05_Profile import B05_Profile_Router_TerrainSwitch as switch
+
+    request = switch.DesignTerrainRequest
+    assert request(terrain="las").force is False
+    monkeypatch.setattr(
+        switch, "DesignTerrainRequest", lambda terrain: request(terrain=terrain, force=True)
+    )
+    result, calls, line = base._run(monkeypatch, "las", "las")
+    assert result["changed"] is True and calls["args"][4] == "las"
+    assert calls["args"][5] == (line, line)  # 노선 그대로