Aislo/B05_Profile/B05_Profile_Router_Terrain.py

"""계획노선 편집 중 **지반고만** 묻는 가벼운 통로.

편집 모달은 [확인] 전까지 아무 계산도 내보내지 않는다(계획서 0-2 확정 7). 다만 두 점을 찍어
**구간 길이와 종단기울기**를 볼 때(0-9 ⑤)는 지반고가 있어야 한다(측점 횡단 미리보기 = 계산
묶음 · 365-3). 새 계산이 아니라 **이미 확정된 지표면을 읽기만** 하는 통로라 그 규칙과 부딪히지
않는다 — 노선을 갈아 끼우지도, 정본을 건드리지도 않는다.

표고 조회는 종·횡단 생성기가 쓰는 것과 **같은 sampler**(`build_surface_sampler`)를 연다.
두 화면이 다른 표고를 보면 같은 자리의 기울기가 갈린다.

  POST /api/projects/{id}/route/elevations    → 점 묶음의 지반고
  GET  /api/projects/{id}/route/terrain-sources → 고를 수 있는 분석 지형(라이다 · 수치지형도)
  GET  /api/projects/{id}/surface/models/{mid}/grid → DTM 격자 원시 이진(브라우저 표본기 · ETag)

분석 지형(`terrain`, PLAN 133-1) — las = 라이다 · sheet = 수치지형도 · None = 지금 설계 지형.
모달 분석만 고른 지형을 쓴다 — 설계 지형(확정)은 [확인] 때만 바뀜(133-2 · `/route/replan`).
"""

import asyncio
import json
import logging
from pathlib import Path
from uuid import UUID

import aiomysql
import numpy as np
from fastapi import APIRouter, Request
from fastapi.responses import JSONResponse, Response
from pydantic import BaseModel, Field

from B03_FileInput.B03_FileInput_Repository import get_project_storage_relative_path
from common_util.common_util_http_cache import cached_file_response
from common_util.common_util_storage import resolve_stored_project_path
from common_util.common_util_surface_confirmation import (
    AnalysisTerrain,
    analysis_terrain_params,
    analysis_terrain_sources,
    design_terrain,
    get_surface_confirmation_params,
)
from common_util.common_util_surface_sampler import build_surface_sampler, dtm_grid_file
from config.config_db import get_db_pool

logger = logging.getLogger(__name__)
router = APIRouter(prefix="/api/projects", tags=["B05 Route Terrain"])

_MODELS_SUBDIR = Path("B04_PreProcess") / "models"
#: 한 번에 물을 수 있는 점 수. 구간 재기는 수십 점, 횡단 한 장은 수백 점이면 넉넉하다 —
#: 상한을 두어 실수로 노선 전체를 밀어 넣는 일을 막는다.
MAX_POINTS = 4000


class ElevationRequest(BaseModel):
    """사업지 좌표계(m) 점 묶음 [[x, y], …]."""

    points: list[tuple[float, float]] = Field(..., min_length=1, max_length=MAX_POINTS)
    terrain: AnalysisTerrain | None = None
    #: 표고 자릿수 — 화면 값은 mm(3) · 계산 묶음 서버 길은 6(격자 표본 · 서버 생성과 같은 자리).
    decimals: int = Field(3, ge=0, le=6)


def _sample(project_root: Path, params: dict, points: list[tuple[float, float]]):
    """확정 지표면에서 표고를 읽는다. 모델을 못 열면 None."""
    try:
        sampler = build_surface_sampler(
            project_root / _MODELS_SUBDIR,
            str(params["source_filter"]),
            str(params["method"]),
            bool(params["smooth"]),
        )
    except (FileNotFoundError, KeyError, OSError, ValueError) as exc:
        logger.warning("계획노선 편집: 지표면을 열지 못했습니다 — %s", exc)
        return None
    z, valid = sampler.sample_xy(np.asarray(points, dtype=np.float64))
    return z, valid


@router.post("/{project_id}/route/elevations", response_model=None)
async def read_route_elevations(project_id: UUID, request: ElevationRequest) -> dict | JSONResponse:
    """점 묶음의 지반고(m)와 유효 여부를 돌려준다.

    지표면 밖이거나 자료가 없는 자리는 `valid=false` 로 나가고 표고는 `null` 이다 —
    **임의 표고로 메우지 않는다**(sampler 규칙 그대로). 화면은 그 자리를 「모름」으로 낸다.
    """
    pool = get_db_pool()
    async with pool.acquire() as connection:
        stored = await get_project_storage_relative_path(connection, project_id)
        if not stored:
            return JSONResponse(
                status_code=404,
                content={"status": "error", "message": "프로젝트 저장 경로를 찾을 수 없습니다."},
            )
        params = await get_surface_confirmation_params(connection, str(project_id))
    params = analysis_terrain_params(params, request.terrain)

    project_root = Path(resolve_stored_project_path(stored))
    sampled = await asyncio.to_thread(_sample, project_root, params, request.points)
    if sampled is None:
        return JSONResponse(
            status_code=409,
            content={
                "status": "error",
                "message": "확정된 지표면이 없어 지반고를 읽을 수 없습니다.",
            },
        )
    z, valid = sampled
    return {
        "status": "success",
        "project_id": str(project_id),
        "z": [
            None if not ok else round(float(value), request.decimals) for value, ok in zip(z, valid)
        ],
        "valid": [bool(ok) for ok in valid],
    }


@router.get("/{project_id}/route/terrain-sources", response_model=None)
async def read_terrain_sources(project_id: UUID) -> dict | JSONResponse:
    """편집 모달이 고를 수 있는 분석 지형 — 둘이면 고르기를 띄움 · 하나면 숨김."""
    pool = get_db_pool()
    async with pool.acquire() as connection:
        stored = await get_project_storage_relative_path(connection, project_id)
        if not stored:
            return JSONResponse(
                status_code=404,
                content={"status": "error", "message": "프로젝트 저장 경로를 찾을 수 없습니다."},
            )
        params = await get_surface_confirmation_params(connection, str(project_id))
        confirmed = await _confirmed_model(connection, project_id)
    models_dir = Path(resolve_stored_project_path(stored)) / _MODELS_SUBDIR
    sources = await asyncio.to_thread(analysis_terrain_sources, params, models_dir)
    return {
        "status": "success",
        "project_id": str(project_id),
        "sources": sources,
        "design": design_terrain(params),  # 지금 설계 지형(PLAN 133-2)
        "grid": await asyncio.to_thread(_design_grid, params, models_dir, confirmed),
    }


async def _confirmed_model(
    connection: aiomysql.Connection, project_id: UUID
) -> tuple[int, str, str] | None:
    """확정 지표면 모델 (id, 표현, 파일 경로) — 없으면 None."""
    async with connection.cursor() as cursor:
        await cursor.execute(
            "SELECT id, model_type, model_file_path FROM surface_models"
            " WHERE project_id = %s AND status = 'CONFIRMED' ORDER BY created_at DESC LIMIT 1",
            (str(project_id),),
        )
        row = await cursor.fetchone()
    return (int(row[0]), str(row[1]), str(row[2] or "")) if row else None


def _design_grid(
    params: dict, models_dir: Path, confirmed: tuple[int, str, str] | None
) -> dict | None:
    """설계 지형 격자 받을 자리(PLAN 344-1) — 서버 표본기(`build_surface_sampler`)와 같은 파일만.

    DTM 아닌 표현(TIN 등)은 None → 화면은 서버 표본 길 그대로. `signature` = manifest 구성 ·
    평활 서명 — 화면이 주소에 붙여 지형이 바뀐 뒤 옛 격자를 먼저 내지 않음.
    """
    source_filter = str(params.get("source_filter") or "")
    if not confirmed or params.get("method") != "dtm" or confirmed[1] != "dtm":
        return None
    if Path(confirmed[2]).name != f"dtm_{source_filter}.npz":
        return None
    smooth = bool(params.get("smooth"))
    try:
        manifest = json.loads((models_dir / "manifest.json").read_text(encoding="utf-8"))
    except (OSError, ValueError):
        manifest = {}
    dtm = (
        manifest.get("source_filters", {}).get(source_filter, {}).get("methods", {}).get("dtm", {})
    )
    smoothing = (dtm.get("smooth") or {}).get("smoothing_signature") if smooth else None
    return {
        "model_id": confirmed[0],
        "smooth": smooth,
        "signature": f"{manifest.get('config_signature') or ''}.{smoothing or ''}",
    }


@router.get("/{project_id}/surface/models/{model_id}/grid", response_model=None)
async def read_surface_grid(
    request: Request, project_id: UUID, model_id: int, smooth: bool = False
) -> Response | JSONResponse:
    """DTM 격자 원시 이진(`dtm_grid_bytes` 꼴 · ETag) — 첫 요청 때 npz 곁에 `.bin` 저장본을 씀.

    평활 고르기는 서버 표본기와 같음(평활 파일 없으면 원본). DTM 아닌 표현은 404.
    """
    pool = get_db_pool()
    async with pool.acquire() as connection:
        stored = await get_project_storage_relative_path(connection, project_id)
        async with connection.cursor() as cursor:
            await cursor.execute(
                "SELECT model_type, model_file_path FROM surface_models"
                " WHERE id = %s AND project_id = %s",
                (model_id, str(project_id)),
            )
            row = await cursor.fetchone()
    if not stored or not row or row[0] != "dtm" or not row[1]:
        return JSONResponse(
            status_code=404,
            content={"status": "error", "message": "DTM 격자가 없는 지표면 모델입니다."},
        )
    model_path = Path(resolve_stored_project_path(stored)) / str(row[1])
    smooth_path = model_path.with_name(f"{model_path.stem}_smooth.npz")
    npz_path = smooth_path if smooth and smooth_path.is_file() else model_path
    if not npz_path.is_file():
        return JSONResponse(
            status_code=404,
            content={"status": "error", "message": "DTM 격자 파일이 없습니다."},
        )
    grid_path = await asyncio.to_thread(dtm_grid_file, npz_path)
    return cached_file_response(request, grid_path, "application/octet-stream")