From de8f2fee4a41d625c77b0d79eb0fe0e20f60613d Mon Sep 17 00:00:00 2001 From: umsangdon Date: Sat, 31 Oct 2026 20:33:47 +0900 Subject: [PATCH] =?UTF-8?q?feat(D03):=20=EC=98=88=EC=A0=95=EB=85=B8?= =?UTF-8?q?=EC=84=A0=20=EB=9D=BC=EC=9D=B4=EB=8B=A4=20=E2=80=94=20LAS=20?= =?UTF-8?q?=EC=98=AC=EB=A6=AC=EA=B8=B0=20=C2=B7=20B04=20=EC=97=94=EC=A7=84?= =?UTF-8?q?=20=EC=A7=80=ED=98=95=20=C2=B7=20terrain=3Dlidar=20(PLAN=20624-?= =?UTF-8?q?4=20=C2=B7=20=EC=9A=94=EA=B5=AC=20530=20=C2=B7=20531)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - POST /plan/lidar(LAS · LAZ + .prj · 예정노선만 · 다시 올리면 갈음) → 뒤에서 지형(진행 = /plan/progress lidar) · GET /plan/lidar - 자리 = D 의 B04_PreProcess/lidar/(안은 B 와 같은 꼴) · LAS 좌표계 다르면 x · y 옮겨 적음 · DTM = surface_models 행(높이 격자) - terrain=lidar 급경사 · 음영 · 0차곡 = 올린 DTM · 범위 밖 = 빈칸(수치지형도로 안 메움 · 결정 A⑥) - B04 run_surface_analysis 에 maps 인자 덧(기본 True = B 그대로 · False = 지도 · GIS 받기 · 도엽 서피스 건너뜀) - 시험 test_624_4_plan_lidar 4 Co-Authored-By: Claude Opus 5.5 --- B04_PreProcess/B04_PreProcess_Engine.py | 40 ++-- D03_PlanAnalysis/D03_PlanAnalysis_Lidar.py | 194 ++++++++++++++++++ D03_PlanAnalysis/D03_PlanAnalysis_Router.py | 2 +- .../D03_PlanAnalysis_TerrainSource.py | 11 +- main.py | 2 + resources/tester/test_624_4_plan_lidar.py | 128 ++++++++++++ 6 files changed, 356 insertions(+), 21 deletions(-) create mode 100644 D03_PlanAnalysis/D03_PlanAnalysis_Lidar.py create mode 100644 resources/tester/test_624_4_plan_lidar.py diff --git a/B04_PreProcess/B04_PreProcess_Engine.py b/B04_PreProcess/B04_PreProcess_Engine.py index 6e18bfc88..71f1c0583 100644 --- a/B04_PreProcess/B04_PreProcess_Engine.py +++ b/B04_PreProcess/B04_PreProcess_Engine.py @@ -197,9 +197,12 @@ def run_surface_analysis( methods: list[str], force: bool = False, on_progress: ProgressCallback | None = None, + maps: bool = True, ) -> dict[str, Any]: """구조화→필터→모델 빌드를 수행하고 산출 메타데이터를 반환한다. + `maps=False` = 지도 · GIS 받기와 도엽 서피스를 건너뜀(D 라이다 · PLAN 624-4). + `las_path`는 지형 파일 한 장 또는 여러 장이다 — 여러 장이면 합친 범위로 한 벌을 만든다(2026-09-06 사용자 확정). @@ -349,31 +352,38 @@ def run_surface_analysis( ) # 3-2·3-3. VWorld 지도·국가 GIS 벡터·수치지형도 도엽 (공용 블록 — 도엽 서피스도 사용) - _report(90, "download_maps", "VWorld 지도 및 GIS 벡터 데이터 다운로드 중") las_bounds_dict = { "x": [float(bounds[0, 0]), float(bounds[0, 1])], "y": [float(bounds[1, 0]), float(bounds[1, 1])], "z": [float(bounds[2, 0]), float(bounds[2, 1])], } - download_geodata( - project_root, processed_dir, las_bounds_dict, las_paths[0].parent, rebuild, report=_report - ) + if maps: + _report(90, "download_maps", "VWorld 지도 및 GIS 벡터 데이터 다운로드 중") + download_geodata( + project_root, + processed_dir, + las_bounds_dict, + las_paths[0].parent, + rebuild, + report=_report, + ) # 3-4. 도엽등고선 3D 서피스 — LAS가 있어도 참고용으로 같이 만들어 영구저장한다 # (2026-08-30 사용자 확정). 실패해도 분석은 계속한다. - _report(94, "surface_model", "도엽등고선 3D 서피스 생성 중") sheet_models: list[dict[str, Any]] = [] - try: - from B04_PreProcess.B04_PreProcess_Engine_SheetSurface import ( - build_sheet_surface_from_route, - ) + if maps: + _report(94, "surface_model", "도엽등고선 3D 서피스 생성 중") + try: + from B04_PreProcess.B04_PreProcess_Engine_SheetSurface import ( + build_sheet_surface_from_route, + ) - # 자동 전처리는 기본 방식 하나만 만든다 — 나머지는 관리자가 화면에서 고를 때. - sheet_models = build_sheet_surface_from_route( - project_root, processed_dir, models_dir, list(SHEET_SURFACE_AUTO_METHODS) - ) - except Exception as exc: - logger.warning("도엽등고선 서피스 생성 실패: %s", exc) + # 자동 전처리는 기본 방식 하나만 만든다 — 나머지는 관리자가 화면에서 고를 때. + sheet_models = build_sheet_surface_from_route( + project_root, processed_dir, models_dir, list(SHEET_SURFACE_AUTO_METHODS) + ) + except Exception as exc: + logger.warning("도엽등고선 서피스 생성 실패: %s", exc) _report(95, "saving", "결과 저장 중") diff --git a/D03_PlanAnalysis/D03_PlanAnalysis_Lidar.py b/D03_PlanAnalysis/D03_PlanAnalysis_Lidar.py new file mode 100644 index 000000000..c50f3d8c3 --- /dev/null +++ b/D03_PlanAnalysis/D03_PlanAnalysis_Lidar.py @@ -0,0 +1,194 @@ +"""D 예정노선 라이다 — LAS 올리기 → B04 엔진 그대로 지형(요구 530 · 531 · PLAN 624-4). + +- `POST /api/projects/{id}/plan/lidar` — LAS(+ .prj) 올리기 · 뒤에서 지형 짓기 + (진행 = `/plan/progress` 의 `lidar`) · 다시 올리면 옛 LAS 를 갈음 +- `GET …/plan/lidar` — 지은 라이다 지형(없으면 `ready: false`) + +자리 = `B04_PreProcess/lidar/`(D 몫 · 안은 B 와 같은 꼴 — `B03_FileInput/input/las/` · +`B04_PreProcess/models/`). +LAS 좌표계(.prj · LAS 머리 · 없으면 프로젝트 좌표계)가 프로젝트와 다르면 옮겨 적음(높이 그대로). +지도 · GIS 받기 · 도엽 서피스는 안 함(`maps=False`) · 지은 DTM = `surface_models` 행 +(높이 표본 격자). +라이다 범위 밖 = 빈칸(격자 `valid_mask`) · 수치지형도로 메우지 않음(결정 A⑥). + +계산 자리 = ③ 서버 단독. +""" + +from __future__ import annotations + +import asyncio +import json +import shutil +from pathlib import Path +from typing import Any, Callable +from uuid import UUID + +from fastapi import APIRouter, File, HTTPException, UploadFile + +from common_util.common_util_json import atomic_write_json + +router = APIRouter(prefix="/api/projects", tags=["D03 Plan Lidar"]) + +LIDAR = Path("B04_PreProcess") / "lidar" # 표면 모델 행 경로 = B04_PreProcess 아래 +UPLOAD = "upload" +LAS_DIR = Path("B03_FileInput") / "input" / "las" +MODELS = Path("B04_PreProcess") / "models" +META = "lidar_meta.json" +LAS_SUFFIXES = (".las", ".laz") +CHUNK = 2_000_000 + + +def lidar_root(root: Path) -> Path: + return Path(root) / LIDAR + + +def read_meta(root: Path) -> dict[str, Any]: + try: + return json.loads((lidar_root(root) / META).read_text(encoding="utf-8")) + except (OSError, ValueError): + return {} + + +def lidar_dem(root: Path) -> Path | None: + """지은 라이다 DTM(평활 안 함) — 없으면 None.""" + name = read_meta(root).get("model_file") + path = Path(root) / name if name else None + return path if path is not None and path.is_file() else None + + +def check_names(names: list[str]) -> None: + """LAS · LAZ 하나 이상 · 그 밖은 .prj 만 · 같은 이름 없음.""" + suffixes = [Path(name).suffix.lower() for name in names] + if not any(s in LAS_SUFFIXES for s in suffixes): + raise HTTPException(status_code=400, detail="LAS 또는 LAZ 파일을 하나 이상 올리세요.") + if any(s not in (*LAS_SUFFIXES, ".prj") for s in suffixes): + raise HTTPException(status_code=400, detail="LAS · LAZ · PRJ 만 올릴 수 있습니다.") + if len({Path(name).name.casefold() for name in names}) != len(names): + raise HTTPException(status_code=400, detail="같은 이름 파일이 있습니다.") + + +def las_epsg(path: Path, prj_text: str | None) -> int | None: + """LAS 좌표계 — .prj 먼저 · 없으면 LAS 머리 · 모르면 None.""" + import laspy + from pyproj import CRS + + from common_util.common_util_crs import identify_epsg + + if prj_text: + return identify_epsg(CRS.from_wkt(prj_text), prj_text) + with laspy.open(path) as reader: + crs = reader.header.parse_crs() + return identify_epsg(crs, crs.to_wkt()) if crs is not None else None + + +def place_las(src: Path, dst_dir: Path, src_epsg: int | None, epsg: int) -> Path: + """올린 LAS → 지형 입력 자리 — 좌표계가 같거나 모르면 옮김 · 다르면 x · y 만 옮겨 적음(.las).""" + import laspy + from pyproj import CRS, Transformer + + dst_dir.mkdir(parents=True, exist_ok=True) + if src_epsg in (None, int(epsg)): + return Path(shutil.move(src, dst_dir / src.name)) + dst = dst_dir / f"{src.stem}.las" + move = Transformer.from_crs(src_epsg, int(epsg), always_xy=True) + with laspy.open(src) as reader: + old = reader.header + header = laspy.LasHeader(point_format=old.point_format, version=old.version) + header.scales = old.scales + x0, y0 = move.transform(old.mins[0], old.mins[1]) + header.offsets = [round(x0), round(y0), old.offsets[2]] + header.add_crs(CRS.from_epsg(int(epsg))) + with laspy.open(dst, mode="w", header=header) as writer: + for points in reader.chunk_iterator(CHUNK): + x, y = move.transform(points.x, points.y) + record = laspy.ScaleAwarePointRecord.zeros(len(points), header=header) + record.array[:] = points.array + record.x, record.y, record.z = x, y, points.z + writer.write_points(record) + src.unlink() + return dst + + +def build_lidar(root: Path, epsg: int, report: Callable[[int, str], None]) -> dict[str, Any]: + """올린 자리(`upload/`)의 LAS → 입력 자리 갈음 → B04 엔진(지면 필터 · DTM) → 메타.""" + from B04_PreProcess.B04_PreProcess_Engine import run_surface_analysis + from config.config_system import SURFACE_AUTO_METHODS + + base = lidar_root(root) + upload, las_dir = base / UPLOAD, base / LAS_DIR + prj = next(iter(sorted(upload.glob("*.prj"))), None) + prj_text = prj.read_text(encoding="utf-8", errors="ignore") if prj else None + sources = sorted(p for p in upload.iterdir() if p.suffix.lower() in LAS_SUFFIXES) + if sources: # 새로 올림 = 옛 LAS 갈음 + shutil.rmtree(las_dir, ignore_errors=True) + report(5, "LAS 좌표계 맞추는 중") + for path in sources: + place_las(path, las_dir, las_epsg(path, prj_text), epsg) + shutil.rmtree(upload, ignore_errors=True) + las_paths = sorted(p for p in las_dir.glob("*") if p.suffix.lower() in LAS_SUFFIXES) + if not las_paths: + raise FileNotFoundError("올린 LAS 가 없습니다.") + result = run_surface_analysis( + base, + las_paths, + source_filters=None, + methods=list(SURFACE_AUTO_METHODS), + on_progress=lambda percent, _stage, message: report(10 + int(percent * 0.85), message), + maps=False, + ) + prefix = LIDAR.as_posix() + models = [ + {**model, "model_file_path": f"{prefix}/{model['model_file_path']}"} + for model in result["models"] + if model.get("model_file_path") + ] + picked = result["source_filters"][0] + meta = { + "epsg": int(epsg), + "files": [p.name for p in las_paths], + "source_filter": picked, + "model_file": f"{prefix}/{(MODELS / f'dtm_{picked}.npz').as_posix()}", + "point_count": result["processed"]["point_count"], + "bounds": result["processed"]["bounds"], + } + atomic_write_json(base / META, meta) + return {"meta": meta, "models": models} + + +@router.post("/{project_id}/plan/lidar", response_model=None) +async def upload_plan_lidar(project_id: UUID, files: list[UploadFile] = File(...)) -> dict: + """D 에 LAS 올리기 → 뒤에서 라이다 지형(진행 = `/plan/progress` 의 `lidar`).""" + from B03_FileInput.B03_FileInput_Engine import save_upload_stream + from D03_PlanAnalysis import D03_PlanAnalysis_Router as base + + check_names([upload.filename or "" for upload in files]) + root, epsg, _, _ = await base._plan_ends(project_id) # 예정노선 아님 400 + if (base._JOBS.get((str(project_id), "lidar")) or {}).get("running"): + raise HTTPException(status_code=409, detail="이미 돌고 있습니다.") + upload = lidar_root(root) / UPLOAD + shutil.rmtree(upload, ignore_errors=True) + upload.mkdir(parents=True) + try: + for item in files: + await save_upload_stream(item, upload / Path(item.filename or "").name) + except ValueError as error: + shutil.rmtree(upload, ignore_errors=True) + raise HTTPException(status_code=413, detail=str(error)) from error + + async def work(report: Any) -> None: + got = await asyncio.to_thread(build_lidar, root, epsg, report) + await base._save_models(project_id, got["models"], epsg) + report(100, f"라이다 지형 끝 — 점 {got['meta']['point_count']:,}개") + + return base._start_job(project_id, "lidar", work, root) + + +@router.get("/{project_id}/plan/lidar", response_model=None) +async def plan_lidar(project_id: UUID) -> dict[str, Any]: + """지은 라이다 지형 — `ready` · 메타(좌표계 · 파일 · 필터 · 점 수 · 범위).""" + from D03_PlanAnalysis import D03_PlanAnalysis_Store as store + + root, _ = await store.load_plan(project_id) + meta = await asyncio.to_thread(read_meta, root) + ready = await asyncio.to_thread(lidar_dem, root) is not None + return {"status": "success", "ready": ready, **meta} diff --git a/D03_PlanAnalysis/D03_PlanAnalysis_Router.py b/D03_PlanAnalysis/D03_PlanAnalysis_Router.py index 5dde3db34..676edafa5 100644 --- a/D03_PlanAnalysis/D03_PlanAnalysis_Router.py +++ b/D03_PlanAnalysis/D03_PlanAnalysis_Router.py @@ -57,7 +57,7 @@ router = APIRouter(prefix="/api/projects", tags=["D03 Plan Analysis"]) # 진행 — (프로젝트, 일) → {running, percent, message, error} · 프로세스 메모리 _JOBS: dict[tuple[str, str], dict[str, Any]] = {} _TASKS: set[asyncio.Task] = set() -KINDS = ("terrain", "analyze") +KINDS = ("terrain", "analyze", "lidar") class Center(BaseModel): diff --git a/D03_PlanAnalysis/D03_PlanAnalysis_TerrainSource.py b/D03_PlanAnalysis/D03_PlanAnalysis_TerrainSource.py index 7481ba5f7..079100b32 100644 --- a/D03_PlanAnalysis/D03_PlanAnalysis_TerrainSource.py +++ b/D03_PlanAnalysis/D03_PlanAnalysis_TerrainSource.py @@ -3,7 +3,8 @@ 겹 길(`/plan/layers/*`)은 프로젝트 종류를 안 가림: - 예정노선(PLAN) = 받은 범위 · 지형 = 수치지형도(`models_plan/dtm_sheet_<방법>.npz`) · - 라이다 = D 에 LAS 올리기(624-4) 전이라 없음(404) + `lidar` = D 에 올린 LAS 로 지은 DTM(624-4 · `D03_PlanAnalysis_Lidar` · 안 올렸으면 404 · + 범위 밖 = 빈칸) - 그 밖(임도 등) = 범위 = B04 배경 지도 상자(`vworld_white_meta.json` · 없으면 원천 격자 상자) · 지형 = 설계 지형(stage 1 params · 기본) · `lidar` = 라이다 확정 지표면 · `sheet` = 수치지형도 서피스 · @@ -29,6 +30,7 @@ from common_util.common_util_storage import resolve_stored_project_path from config.config_system_terrain import SHEET_SURFACE_DEFAULT_METHOD from D03_PlanAnalysis import D03_PlanAnalysis_Landslide as landslide from D03_PlanAnalysis import D03_PlanAnalysis_Layers as layers +from D03_PlanAnalysis import D03_PlanAnalysis_Lidar as lidar from D03_PlanAnalysis import D03_PlanAnalysis_Store as store from D03_PlanAnalysis import D03_PlanAnalysis_Terrain as terrain from D03_PlanAnalysis import D03_PlanAnalysis_ZeroOrder as zero_order @@ -80,13 +82,12 @@ def layer_store(root: Path, plan: bool) -> Path: def plan_frame(root: Path, epsg: int, box: Box, choice: str | None) -> Frame: - """예정노선 — 수치지형도 하나(design = sheet) · lidar = 없음.""" + """예정노선 — 수치지형도 하나(design = sheet) · lidar = 올린 LAS 의 DTM(없으면 None).""" source = check_source(choice) source = "sheet" if source == "design" else source npz = terrain.models_dir_of(Path(root)) / f"dtm_sheet_{SHEET_SURFACE_DEFAULT_METHOD}.npz" - return Frame( - Path(root), epsg, box, layer_store(root, True), source, npz if source == "sheet" else None - ) + dem = npz if source == "sheet" else lidar.lidar_dem(root) + return Frame(Path(root), epsg, box, layer_store(root, True), source, dem) def road_dem(models: Path, params: dict[str, Any], source: str) -> tuple[str, Path | None]: diff --git a/main.py b/main.py index 3b05c2b2a..6fee9a373 100644 --- a/main.py +++ b/main.py @@ -75,6 +75,7 @@ from B08_Quantity.B08_Quantity_Router_Milestone import router as b08_milestone_r from B09_Estimation.B09_Estimation_Router import router as b09_estimation_router from D01_PlanRegister.D01_PlanRegister_Router import dashboard_router as d01_dashboard_router from D01_PlanRegister.D01_PlanRegister_Router import router as d01_plan_register_router +from D03_PlanAnalysis.D03_PlanAnalysis_Lidar import router as d03_plan_lidar_router from D03_PlanAnalysis.D03_PlanAnalysis_Network import router as d03_plan_network_router from D03_PlanAnalysis.D03_PlanAnalysis_Router import router as d03_plan_analysis_router from D03_PlanAnalysis.D03_PlanAnalysis_Router_Tiles import router as d03_plan_tiles_router @@ -497,6 +498,7 @@ app.include_router(b05_route_lifecycle_router, dependencies=protected_with_compa app.include_router(d03_plan_analysis_router, dependencies=protected_with_company) app.include_router(d03_plan_tiles_router, dependencies=protected_with_company) app.include_router(d03_plan_network_router, dependencies=protected_with_company) +app.include_router(d03_plan_lidar_router, dependencies=protected_with_company) app.include_router(d04_plan_confirm_router, dependencies=protected_with_company) app.include_router(d04_plan_link_router, dependencies=protected_with_company) app.include_router(e02_stab_ground_router, dependencies=protected_with_company) diff --git a/resources/tester/test_624_4_plan_lidar.py b/resources/tester/test_624_4_plan_lidar.py new file mode 100644 index 000000000..fafb3e3dd --- /dev/null +++ b/resources/tester/test_624_4_plan_lidar.py @@ -0,0 +1,128 @@ +"""PLAN 624-4 — D 예정노선 라이다: LAS 올리기 → B04 엔진 지형 → `terrain=lidar` 겹(요구 530 · 531). + +D 몫 길 = `D03_PlanAnalysis_Lidar` · B04 엔진은 `maps=False` 인자만 덧(기본 = B 그대로). +""" + +from __future__ import annotations + +from pathlib import Path + +import numpy as np +import pytest +from fastapi import HTTPException + +from D03_PlanAnalysis import D03_PlanAnalysis_Layers as layers +from D03_PlanAnalysis import D03_PlanAnalysis_Lidar as lidar +from D03_PlanAnalysis import D03_PlanAnalysis_TerrainSource as source + +D_EPSG, LAS_EPSG = 5187, 5176 # D = GRS80 동부 · 업체 LAS = 베셀 동부(검증 용화와 같음) + + +def _write_las(path: Path, epsg: int | None, x0: float, y0: float) -> None: + """두 덩이(가운데 빈 땅) 지면점 LAS — 비탈 0.3.""" + import laspy + from pyproj import CRS + + rng = np.random.default_rng(7) + xs, ys = [], [] + for dx in (0.0, 160.0): + xs.append(rng.uniform(x0 + dx, x0 + dx + 60, 6000)) + ys.append(rng.uniform(y0, y0 + 60, 6000)) + x, y = np.concatenate(xs), np.concatenate(ys) + header = laspy.LasHeader(point_format=3, version="1.2") + header.offsets = [round(x0), round(y0), 0] + header.scales = [0.01, 0.01, 0.01] + if epsg is not None: + header.add_crs(CRS.from_epsg(epsg)) + data = laspy.LasData(header) + data.x, data.y, data.z = x, y, 500 + 0.3 * (x - x0) + data.classification = np.full(len(x), 2, np.uint8) + path.parent.mkdir(parents=True, exist_ok=True) + data.write(path) + + +def _built(tmp_path: Path, monkeypatch, epsg: int | None = LAS_EPSG) -> dict: # type: ignore[no-untyped-def] + import B04_PreProcess.B04_PreProcess_Engine as engine + + called: list[int] = [] + monkeypatch.setattr(engine, "download_geodata", lambda *a, **k: called.append(1)) + _write_las(lidar.lidar_root(tmp_path) / lidar.UPLOAD / "t.las", epsg, 208300.0, 367500.0) + got = lidar.build_lidar(tmp_path, D_EPSG, lambda *_: None) + assert called == [] # 지도 · GIS 받기 안 함 + return got + + +def test_이름_고르기() -> None: + lidar.check_names(["a.las", "a.prj", "b.LAZ"]) + for names in (["a.prj"], ["a.las", "a.txt"], ["a.las", "A.las"]): + with pytest.raises(HTTPException) as caught: + lidar.check_names(names) + assert caught.value.status_code == 400 + + +def test_올리기_뒤_라이다_지형_B_꼴(tmp_path: Path, monkeypatch) -> None: # type: ignore[no-untyped-def] + got = _built(tmp_path, monkeypatch) + meta = got["meta"] + dem = lidar.lidar_dem(tmp_path) + assert dem is not None and dem.name == f"dtm_{meta['source_filter']}.npz" + assert dem.parent == lidar.lidar_root(tmp_path) / "B04_PreProcess" / "models" + assert meta["files"] == ["t.las"] and meta["point_count"] == 12000 + assert not (lidar.lidar_root(tmp_path) / lidar.UPLOAD).exists() + # surface_models 행 경로 = 프로젝트 뿌리 기준 + from B04_PreProcess.B04_PreProcess_Repository import _validate_stage_path + + assert any(m["model_file_path"] == meta["model_file"] for m in got["models"]) + assert all(_validate_stage_path(m["model_file_path"]) for m in got["models"]) + # 좌표계 옮김 — 베셀 → GRS80 북 +10 만 m 언저리 + from pyproj import Transformer + + want = Transformer.from_crs(LAS_EPSG, D_EPSG, always_xy=True).transform(208300.0, 367500.0) + assert abs(meta["bounds"]["x_min"] - want[0]) < 1.0 + assert abs(meta["bounds"]["y_min"] - want[1]) < 1.0 + # 다시 올리면 갈음 + _write_las(lidar.lidar_root(tmp_path) / lidar.UPLOAD / "u.las", None, 208300.0, 467500.0) + again = lidar.build_lidar(tmp_path, D_EPSG, lambda *_: None)["meta"] + assert again["files"] == ["u.las"] and abs(again["bounds"]["y_min"] - 467500.0) < 1.0 + + +def test_terrain_lidar_급경사_0차곡_범위_밖_빈칸(tmp_path: Path, monkeypatch) -> None: # type: ignore[no-untyped-def] + with pytest.raises(HTTPException) as caught: # 안 올림 = 404 + source.need_dem(source.plan_frame(tmp_path, D_EPSG, (0, 0, 1, 1), "lidar")) + assert caught.value.status_code == 404 + _built(tmp_path, monkeypatch) + frame = source.plan_frame(tmp_path, D_EPSG, (0, 0, 1, 1), "lidar") + npz = source.need_dem(frame) + assert frame.source == "lidar" and npz == lidar.lidar_dem(tmp_path) + value = layers.slope_layer(tmp_path, npz.parent, D_EPSG, "slope", npz) + counts = value["counts"] + assert sum(counts) > 0 and max(range(len(counts)), key=counts.__getitem__) in (29, 30) + # 두 덩이 사이 = 빈칸(수치지형도로 안 메움) + with np.load(npz) as dem: + valid = np.asarray(dem["valid_mask"], bool) + x = np.asarray(dem["x"], float) + x0 = float(x[0]) + gap = (x > x0 + 90) & (x < x0 + 130) + assert gap.any() and not valid[:, gap].any() + work, light, full = source.zero_order_calls(frame, 50.0, 5000.0) + assert work(*light) is None # 저장본 없음 + value = work(*full) + assert value["cell_m"] > 0 and work(*light) == value # 다음 = 저장본 + + +def test_B04_기본은_지도_받기_그대로(tmp_path: Path, monkeypatch) -> None: # type: ignore[no-untyped-def] + """B 프로젝트 전처리(기본 인자) = 지도 · GIS 받기 · 도엽 서피스 그대로 부름.""" + import B04_PreProcess.B04_PreProcess_Engine as engine + import B04_PreProcess.B04_PreProcess_Engine_SheetSurface as sheet + + calls: list[str] = [] + monkeypatch.setattr(engine, "download_geodata", lambda *a, **k: calls.append("maps")) + monkeypatch.setattr( + sheet, "build_sheet_surface_from_route", lambda *a, **k: calls.append("sheet") or [] + ) + las = tmp_path / "B03_FileInput" / "input" / "las" / "t.las" + _write_las(las, None, 200000.0, 500000.0) + engine.run_surface_analysis(tmp_path, las, source_filters=None, methods=["dtm"]) + assert calls == ["maps", "sheet"] + calls.clear() + engine.run_surface_analysis(tmp_path, las, source_filters=None, methods=["dtm"], maps=False) + assert calls == []