Aislo/resources/tester/test_133_1_analysis_terrain.py

"""PLAN 133-1 — B05 계획노선 편집 분석 지형(라이다 · 수치지형도) 고르기.

라이다 있는 프로젝트 → 두 지형 다 고름 · sheet 고르면 dtm_sheet 로 표고 · 라이다 없는 프로젝트 →
고르기 안 보임(지형 하나) · 허용 목록 밖 거절 · 확정 params 는 안 바뀜.
"""

from __future__ import annotations

import asyncio
from contextlib import asynccontextmanager
from pathlib import Path

import numpy as np
import pytest
from pydantic import ValidationError

from B05_Profile import B05_Profile_Router_Replan as replan
from B05_Profile import B05_Profile_Router_RouteInitial as initial
from B05_Profile import B05_Profile_Router_Terrain as terrain_router
from common_util.common_util_surface_confirmation import (
    SHEET_ANALYSIS_FILTER,
    analysis_terrain_params,
    analysis_terrain_sources,
)
from common_util.common_util_surface_sampler import build_surface_sampler, check_surface_keys

ROOT = Path(__file__).resolve().parents[2]
LAS_PARAMS = {"source_filter": "csf", "method": "dtm", "smooth": True, "contour_interval_m": 1.0}
SHEET_PARAMS = {
    "source_filter": SHEET_ANALYSIS_FILTER,
    "method": "dtm",
    "smooth": True,
    "contour_interval_m": 1.0,
}
LAS_Z = 100.0
SHEET_Z = 250.0


def _dtm(path: Path, z: float) -> None:
    x = np.arange(0.0, 11.0)
    y = np.arange(0.0, 11.0)
    np.savez(
        path,
        x=x,
        y=y,
        z=np.full((len(y), len(x)), z),
        valid_mask=np.ones((len(y), len(x)), dtype=bool),
    )


def _project(tmp_path: Path, *, las: bool, sheet: bool) -> Path:
    models = tmp_path / "B04_PreProcess" / "models"
    models.mkdir(parents=True)
    if las:
        _dtm(models / "dtm_csf_smooth.npz", LAS_Z)
    if sheet:  # B04 처럼 원본 · 평활 둘 다(평활 쪽을 읽음)
        _dtm(models / f"dtm_{SHEET_ANALYSIS_FILTER}.npz", SHEET_Z - 1.0)
        _dtm(models / f"dtm_{SHEET_ANALYSIS_FILTER}_smooth.npz", SHEET_Z)
    return tmp_path


def _patch_db(monkeypatch, project_root: Path, params: dict) -> None:
    class _Pool:
        @asynccontextmanager
        async def acquire(self):
            yield object()

    async def stored(_connection, _project_id):
        return "x"

    async def confirmed(_connection, _project_id):
        return dict(params)

    monkeypatch.setattr(terrain_router, "get_db_pool", lambda: _Pool())
    monkeypatch.setattr(terrain_router, "get_project_storage_relative_path", stored)
    monkeypatch.setattr(terrain_router, "get_surface_confirmation_params", confirmed)
    monkeypatch.setattr(terrain_router, "resolve_stored_project_path", lambda _s: project_root)


PID = "1bb0f876-983f-4ca5-9cc7-f59ee1f77e0a"


def _elevations(terrain: str) -> dict:
    request = terrain_router.ElevationRequest(points=[(5.0, 5.0)], terrain=terrain)
    return asyncio.run(terrain_router.read_route_elevations(PID, request))


def test_라이다_있는_프로젝트_두_지형_다_고름(tmp_path, monkeypatch):
    root = _project(tmp_path, las=True, sheet=True)
    models = root / "B04_PreProcess" / "models"
    assert analysis_terrain_sources(LAS_PARAMS, models) == ["las", "sheet"]
    _patch_db(monkeypatch, root, LAS_PARAMS)
    answer = asyncio.run(terrain_router.read_terrain_sources(PID))
    assert answer["sources"] == ["las", "sheet"]


def test_sheet_고르면_dtm_sheet_로_표고(tmp_path, monkeypatch):
    root = _project(tmp_path, las=True, sheet=True)
    _patch_db(monkeypatch, root, LAS_PARAMS)
    assert _elevations("las")["z"] == [LAS_Z]
    assert _elevations("sheet")["z"] == [SHEET_Z]
    # 기본값 = 지금 설계 지형(확정 지표면 · 133-2)
    request = terrain_router.ElevationRequest(points=[(5.0, 5.0)])
    assert request.terrain is None


def test_확정_params_는_안_바뀜():
    params = dict(LAS_PARAMS)
    picked = analysis_terrain_params(params, "sheet")
    assert picked["source_filter"] == SHEET_ANALYSIS_FILTER == "sheet_laplace"
    assert picked["method"] == "dtm" and picked["smooth"] is True
    assert params == LAS_PARAMS
    assert analysis_terrain_params(params, "las") is params


def test_라이다_없는_프로젝트_고르기_안_보임(tmp_path, monkeypatch):
    root = _project(tmp_path, las=False, sheet=True)
    models = root / "B04_PreProcess" / "models"
    assert analysis_terrain_sources(SHEET_PARAMS, models) == ["sheet"]
    # 확정이 이미 수치지형도 — 어느 값을 받아도 같은 서피스
    assert analysis_terrain_params(SHEET_PARAMS, "sheet") is SHEET_PARAMS
    _patch_db(monkeypatch, root, SHEET_PARAMS)
    assert asyncio.run(terrain_router.read_terrain_sources(PID))["sources"] == ["sheet"]
    # 허용 목록에 sheet_* 가 들어가 LAS 없는 프로젝트 지반고가 읽힘(54-22 뒤 막혀 있던 길)
    assert _elevations("las")["z"] == [SHEET_Z]


def test_수치지형도_서피스_없으면_라이다만(tmp_path):
    root = _project(tmp_path, las=True, sheet=False)
    assert analysis_terrain_sources(LAS_PARAMS, root / "B04_PreProcess" / "models") == ["las"]


@pytest.mark.parametrize("bad", ["lidar", "SHEET", "../x", ""])
def test_허용_목록_밖_거절(bad):
    with pytest.raises(ValidationError):
        terrain_router.ElevationRequest(points=[(0.0, 0.0)], terrain=bad)
    with pytest.raises(ValidationError):
        terrain_router.CrossPreviewRequest(
            vertices=[{"x": 0, "y": 0}, {"x": 1, "y": 1}], chainage_m=0, terrain=bad
        )
    with pytest.raises(ValidationError):
        initial.ProfilePreviewRequest(vertices=[{"x": 0, "y": 0}, {"x": 1, "y": 1}], terrain=bad)
    with pytest.raises(ValidationError):
        replan.RouteInitialRequest(terrain=bad)


@pytest.mark.parametrize("bad", ["sheet", "sheet_../x", "sheet_LAPLACE", "dtm_sheet_laplace"])
def test_sheet_필터도_허용_목록_검사(bad, tmp_path):
    with pytest.raises(ValueError):
        check_surface_keys(bad)
    with pytest.raises(ValueError):
        build_surface_sampler(tmp_path, bad, "dtm", True)
    check_surface_keys("sheet_laplace")


def test_화면_고르기_배선():
    page = ROOT / "B05_Profile"
    chrome = (page / "B05_Profile_UI_RouteEdit_Chrome.ts").read_text(encoding="utf-8")
    main = (page / "B05_Profile_UI_RouteEdit.ts").read_text(encoding="utf-8")
    api = (page / "B05_Profile_Api_Replan.ts").read_text(encoding="utf-8")
    initial_api = (page / "B05_Profile_Api_RouteInitial.ts").read_text(encoding="utf-8")
    # 기본 숨김 · 둘 다 있을 때만 보임 · 기본 라이다
    assert '<label class="b05-routeedit__terrain" hidden' in chrome
    assert '<option value="las">라이다</option>' in chrome
    assert '<option value="sheet">수치지형도</option>' in chrome
    assert 'sources?.sources.includes("las") && !!sources?.sources.includes("sheet")' in main
    assert 'saved === "sheet" || saved === "las" ? saved : fallback' in api
    # 모달 분석 넷이 고른 지형을 실어 보냄 · [확인] 재계산은 안 실음
    assert api.count("terrain: analysisTerrain(projectId)") == 3
    assert "terrain: analysisTerrain(projectId)" in initial_api
    replan_fn = api.split("export async function replanRoute")[1].split("export ")[0]
    assert "terrain" not in replan_fn