Aislo/common_util/common_util_surface_confirmation.py

"""WF1 지표면 확정 선택값의 기본값·DB 스냅샷 처리."""

import json
from pathlib import Path
from typing import Any, Literal

import aiomysql

from config.config_system import (
    SHEET_SURFACE_DEFAULT_METHOD,
    SURFACE_CONFIRM_DEFAULT_FILTER,
    SURFACE_CONFIRM_DEFAULT_METHOD,
    SURFACE_CONFIRM_DEFAULT_SMOOTH,
    SURFACE_CONTOUR_INTERVAL_M,
)

#: B05 계획노선 편집 모달의 분석 지형 — las = 확정 지표면 그대로 · sheet = 수치지형도 서피스
AnalysisTerrain = Literal["las", "sheet"]
#: 수치지형도 분석 지형 — LAS 없는 프로젝트 자동 확정(WF1)과 같은 모델
SHEET_ANALYSIS_FILTER = f"sheet_{SHEET_SURFACE_DEFAULT_METHOD}"
#: 설계 지형을 수치지형도로 바꿀 때 적어 두는 라이다 확정값 — 라이다로 되돌릴 자리(PLAN 133-2)
LAS_STASH_KEY = "las_selection"
_SURFACE_KEYS = ("source_filter", "method", "smooth")

SURFACE_CONFIRM_PARAM_KEYS = (
    "source_filter",
    "method",
    "smooth",
    "contour_interval_m",
)


def surface_confirmation_defaults() -> dict[str, Any]:
    """현재 config에 설정된 지표면 자동 확정 기본값을 반환한다."""
    return {
        "source_filter": SURFACE_CONFIRM_DEFAULT_FILTER,
        "method": SURFACE_CONFIRM_DEFAULT_METHOD,
        "smooth": SURFACE_CONFIRM_DEFAULT_SMOOTH,
        "contour_interval_m": SURFACE_CONTOUR_INTERVAL_M,
    }


def _decode_params(value: Any) -> dict[str, Any]:
    if isinstance(value, dict):
        return dict(value)
    if isinstance(value, str) and value:
        decoded = json.loads(value)
        return dict(decoded) if isinstance(decoded, dict) else {}
    return {}


async def get_surface_confirmation_params(
    connection: aiomysql.Connection,
    project_id: str,
) -> dict[str, Any]:
    """stage 1 스냅샷을 우선하고, 없으면 config 기본값으로 보완한다."""
    async with connection.cursor(aiomysql.DictCursor) as cursor:
        await cursor.execute(
            """
            SELECT params
            FROM project_workflow_stages
            WHERE project_id = %s AND stage_no = 1
            """,
            (project_id,),
        )
        row = await cursor.fetchone()

    resolved = surface_confirmation_defaults()
    params = _decode_params(row.get("params") if row else None)
    for key in SURFACE_CONFIRM_PARAM_KEYS:
        if key in params and params[key] is not None:
            resolved[key] = params[key]
    if isinstance(params.get(LAS_STASH_KEY), dict):
        resolved[LAS_STASH_KEY] = params[LAS_STASH_KEY]
    return resolved


def _is_sheet(params: dict[str, Any]) -> bool:
    return str(params.get("source_filter") or "").startswith("sheet_")


def design_terrain(params: dict[str, Any]) -> AnalysisTerrain:
    """지금 설계 지형 — 확정 params 가 수치지형도면 sheet."""
    return "sheet" if _is_sheet(params) else "las"


def analysis_terrain_params(
    params: dict[str, Any], terrain: AnalysisTerrain | None
) -> dict[str, Any]:
    """고른 지형의 지표면 params — None · 지금 설계 지형과 같으면 확정 params 그대로.

    설계 지형이 수치지형도여도 적어 둔 라이다 확정값(`LAS_STASH_KEY`)이 있으면 las 로 돌아감.
    라이다 없는 프로젝트는 무엇을 받아도 확정 params 그대로.
    """
    if terrain is None or terrain == design_terrain(params):
        return params
    if terrain == "sheet":
        return {**params, "source_filter": SHEET_ANALYSIS_FILTER, "method": "dtm", "smooth": True}
    stash = params.get(LAS_STASH_KEY)
    if not isinstance(stash, dict) or not stash.get("source_filter"):
        return params
    return {**params, **{key: stash[key] for key in _SURFACE_KEYS if key in stash}}


def terrain_changes(params: dict[str, Any], target: dict[str, Any]) -> bool:
    """지표면(필터 · 표현 · 평활)이 바뀌는가."""
    return any(params.get(key) != target.get(key) for key in _SURFACE_KEYS)


def analysis_terrain_sources(params: dict[str, Any], models_dir: Path) -> list[str]:
    """고를 수 있는 분석 지형 — 라이다 확정값(지금 것 · 적어 둔 것)과 수치지형도가 둘 다 있으면 둘."""
    if _is_sheet(params):
        return ["las", "sheet"] if isinstance(params.get(LAS_STASH_KEY), dict) else ["sheet"]
    if (Path(models_dir) / f"dtm_{SHEET_ANALYSIS_FILTER}.npz").is_file():
        return ["las", "sheet"]
    return ["las"]


async def update_contour_interval_param(
    connection: aiomysql.Connection,
    project_id: str,
    contour_interval_m: float,
) -> None:
    """stage 1 params의 등고선 간격만 갱신한다 (B05 재적용 영속화)."""
    async with connection.cursor(aiomysql.DictCursor) as cursor:
        await cursor.execute(
            """
            SELECT params
            FROM project_workflow_stages
            WHERE project_id = %s AND stage_no = 1
            FOR UPDATE
            """,
            (project_id,),
        )
        row = await cursor.fetchone()
        if row is None:
            raise LookupError("WF1 단계 상태를 찾을 수 없습니다.")
        params = _decode_params(row.get("params") if row else None)
        params["contour_interval_m"] = float(contour_interval_m)
        await cursor.execute(
            """
            UPDATE project_workflow_stages
            SET params = %s
            WHERE project_id = %s AND stage_no = 1
            """,
            (json.dumps(params, ensure_ascii=False), project_id),
        )


async def merge_surface_confirmation_params(
    connection: aiomysql.Connection,
    project_id: str,
    selection: dict[str, Any],
    extra: dict[str, Any] | None = None,
) -> dict[str, Any]:
    """기존 stage 1 params에 확정 선택값 전체를 병합 저장한다. `extra` 는 함께 적을 칸."""
    normalized = {
        "source_filter": str(selection["source_filter"]),
        "method": str(selection["method"]),
        "smooth": bool(selection["smooth"]),
        "contour_interval_m": float(selection["contour_interval_m"]),
    }
    async with connection.cursor(aiomysql.DictCursor) as cursor:
        await cursor.execute(
            """
            SELECT params
            FROM project_workflow_stages
            WHERE project_id = %s AND stage_no = 1
            FOR UPDATE
            """,
            (project_id,),
        )
        row = await cursor.fetchone()
        if row is None:
            raise LookupError("WF1 단계 상태를 찾을 수 없습니다.")
        params = _decode_params(row.get("params") if row else None)
        params.update(normalized)
        params.update(extra or {})
        await cursor.execute(
            """
            UPDATE project_workflow_stages
            SET params = %s
            WHERE project_id = %s AND stage_no = 1
            """,
            (json.dumps(params, ensure_ascii=False), project_id),
        )
    return normalized