Aislo/D03_PlanAnalysis/D03_PlanAnalysis_TerrainSource.py

"""D03 지도 겹의 프로젝트 틀 · 지형 원천 (요구 530 · 531 · PLAN 624-1).

겹 길(`/plan/layers/*`)은 프로젝트 종류를 안 가림:

- 예정노선(PLAN) = 받은 범위 · 지형 = 수치지형도(`models_plan/dtm_sheet_<방법>.npz`) ·
  `lidar` = D 에 올린 LAS 로 지은 DTM(624-4 · `D03_PlanAnalysis_Lidar` · 안 올렸으면 404 ·
  범위 밖 = 빈칸)
- 그 밖(임도 등) = 범위 = B04 배경 지도 상자(`vworld_white_meta.json` · 없으면 원천 격자 상자) ·
  지형 = 설계 지형(stage 1 params · 기본) · `lidar` = 라이다 확정 지표면 ·
  `sheet` = 수치지형도 서피스 ·
  평활 고르기 = 서버 표본기와 같음(`B05_Profile_Router_Terrain.dtm_grid_npz`)
- 원천 인자 `terrain=sheet|lidar|design`(빈 값 = design) · 파생 저장본 열쇠 = 격자 파일 이름(원천)
- 받은 겹 저장 = 종류 상관없이 `D03_PlanAnalysis/layers/`(B04 `processed/` 파일과 이름 겹침 0) ·
  D 옛 자리(`B04_PreProcess/processed/`) 파일은 새 자리에 없을 때 한 번 복사(다시 받지 않게)

계산 자리 = ③ 서버 단독.
"""

from __future__ import annotations

import json
import shutil
from pathlib import Path
from typing import Any, Callable, NamedTuple

import numpy as np
from fastapi import HTTPException

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

Box = tuple[float, float, float, float]

SOURCES = ("sheet", "lidar", "design")
LAYER_DIR = Path("D03_PlanAnalysis") / "layers"
OLD_DIR = Path("B04_PreProcess") / "processed"
ROAD_MODELS = Path("B04_PreProcess") / "models"
ROAD_META = ("vworld_white_meta.json", "vworld_satellite_meta.json")


def _legacy_names() -> list[str]:
    names = [layers.CADASTRAL_FILE, layers.CADASTRAL_KEY, layers.ZONE_FILE, layers.ZONE_KEY]
    names += [layers.SOIL_FILE, layers.SOIL_KEY, landslide.LANDSLIDE_FILE]
    for label, _ in layers.LINE_KINDS.values():
        names += [f"{label}_bounds.geojson", f"{label}_bounds.key.json"]
    return names


class Frame(NamedTuple):
    root: Path
    epsg: int
    box: Box
    store: Path  # 받은 겹 자리(D03 몫)
    source: str  # 실제 원천 — sheet · lidar
    dem: Path | None  # 원천 격자(없으면 None)


def check_source(choice: str | None) -> str:
    if choice in (None, ""):
        return "design"
    if choice not in SOURCES:
        raise HTTPException(status_code=400, detail=f"없는 지형 원천: {choice}")
    return choice


def layer_store(root: Path, plan: bool) -> Path:
    """받은 겹 자리 — D 는 옛 자리 파일을 처음 한 번 복사(덮지 않음)."""
    store = Path(root) / LAYER_DIR
    store.mkdir(parents=True, exist_ok=True)
    if plan:
        for name in _legacy_names():
            old, new = Path(root) / OLD_DIR / name, store / name
            if old.is_file() and not new.exists():
                shutil.copy2(old, new)
    return store


def plan_frame(root: Path, epsg: int, box: Box, choice: str | None) -> Frame:
    """예정노선 — 수치지형도 하나(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"
    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]:
    """(실제 원천, 격자) — 고른 원천 params 의 `dtm_<필터>`(평활이면 `_smooth` 먼저) ·
    라이다 없는 프로젝트의 lidar = None."""
    from common_util.common_util_surface_confirmation import (
        analysis_terrain_params,
        design_terrain,
    )

    want = None if source == "design" else ("las" if source == "lidar" else "sheet")
    picked = analysis_terrain_params(params, want)
    got = "sheet" if design_terrain(picked) == "sheet" else "lidar"
    if source != "design" and got != source:
        return source, None
    npz = Path(models) / f"dtm_{picked.get('source_filter') or ''}.npz"
    smooth = npz.with_name(f"{npz.stem}_smooth.npz")
    npz = smooth if picked.get("smooth") and smooth.is_file() else npz
    return got, npz if npz.is_file() else None


def grid_box(npz: Path) -> Box:
    with np.load(npz) as dem:
        x, y = np.asarray(dem["x"], float), np.asarray(dem["y"], float)
    half = abs(float(x[1] - x[0])) / 2 if len(x) > 1 else 0.5
    return (x[0] - half, y[0] - half, x[-1] + half, y[-1] + half)


def road_box(root: Path, dem: Path | None) -> Box | None:
    """B04 배경 지도 상자(작업 좌표계 m) · 없으면 격자 상자."""
    for name in ROAD_META:
        try:
            meta = json.loads((Path(root) / OLD_DIR / name).read_text(encoding="utf-8"))
            return (meta["x_min"], meta["y_min"], meta["x_max"], meta["y_max"])
        except (OSError, ValueError, KeyError, TypeError):
            continue
    return grid_box(dem) if dem is not None else None


async def road_frame(row: dict[str, Any], project_id: Any, choice: str | None) -> Frame:
    """임도 등 — 범위 · 좌표계 · 설계 지형 params 로 원천 격자."""
    from pyproj import CRS

    from common_util.common_util_crs import identify_epsg, resolve_project_crs
    from common_util.common_util_surface_confirmation import get_surface_confirmation_params
    from config.config_db import get_db_pool

    source = check_source(choice)
    root = Path(resolve_stored_project_path(row["storage_path"]))
    async with get_db_pool().acquire() as connection:
        params = await get_surface_confirmation_params(connection, str(project_id))
    got, dem = road_dem(root / ROAD_MODELS, params, source)
    box = road_box(root, dem)
    if box is None:
        raise HTTPException(status_code=404, detail="지형 전처리 뒤 보입니다.")
    text = resolve_project_crs(root)
    epsg = identify_epsg(CRS.from_user_input(text), text)
    if epsg is None:
        raise HTTPException(status_code=404, detail="프로젝트 좌표계를 모릅니다.")
    return Frame(root, epsg, tuple(float(v) for v in box), layer_store(root, False), got, dem)


def need_dem(frame: Frame) -> Path:
    if frame.dem is None:
        what = "라이다" if frame.source == "lidar" else "수치지형도"
        raise HTTPException(status_code=404, detail=f"{what} 지형이 없습니다.")
    return frame.dem


async def layer_frame(project_id: Any, source: str | None, plan_ends: Callable) -> Frame:
    """겹 틀 — 예정노선 = 받은 범위(받기 전 404) · 그 밖 = `road_frame`."""
    row = await store.project_row(project_id)
    if row["project_kind"] != "PLAN":
        return await road_frame(row, project_id, source)
    root, epsg, _, _ = await plan_ends(project_id)
    box = terrain.received_box(root, epsg)
    if box is None:
        raise HTTPException(status_code=404, detail="지형 받기 뒤 보입니다.")
    return plan_frame(root, epsg, box, source)


def zero_order_calls(frame: Frame, min_m2: float, max_m2: float) -> tuple[Callable, tuple, tuple]:
    """(셈, 가벼운 인자 = 저장본만, 무거운 인자) — 수치지형도 = 창고 도엽 선(범위 = 원천 격자
    상자) · 라이다 = 격자 통째."""
    npz = need_dem(frame)
    if frame.source == "sheet":
        args = (frame.root, npz.parent, frame.epsg, min_m2, max_m2)
        return zero_order.zero_order_layer, (*args, False, npz), (*args, True, npz)
    args = (frame.root, npz, frame.epsg, min_m2, max_m2)
    return zero_order.dem_zero_order_layer, (*args, False), args