feat(D03): 예정노선 라이다 — LAS 올리기 · B04 엔진 지형 · terrain=lidar (PLAN 624-4 · 요구 530 · 531)
- 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 <noreply@anthropic.com>
This commit is contained in:
Claude Opus 5.5
@@ -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", "결과 저장 중")
|
||||
|
||||
|
||||
@@ -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}
|
||||
@@ -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):
|
||||
|
||||
@@ -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]:
|
||||
|
||||
@@ -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)
|
||||
|
||||
@@ -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 == []
|
||||
Reference in New Issue
Block a user