refactor(637-1): 예정노선 LAS 올리기 · 짓기 공용 모듈 common_util_lidar_upload — 이름 검사 · 좌표계 · 옮겨 놓기 · 받기 · 갈음 · 짓기(maps=False) · 뒤 일 · 진행 기록 · DB 넣기/고치기만(keep = 갈음 때 id · 확정 · 입력 연결 살림) · 예정노선은 그걸 부름(동작 · 응답 그대로) · 지킴 시험 test_637_lidar(요구 546)
Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
This commit is contained in:
Claude Opus 5.5
@@ -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)
|
||||
|
||||
@@ -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)
|
||||
|
||||
@@ -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
|
||||
@@ -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]
|
||||
Reference in New Issue
Block a user