Aislo/resources/tester/test_133_2_design_terrain.py

"""PLAN 133-2 — 노선 편집 [확인] 때 고른 분석 지형 = 설계 지형(뒤 단계 전부).

수치지형도 고르고 [확인] → 설계 지형 sheet(확정 모델 · stage 1 params) · 종 · 횡단 · 배수 · 유토가
stage 1 params 로 sheet sampler · 라이다로 되돌려 [확인] → 라이다 · [확인] 안 하면 그대로 ·
라이다 없는 프로젝트 그대로 · 체인 실패면 설계 지형도 되돌림.
"""

from __future__ import annotations

import asyncio
import json
from contextlib import asynccontextmanager
from pathlib import Path
from uuid import UUID

import numpy as np

import B03_FileInput.B03_FileInput_Service_Chain as chain
import B04_PreProcess.B04_PreProcess_Service as service
import B05_Profile.B05_Profile_Repository as route_repo
import B05_Profile.B05_Profile_Route_Save as route_save
import common_util.common_util_workflow_state as workflow_state
from B05_Profile import B05_Profile_Router_Replan as replan
from B06_Section import B06_Section_Router_Stations as stations
from common_util.common_util_surface_confirmation import (
    LAS_STASH_KEY,
    SHEET_ANALYSIS_FILTER,
    analysis_terrain_sources,
    design_terrain,
    get_surface_confirmation_params,
)
from common_util.common_util_surface_sampler import build_surface_sampler

ROOT = Path(__file__).resolve().parents[2]
PID = UUID("1bb0f876-983f-4ca5-9cc7-f59ee1f77e0a")
LAS_ID, SHEET_ID = 582, 584
LAS_Z, SHEET_Z = 100.0, 250.0
LAS_STAGE1 = {"source_filter": "classification", "method": "dtm", "smooth": True}
SHEET_STAGE1 = {"source_filter": SHEET_ANALYSIS_FILTER, "method": "dtm", "smooth": True}


class _Cursor:
    def __init__(self, db: _Db):
        self.db = db
        self.row = None
        self.rowcount = 1

    async def __aenter__(self):
        return self

    async def __aexit__(self, *_):
        return False

    async def execute(self, sql: str, args=()):
        if "SELECT params" in sql:
            self.row = {"params": json.dumps(self.db.stage1)}
        elif "UPDATE project_workflow_stages" in sql:
            self.db.stage1 = json.loads(args[0])
            self.db.writes += 1
        elif "route_start_m" in sql:
            self.row = (None, None)

    async def fetchone(self):
        return self.row


class _Db:
    """stage 1 params 한 행 · 지표면 모델 표 — 실제 DB 대신."""

    def __init__(self, stage1: dict, models: list[dict]):
        self.stage1 = dict(stage1)
        self.models = models
        self.writes = 0

    def cursor(self, *_):
        return _Cursor(self)

    async def begin(self):
        return None

    async def commit(self):
        return None

    async def rollback(self):
        return None

    def confirmed(self) -> int | None:
        return next((m["id"] for m in self.models if m["status"] == "CONFIRMED"), None)


def _models(las: bool) -> list[dict]:
    rows = [
        {
            "id": SHEET_ID,
            "model_type": "dtm",
            "status": "COMPLETE" if las else "CONFIRMED",
            "generation_params": {"source_filter": SHEET_ANALYSIS_FILTER},
        }
    ]
    if las:
        rows.insert(
            0,
            {
                "id": LAS_ID,
                "model_type": "dtm",
                "status": "CONFIRMED",
                "generation_params": {"source_filter": "classification"},
            },
        )
    return rows


def _patch_models(monkeypatch, db: _Db) -> None:
    async def listed(_connection, _project_id):
        return db.models

    async def confirm(_connection, _project_id, model_id):
        for model in db.models:
            if model["status"] == "CONFIRMED":
                model["status"] = "COMPLETE"
        next(m for m in db.models if m["id"] == model_id)["status"] = "CONFIRMED"

    monkeypatch.setattr(service, "list_surface_models", listed)
    monkeypatch.setattr(service, "confirm_surface_model", confirm)


def _switch(db: _Db, terrain):
    return asyncio.run(service.switch_design_terrain(db, PID, terrain))


def _params(db: _Db) -> dict:
    return asyncio.run(get_surface_confirmation_params(db, str(PID)))


def _models_dir(tmp_path: Path) -> Path:
    models = tmp_path / "B04_PreProcess" / "models"
    models.mkdir(parents=True, exist_ok=True)
    for name, z in (("classification", LAS_Z), (SHEET_ANALYSIS_FILTER, SHEET_Z)):
        for suffix in ("", "_smooth"):
            x = y = np.arange(0.0, 11.0)
            np.savez(
                models / f"dtm_{name}{suffix}.npz",
                x=x,
                y=y,
                z=np.full((11, 11), z),
                valid_mask=np.ones((11, 11), dtype=bool),
            )
    return models


def _z(models: Path, params: dict) -> float:
    sampler = build_surface_sampler(
        models, params["source_filter"], params["method"], params["smooth"]
    )
    z, _ = sampler.sample_xy(np.array([[5.0, 5.0]]))
    return float(z[0])


def test_수치지형도_확인_뒤_설계_지형_sheet_라이다로_되돌림(tmp_path, monkeypatch):
    db = _Db(LAS_STAGE1, _models(las=True))
    _patch_models(monkeypatch, db)
    models = _models_dir(tmp_path)
    assert _z(models, _params(db)) == LAS_Z

    assert _switch(db, "sheet") == (SHEET_ID, "las")
    params = _params(db)
    assert design_terrain(params) == "sheet" and db.confirmed() == SHEET_ID
    assert params[LAS_STASH_KEY] == LAS_STAGE1  # 되돌릴 자리
    # 뒤 단계 sampler 는 모두 stage 1 params 를 읽음 → 수치지형도 표고
    assert _z(models, params) == SHEET_Z
    # 설계 지형이 수치지형도여도 고르기는 둘 다 남음
    assert analysis_terrain_sources(params, models) == ["las", "sheet"]

    assert _switch(db, "las") == (LAS_ID, "sheet")
    params = _params(db)
    assert design_terrain(params) == "las" and db.confirmed() == LAS_ID
    assert {k: params[k] for k in LAS_STAGE1} == LAS_STAGE1
    assert _z(models, params) == LAS_Z


def test_확인_안_하면_그대로(monkeypatch):
    db = _Db(LAS_STAGE1, _models(las=True))
    _patch_models(monkeypatch, db)
    assert _switch(db, None) is None  # 고르기 안 보임 · 고른 값 안 실음
    assert _switch(db, "las") is None  # 지금 설계 지형과 같음
    assert db.writes == 0 and db.confirmed() == LAS_ID


def test_라이다_없는_프로젝트_그대로(monkeypatch, tmp_path):
    db = _Db(SHEET_STAGE1, _models(las=False))
    _patch_models(monkeypatch, db)
    assert _switch(db, "las") is None and _switch(db, "sheet") is None
    assert db.writes == 0 and db.confirmed() == SHEET_ID
    assert analysis_terrain_sources(_params(db), _models_dir(tmp_path)) == ["sheet"]


def _patch_recompute(monkeypatch, tmp_path: Path, db: _Db, failure: str | None):
    _patch_models(monkeypatch, db)
    seen: dict = {}

    class _Pool:
        @asynccontextmanager
        async def acquire(self):
            yield db

    async def latest(_connection, _project_id):
        return {"id": 237, "surface_model_id": LAS_ID}

    async def state(_cursor, _project_id):
        return {"stages": []}

    async def start(*_args):
        return None

    async def run_chain(_project_id, _root, _stored, model_id, selection, *_rest):
        seen["model_id"], seen["selection"] = model_id, dict(selection)
        return {"error": failure} if failure else {"route_id": 238}

    points = [{"x": 0.0, "y": 0.0}, {"x": 5.0, "y": 5.0}]
    monkeypatch.setattr(replan, "get_db_pool", lambda: _Pool())
    monkeypatch.setattr(route_repo, "get_latest_route", latest)
    monkeypatch.setattr(chain, "_planned_route_points_in_project_crs", lambda *_a: points)
    monkeypatch.setattr(route_save, "run_route_save_chain", run_chain)
    monkeypatch.setattr(workflow_state, "get_workflow_state", state)
    monkeypatch.setattr(workflow_state, "start_stage", start)
    return seen


def test_확인_체인이_수치지형도_모델과_params_로_돎(tmp_path, monkeypatch):
    db = _Db(LAS_STAGE1, _models(las=True))
    seen = _patch_recompute(monkeypatch, tmp_path, db, None)
    result = asyncio.run(replan._recompute(PID, tmp_path, "x", None, "sheet"))
    assert "error" not in result
    # 옛 노선의 모델 id(라이다)가 아니라 새 설계 지형 모델 · 종 · 횡단 · 배수 · 유토 체인 params
    assert seen["model_id"] == SHEET_ID
    assert seen["selection"]["source_filter"] == SHEET_ANALYSIS_FILTER
    assert design_terrain(_params(db)) == "sheet"


def test_체인_실패면_설계_지형도_되돌림(tmp_path, monkeypatch):
    db = _Db(LAS_STAGE1, _models(las=True))
    _patch_recompute(monkeypatch, tmp_path, db, "노선 재계산 실패")
    result = asyncio.run(replan._recompute(PID, tmp_path, "x", None, "sheet"))
    assert result == {"error": "노선 재계산 실패"}
    assert design_terrain(_params(db)) == "las" and db.confirmed() == LAS_ID


def test_확인_요청_기본은_설계_지형_그대로():
    request = replan.RouteReplanRequest(vertices=[])
    assert request.terrain is None


def test_B06_구조물_측점은_옛_지형_조건을_버림(tmp_path, monkeypatch):
    snapshot = tmp_path / "B06_Section" / "sampling.json"
    snapshot.parent.mkdir(parents=True)
    snapshot.write_text(
        json.dumps({"filter_key": "classification", "method": "dtm", "smooth": True}),
        encoding="utf-8",
    )

    def stage1(params):
        async def run(_fn, _project_id):
            return {**params, "contour_interval_m": 1.0}

        return run

    monkeypatch.setattr(stations, "run_with_connection", stage1(LAS_STAGE1))
    same = asyncio.run(stations._sampling_conditions(PID, tmp_path))
    assert same["filter_key"] == "classification" and "source" not in same
    monkeypatch.setattr(stations, "run_with_connection", stage1(SHEET_STAGE1))
    moved = asyncio.run(stations._sampling_conditions(PID, tmp_path))
    assert moved["filter_key"] == SHEET_ANALYSIS_FILTER and moved["source"] == "stage1"


def test_뒤_단계는_stage1_params_를_읽음():
    """종 · 횡단 · 배수 · 3D 가 설계 지형 한 곳(stage 1 params)을 보는지 — 배선 고정."""
    readers = [
        "common_util/common_util_drainage_context.py",
        "B06_Section/B06_Section_Router.py",
        "B06_Section/B06_Section_Server_Calc_Prebuild.py",
        "B06_Section/B06_Section_Router_Stations.py",
        "B05_Profile/B05_Profile_Router.py",
        "B03_FileInput/B03_FileInput_Service_Chain.py",
        "B04_PreProcess/B04_PreProcess_Router.py",
    ]
    for path in readers:
        assert "get_surface_confirmation_params" in (ROOT / path).read_text(encoding="utf-8"), path


def test_화면_확인이_고른_지형을_실음():
    page = ROOT / "B05_Profile"
    apply = (page / "B05_Profile_UI_RouteEdit_Apply.ts").read_text(encoding="utf-8")
    main = (page / "B05_Profile_UI_RouteEdit.ts").read_text(encoding="utf-8")
    # 626-1 — 지형 바꿈 = 도구줄 지형 칩(`/route/design-terrain`) · 손 [저장] 초안 = 설계 지형(null)
    assert "terrain: settings.terrain," in apply
    assert "const settings = () => ({ ...params.snap(), terrain: null });" in apply
    assert "saveRecommendedRoute" not in apply  # 옛 추천 [적용] 화면 부름 걷음
    chip = (page / "B05_Profile_UI_RouteEdit_Terrain.ts").read_text(encoding="utf-8")
    assert "/route/design-terrain" in chip
    assert "pickedTerrain" not in main and "terrainSelect" not in main