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}"

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]
    return resolved


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


def analysis_terrain_params(params: dict[str, Any], terrain: AnalysisTerrain) -> dict[str, Any]:
    """편집 모달 분석에 쓸 지표면 params — 확정 params 는 안 고침(정본 · 종 · 횡단 그대로)."""
    if terrain != "sheet" or _is_sheet(params):
        return params
    return {**params, "source_filter": SHEET_ANALYSIS_FILTER, "method": "dtm", "smooth": True}


def analysis_terrain_sources(params: dict[str, Any], models_dir: Path) -> list[str]:
    """고를 수 있는 분석 지형 — 확정이 LAS 이고 수치지형도 서피스가 있을 때만 둘."""
    if _is_sheet(params):
        return ["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],
) -> dict[str, Any]:
    """기존 stage 1 params에 확정 선택값 전체를 병합 저장한다."""
    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)
        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