133 lines
4.2 KiB
Python
133 lines
4.2 KiB
Python
"""WF1 지표면 확정 선택값의 기본값·DB 스냅샷 처리."""
|
|
|
|
import json
|
|
from typing import Any
|
|
|
|
import aiomysql
|
|
|
|
from config.config_system import (
|
|
SURFACE_CONFIRM_DEFAULT_FILTER,
|
|
SURFACE_CONFIRM_DEFAULT_METHOD,
|
|
SURFACE_CONFIRM_DEFAULT_SMOOTH,
|
|
SURFACE_CONTOUR_INTERVAL_M,
|
|
)
|
|
|
|
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
|
|
|
|
|
|
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
|