266 lines
11 KiB
Python
266 lines
11 KiB
Python
"""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
|