From f6708c6818b5027243762d75522ddf804b82bbb2 Mon Sep 17 00:00:00 2001 From: umsangdon Date: Sun, 1 Nov 2026 11:28:57 +0900 Subject: [PATCH] =?UTF-8?q?refactor(637-1):=20=EC=98=88=EC=A0=95=EB=85=B8?= =?UTF-8?q?=EC=84=A0=20LAS=20=EC=98=AC=EB=A6=AC=EA=B8=B0=20=C2=B7=20?= =?UTF-8?q?=EC=A7=93=EA=B8=B0=20=EA=B3=B5=EC=9A=A9=20=EB=AA=A8=EB=93=88=20?= =?UTF-8?q?common=5Futil=5Flidar=5Fupload=20=E2=80=94=20=EC=9D=B4=EB=A6=84?= =?UTF-8?q?=20=EA=B2=80=EC=82=AC=20=C2=B7=20=EC=A2=8C=ED=91=9C=EA=B3=84=20?= =?UTF-8?q?=C2=B7=20=EC=98=AE=EA=B2=A8=20=EB=86=93=EA=B8=B0=20=C2=B7=20?= =?UTF-8?q?=EB=B0=9B=EA=B8=B0=20=C2=B7=20=EA=B0=88=EC=9D=8C=20=C2=B7=20?= =?UTF-8?q?=EC=A7=93=EA=B8=B0(maps=3DFalse)=20=C2=B7=20=EB=92=A4=20?= =?UTF-8?q?=EC=9D=BC=20=C2=B7=20=EC=A7=84=ED=96=89=20=EA=B8=B0=EB=A1=9D=20?= =?UTF-8?q?=C2=B7=20DB=20=EB=84=A3=EA=B8=B0/=EA=B3=A0=EC=B9=98=EA=B8=B0?= =?UTF-8?q?=EB=A7=8C(keep=20=3D=20=EA=B0=88=EC=9D=8C=20=EB=95=8C=20id=20?= =?UTF-8?q?=C2=B7=20=ED=99=95=EC=A0=95=20=C2=B7=20=EC=9E=85=EB=A0=A5=20?= =?UTF-8?q?=EC=97=B0=EA=B2=B0=20=EC=82=B4=EB=A6=BC)=20=C2=B7=20=EC=98=88?= =?UTF-8?q?=EC=A0=95=EB=85=B8=EC=84=A0=EC=9D=80=20=EA=B7=B8=EA=B1=B8=20?= =?UTF-8?q?=EB=B6=80=EB=A6=84(=EB=8F=99=EC=9E=91=20=C2=B7=20=EC=9D=91?= =?UTF-8?q?=EB=8B=B5=20=EA=B7=B8=EB=8C=80=EB=A1=9C)=20=C2=B7=20=EC=A7=80?= =?UTF-8?q?=ED=82=B4=20=EC=8B=9C=ED=97=98=20test=5F637=5Flidar(=EC=9A=94?= =?UTF-8?q?=EA=B5=AC=20546)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Co-Authored-By: Claude Opus 5.5 --- D03_PlanAnalysis/D03_PlanAnalysis_Lidar.py | 101 +------- D03_PlanAnalysis/D03_PlanAnalysis_Router.py | 73 +----- common_util/common_util_lidar_upload.py | 265 ++++++++++++++++++++ resources/tester/test_637_lidar.py | 157 ++++++++++++ 4 files changed, 440 insertions(+), 156 deletions(-) create mode 100644 common_util/common_util_lidar_upload.py create mode 100644 resources/tester/test_637_lidar.py diff --git a/D03_PlanAnalysis/D03_PlanAnalysis_Lidar.py b/D03_PlanAnalysis/D03_PlanAnalysis_Lidar.py index c50f3d8c3..42c34a3fe 100644 --- a/D03_PlanAnalysis/D03_PlanAnalysis_Lidar.py +++ b/D03_PlanAnalysis/D03_PlanAnalysis_Lidar.py @@ -11,6 +11,7 @@ LAS 좌표계(.prj · LAS 머리 · 없으면 프로젝트 좌표계)가 프로 (높이 표본 격자). 라이다 범위 밖 = 빈칸(격자 `valid_mask`) · 수치지형도로 메우지 않음(결정 A⑥). +이름 검사 · 좌표계 · 옮겨 놓기 · 받기 · 짓기 = 공용 `common_util_lidar_upload`(B05 와 같이 · 637-1). 계산 자리 = ③ 서버 단독. """ @@ -18,14 +19,20 @@ 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 import common_util_lidar_upload as lidar_upload from common_util.common_util_json import atomic_write_json +from common_util.common_util_lidar_upload import ( # noqa: F401 — 옛 이름(시험 · 부르는 쪽) + LAS_SUFFIXES, + check_names, + las_epsg, + place_las, +) router = APIRouter(prefix="/api/projects", tags=["D03 Plan Lidar"]) @@ -34,8 +41,6 @@ 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: @@ -56,86 +61,11 @@ def lidar_dem(root: Path) -> Path | 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, - ) + las_paths = lidar_upload.stage_las(base / UPLOAD, base / LAS_DIR, epsg, report) + result = lidar_upload.build_surface(base, las_paths, report) prefix = LIDAR.as_posix() models = [ {**model, "model_file_path": f"{prefix}/{model['model_file_path']}"} @@ -158,22 +88,13 @@ def build_lidar(root: Path, epsg: int, report: Callable[[int, str], None]) -> di @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 + await lidar_upload.receive_uploads(files, lidar_root(root) / UPLOAD) async def work(report: Any) -> None: got = await asyncio.to_thread(build_lidar, root, epsg, report) diff --git a/D03_PlanAnalysis/D03_PlanAnalysis_Router.py b/D03_PlanAnalysis/D03_PlanAnalysis_Router.py index 676edafa5..74dfc0d70 100644 --- a/D03_PlanAnalysis/D03_PlanAnalysis_Router.py +++ b/D03_PlanAnalysis/D03_PlanAnalysis_Router.py @@ -35,9 +35,9 @@ from pydantic import BaseModel from B03_FileInput.B03_FileInput_Repository import get_project_storage_relative_path from common_util import common_util_contour_tiles as contour_tiles +from common_util import common_util_lidar_upload as lidar_upload from common_util.common_util_crs import lonlat_to_tm, tm_epsg_for_lon from common_util.common_util_http_cache import cached_file_response -from common_util.common_util_json import atomic_write_json from common_util.common_util_storage import resolve_stored_project_path from config.config_db import get_db_pool from D03_PlanAnalysis import D03_PlanAnalysis_ForestRoad as forest_road @@ -118,7 +118,7 @@ def _query_xy(extra: str | None) -> list[tuple[float, float]]: raise HTTPException(status_code=400, detail="extra 꼴이 다릅니다.") from error -IDLE = {"running": False, "percent": 100, "message": None, "error": None, "interrupted": False} +IDLE = lidar_upload.IDLE def _job_path(root: Path, kind: str) -> Path: @@ -126,81 +126,22 @@ def _job_path(root: Path, kind: str) -> Path: def _note_job(path: Path | None, job: dict[str, Any]) -> None: - """진행 기록 — 서버가 다시 켜져도 「돌던 일」을 알게(실패해도 일은 계속).""" - if path is None: - return - try: - path.parent.mkdir(parents=True, exist_ok=True) - atomic_write_json(path, job) - except OSError as error: - logger.warning("D 진행 기록 실패: %s", error) + lidar_upload.note_job(path, job) def _saved_job(root: Path, kind: str) -> dict[str, Any]: - """기억에 없는 일 — 기록이 「도는 중」이면 서버가 다시 켜져 끊김(percent < 100 · - message 없음).""" - try: - saved = json.loads(_job_path(root, kind).read_text(encoding="utf-8")) - except (OSError, ValueError): - return dict(IDLE) - if not (isinstance(saved, dict) and saved.get("running")): - return dict(IDLE) - percent = min(99, int(saved.get("percent") or 0)) - return {**IDLE, "percent": percent, "interrupted": True} + """기억에 없는 일 — 기록이 「도는 중」이면 서버가 다시 켜져 끊김(공용 `saved_job`).""" + return lidar_upload.saved_job(_job_path(root, kind)) def _start_job(project_id: UUID, kind: str, work: Any, root: Path | None = None) -> dict[str, Any]: - key = (str(project_id), kind) - if (_JOBS.get(key) or {}).get("running"): - raise HTTPException(status_code=409, detail="이미 돌고 있습니다.") - job = {"running": True, "percent": 0, "message": "시작", "error": None, "interrupted": False} - _JOBS[key] = job path = _job_path(root, kind) if root is not None else None - _note_job(path, job) - - def report(percent: int, message: str) -> None: - job.update(percent=int(percent), message=message) - _note_job(path, job) - - async def run() -> None: - try: - await work(report) - job.update(percent=100) - except Exception as error: # noqa: BLE001 — 진행률에 까닭을 그대로 올림 - logger.exception("D %s 실패: project_id=%s", kind, project_id) - job["error"] = str(getattr(error, "detail", None) or error) - finally: - job["running"] = False - _note_job(path, job) - - task = asyncio.create_task(run(), name=f"D {kind} {project_id}") - _TASKS.add(task) - task.add_done_callback(_TASKS.discard) - return {"status": "accepted", **job} + return lidar_upload.start_job(_JOBS, _TASKS, project_id, kind, work, path, "D") async def _save_models(project_id: UUID, models: list[dict[str, Any]], epsg: int) -> None: """지형면 → `surface_models` 행(같은 파일 · 같은 지문이면 id 살림 · 다른 행 그대로).""" - from B04_PreProcess.B04_PreProcess_Repository import _existing_models, _upsert_surface_model - - async with get_db_pool().acquire() as connection: - await connection.begin() - try: - existing = await _existing_models(connection, project_id) - for model in models: - await _upsert_surface_model( - connection, - existing, - project_id=project_id, - model=model, - source_file_id=None, - processed_cloud_id=None, - crs_epsg=epsg, - ) - await connection.commit() - except Exception: - await connection.rollback() - raise + await lidar_upload.save_models(project_id, models, epsg) @router.get("/{project_id}/plan/terrain/extent", response_model=None) diff --git a/common_util/common_util_lidar_upload.py b/common_util/common_util_lidar_upload.py new file mode 100644 index 000000000..a78e57d15 --- /dev/null +++ b/common_util/common_util_lidar_upload.py @@ -0,0 +1,265 @@ +"""LAS 나중 올리기 공용 — 예정노선(D03) · B05 가 같이 씀(PLAN 637-1 · 요구 546). + +- 이름 검사 `check_names` · LAS 좌표계 `las_epsg` · 옮겨 놓기 `place_las` + (좌표계 다르면 x · y 옮겨 적음) +- 받기 고리 `receive_uploads` · 입력 자리 갈음 `stage_las` · 짓기 `build_surface` + (B04 엔진 `run_surface_analysis(maps=False)` — 지도 · GIS 받기 · 도엽 서피스 안 함) +- 뒤 일 `start_job` · 진행 기록 `note_job` · `saved_job`(서버 다시 켜짐 = 「끊김」) +- DB `save_models` — `surface_models` 넣기/고치기만 + (다른 행 안 지움 · `save_surface_analysis_to_db` 안 씀) + +부르는 쪽 = `D03_PlanAnalysis_Lidar`(자리 `B04_PreProcess/lidar/`) · +`B05_Profile_Router_Lidar`(프로젝트 뿌리). +계산 자리 = ③ 서버 단독. +""" + +from __future__ import annotations + +import asyncio +import json +import logging +import shutil +from pathlib import Path +from typing import Any, Callable +from uuid import UUID + +from fastapi import HTTPException, UploadFile + +from common_util.common_util_json import atomic_write_json + +logger = logging.getLogger(__name__) + +LAS_SUFFIXES = (".las", ".laz") +CHUNK = 2_000_000 +Report = Callable[[int, str], 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 + + +async def receive_uploads(files: list[UploadFile], upload: Path) -> None: + """올린 파일 → `upload/`(옛 것 비움) · 상한 넘음 = 413.""" + from B03_FileInput.B03_FileInput_Engine import save_upload_stream + + 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 + + +def stage_las(upload: Path, las_dir: Path, epsg: int, report: Report) -> list[Path]: + """`upload/` 의 LAS → 입력 자리 갈음(새로 올림 = 옛 LAS 지움) → 입력 LAS 목록.""" + prj = next(iter(sorted(upload.glob("*.prj"))), None) if upload.is_dir() else 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 upload.is_dir() + else [] + ) + if sources: + 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 가 없습니다.") + return las_paths + + +def build_surface( + base: Path, + las_paths: list[Path], + report: Report, + source_filters: list[str] | None = None, + methods: list[str] | None = None, +) -> dict[str, Any]: + """B04 엔진으로 라이다 지형(지면 필터 · DTM · 평활) — `base/B04_PreProcess/models` 에 지음.""" + from B04_PreProcess.B04_PreProcess_Engine import run_surface_analysis + from config.config_system import SURFACE_AUTO_METHODS + + return run_surface_analysis( + base, + las_paths, + source_filters=source_filters, + methods=list(methods or SURFACE_AUTO_METHODS), + on_progress=lambda percent, _stage, message: report(10 + int(percent * 0.85), message), + maps=False, + ) + + +# ── 뒤 일 · 진행 기록 ───────────────────────────────────────────────────────── + +IDLE = {"running": False, "percent": 100, "message": None, "error": None, "interrupted": False} + + +def note_job(path: Path | None, job: dict[str, Any]) -> None: + """진행 기록 — 서버가 다시 켜져도 「돌던 일」을 알게(실패해도 일은 계속).""" + if path is None: + return + try: + path.parent.mkdir(parents=True, exist_ok=True) + atomic_write_json(path, job) + except OSError as error: + logger.warning("진행 기록 실패: %s", error) + + +def saved_job(path: Path) -> dict[str, Any]: + """기억에 없는 일 — 기록이 「도는 중」이면 서버가 다시 켜져 끊김(percent < 100 · + message 없음).""" + try: + saved = json.loads(path.read_text(encoding="utf-8")) + except (OSError, ValueError): + return dict(IDLE) + if not (isinstance(saved, dict) and saved.get("running")): + return dict(IDLE) + percent = min(99, int(saved.get("percent") or 0)) + return {**IDLE, "percent": percent, "interrupted": True} + + +def start_job( + jobs: dict[tuple[str, str], dict[str, Any]], + tasks: set[asyncio.Task], + project_id: UUID, + kind: str, + work: Any, + path: Path | None = None, + label: str = "D", +) -> dict[str, Any]: + """뒤 일 하나 — 같은 (프로젝트, 일)이 돌면 409 · 진행 = `jobs` · 기록 = `path`.""" + key = (str(project_id), kind) + if (jobs.get(key) or {}).get("running"): + raise HTTPException(status_code=409, detail="이미 돌고 있습니다.") + job = {"running": True, "percent": 0, "message": "시작", "error": None, "interrupted": False} + jobs[key] = job + note_job(path, job) + + def report(percent: int, message: str) -> None: + job.update(percent=int(percent), message=message) + note_job(path, job) + + async def run() -> None: + try: + await work(report) + job.update(percent=100) + except Exception as error: # noqa: BLE001 — 진행률에 까닭을 그대로 올림 + logger.exception("%s %s 실패: project_id=%s", label, kind, project_id) + job["error"] = str(getattr(error, "detail", None) or error) + finally: + job["running"] = False + note_job(path, job) + + task = asyncio.create_task(run(), name=f"{label} {kind} {project_id}") + tasks.add(task) + task.add_done_callback(tasks.discard) + return {"status": "accepted", **job} + + +# ── DB ───────────────────────────────────────────────────────────────────── + + +async def save_models( + project_id: UUID, models: list[dict[str, Any]], epsg: int, *, keep: bool = False +) -> None: + """지형면 → `surface_models` 행(같은 파일 · 같은 지문이면 id 살림 · 다른 행 그대로). + + `keep` = 같은 파일을 가리키는 옛 행은 지문이 달라도 id · 상태(확정) · 입력 연결을 살림 + (LAS 갈음 — 노선 `surface_model_id` · 설계 확정이 끊기지 않게 · B05).""" + from B04_PreProcess.B04_PreProcess_Repository import ( + _existing_models, + _model_key, + _upsert_surface_model, + confirm_surface_model, + ) + from config.config_db import get_db_pool + + async with get_db_pool().acquire() as connection: + await connection.begin() + try: + existing = await _existing_models(connection, project_id) + olds: dict[int, tuple[Any, Any, str]] = {} + if keep: + async with connection.cursor() as cursor: + await cursor.execute( + "SELECT id, source_file_id, processed_cloud_id, status " + "FROM surface_models WHERE project_id = %s", + (str(project_id),), + ) + olds = {int(r[0]): (r[1], r[2], str(r[3])) for r in await cursor.fetchall()} + for model in models: + params = model.get("generation_params") or {} + key = _model_key(model["model_type"], params, model["model_file_path"]) + hit = existing.get(key) if keep else None + if hit is not None: # 지문 맞춤 = 고쳐 씀(id 살림) + existing[key] = (hit[0], params.get("file_stamp")) + source_file_id, cloud_id, status = olds.get(hit[0] if hit else -1, (None, None, "")) + model_id = await _upsert_surface_model( + connection, + existing, + project_id=project_id, + model=model, + source_file_id=source_file_id, + processed_cloud_id=cloud_id, + crs_epsg=epsg, + ) + if status == "CONFIRMED": # 고쳐 쓰기가 COMPLETE 로 내린 확정을 되살림 + await confirm_surface_model(connection, project_id, model_id) + await connection.commit() + except Exception: + await connection.rollback() + raise diff --git a/resources/tester/test_637_lidar.py b/resources/tester/test_637_lidar.py new file mode 100644 index 000000000..7679caafa --- /dev/null +++ b/resources/tester/test_637_lidar.py @@ -0,0 +1,157 @@ +"""PLAN 637 — LAS 나중 올리기 공용(`common_util_lidar_upload`) · B05 창구(요구 546). + +637-1 = 예정노선 길을 공용으로(동작 그대로 · `test_624_4_plan_lidar` 함께) — +갈음 · 짓기 · DB 넣기/고치기만. +""" + +from __future__ import annotations + +from pathlib import Path +from typing import Any +from uuid import UUID + +import numpy as np +import pytest + +from common_util import common_util_lidar_upload as lidar_upload + +PID = UUID("00000000-0000-0000-0000-000000000637") +EPSG = 5186 + + +def write_las(path: Path, x0: float, y0: float, size: float = 60.0, n: int = 6000) -> None: + """지면점 LAS(비탈 0.3 · 좌표계 없음).""" + import laspy + + rng = np.random.default_rng(7) + x, y = rng.uniform(x0, x0 + size, n), rng.uniform(y0, y0 + size, n) + header = laspy.LasHeader(point_format=3, version="1.2") + header.offsets = [round(x0), round(y0), 0] + header.scales = [0.01, 0.01, 0.01] + 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 test_갈음_짓기_지도_안_받음(tmp_path: Path, monkeypatch) -> None: # type: ignore[no-untyped-def] + import B04_PreProcess.B04_PreProcess_Engine as engine + + monkeypatch.setattr(engine, "download_geodata", lambda *a, **k: pytest.fail("지도 받기")) + upload, las_dir = tmp_path / "up", tmp_path / "las" + write_las(las_dir / "old.las", 200000.0, 500000.0) + write_las(upload / "new.las", 200000.0, 500000.0) + got = lidar_upload.stage_las(upload, las_dir, EPSG, lambda *_: None) + assert [p.name for p in got] == ["new.las"] and not upload.exists() # 옛 LAS 갈음 + again = lidar_upload.stage_las(upload, las_dir, EPSG, lambda *_: None) + assert [p.name for p in again] == ["new.las"] # 올림 없음 = 있는 것 그대로 + seen: list[int] = [] + result = lidar_upload.build_surface(tmp_path, got, lambda p, _m: seen.append(p)) + picked = result["source_filters"][0] + assert (tmp_path / "B04_PreProcess" / "models" / f"dtm_{picked}.npz").is_file() + assert seen and min(seen) >= 10 and max(seen) <= 95 + with pytest.raises(FileNotFoundError): + lidar_upload.stage_las(upload, tmp_path / "빈", EPSG, lambda *_: None) + + +def test_진행_기록_끊김(tmp_path: Path) -> None: + path = tmp_path / "job.json" + assert lidar_upload.saved_job(path) == lidar_upload.IDLE + lidar_upload.note_job(path, {"running": True, "percent": 42}) + assert lidar_upload.saved_job(path) == {**lidar_upload.IDLE, "percent": 42, "interrupted": True} + + +class _Cursor: + def __init__(self, rows: list[tuple]) -> None: + self.rows = rows + + async def __aenter__(self) -> "_Cursor": + return self + + async def __aexit__(self, *_: Any) -> None: + return None + + async def execute(self, *_: Any) -> None: + return None + + async def fetchall(self) -> list[tuple]: + return self.rows + + +class _Connection: + def __init__(self, rows: list[tuple]) -> None: + self.rows, self.done = rows, [] + + async def begin(self) -> None: + return None + + async def commit(self) -> None: + self.done.append("commit") + + async def rollback(self) -> None: + self.done.append("rollback") + + def cursor(self) -> _Cursor: + return _Cursor(self.rows) + + +class _Pool: + def __init__(self, connection: _Connection) -> None: + self.connection = connection + + def acquire(self) -> "_Pool": + return self + + async def __aenter__(self) -> _Connection: + return self.connection + + async def __aexit__(self, *_: Any) -> None: + return None + + +def _save(monkeypatch, keep: bool) -> tuple[list[dict], list[int], _Connection]: # type: ignore[no-untyped-def] + """옛 행 둘(확정 dtm_a · 수치지형도) + 새 모델 dtm_a(지문 다름) → 부른 값.""" + import asyncio + + import B04_PreProcess.B04_PreProcess_Repository as repo + import config.config_db as config_db + + path_a = "B04_PreProcess/models/dtm_a.npz" + existing = {("dtm", "a", path_a): (11, "old"), ("dtm", "sheet_laplace", "s.npz"): (12, "s")} + calls: list[dict] = [] + confirmed: list[int] = [] + + async def existing_models(*_: Any) -> dict: + return dict(existing) + + async def upsert(_c: Any, have: dict, **kw: Any) -> int: + key = ("dtm", "a", path_a) + calls.append({**kw, "have": have.get(key)}) + return 11 if have.get(key, (0, ""))[1] == "new" else 99 + + async def confirm(_c: Any, _p: Any, model_id: int) -> None: + confirmed.append(model_id) + + connection = _Connection([(11, 1320, 7, "CONFIRMED"), (12, None, None, "COMPLETE")]) + monkeypatch.setattr(repo, "_existing_models", existing_models) + monkeypatch.setattr(repo, "_upsert_surface_model", upsert) + monkeypatch.setattr(repo, "confirm_surface_model", confirm) + monkeypatch.setattr(config_db, "get_db_pool", lambda: _Pool(connection)) + model = { + "model_type": "dtm", + "model_file_path": path_a, + "generation_params": {"source_filter": "a", "file_stamp": "new"}, + } + asyncio.run(lidar_upload.save_models(PID, [model], EPSG, keep=keep)) + return calls, confirmed, connection + + +def test_DB_넣기_고치기만_갈음은_id_확정_살림(monkeypatch) -> None: # type: ignore[no-untyped-def] + calls, confirmed, connection = _save(monkeypatch, keep=False) # 예정노선 = 옛 그대로 + assert calls[0]["have"] == (11, "old") and calls[0]["source_file_id"] is None + assert confirmed == [] and connection.done == ["commit"] + calls, confirmed, _ = _save(monkeypatch, keep=True) # B 갈음 = 지문 맞춤 · 연결 · 확정 살림 + assert calls[0]["have"] == (11, "new") + assert (calls[0]["source_file_id"], calls[0]["processed_cloud_id"]) == (1320, 7) + assert confirmed == [11]