eomsangdonandClaude Opus 5.5
e629584223
- [확인] 이 고른 지형으로 확정 모델 · stage 1 params 를 바꾼 뒤 재계산 체인을 돎(switch_design_terrain · B04 확정 표시도 따라감) · 체인 실패면 노선과 함께 되돌림 - 수치지형도로 갈 때 라이다 확정값을 stage 1 params 에 적어 둠 — 라이다로 되돌려 [확인]하면 원래 모델 · params - 지형만 바꿔도 [확인]이 다시 셈 · 고르기 안 보이면(라이다 없는 프로젝트) 지형 안 실음 · 고른 값 없으면 지금 설계 지형 · 상태줄에 설계 지형 표시 - 분석 통로 terrain 기본 = 지금 설계 지형(None) · terrain-sources 가 design 을 함께 냄 - B06 구조물 측점 추가가 노선 확정 때 남긴 sampling.json 이 지금 지형과 다르면 stage 1 params 로 뜸(B04 재확정 때도 옛 지형으로 뜨던 흠) Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01UoSr2sqvYtBf1dtQdus88s
88 lines
3.2 KiB
Python
88 lines
3.2 KiB
Python
"""B04 지표면 모델 확정 서비스."""
|
|
|
|
from typing import Any
|
|
from uuid import UUID
|
|
|
|
import aiomysql
|
|
|
|
from B04_PreProcess.B04_PreProcess_Repository import (
|
|
confirm_surface_model,
|
|
list_surface_models,
|
|
)
|
|
from common_util.common_util_surface_confirmation import (
|
|
LAS_STASH_KEY,
|
|
AnalysisTerrain,
|
|
analysis_terrain_params,
|
|
design_terrain,
|
|
get_surface_confirmation_params,
|
|
merge_surface_confirmation_params,
|
|
terrain_changes,
|
|
)
|
|
from common_util.common_util_workflow_state import complete_stage
|
|
|
|
|
|
async def find_surface_model_for_selection(
|
|
connection: aiomysql.Connection,
|
|
project_id: UUID,
|
|
selection: dict[str, Any],
|
|
) -> int:
|
|
"""필터·표현 기본값과 일치하는 COMPLETE 모델 ID를 찾는다."""
|
|
for model in await list_surface_models(connection, project_id):
|
|
generation_params = model.get("generation_params") or {}
|
|
if (
|
|
model["status"] == "COMPLETE"
|
|
and model["model_type"].lower() == str(selection["method"]).lower()
|
|
and str(generation_params.get("source_filter", "")).lower()
|
|
== str(selection["source_filter"]).lower()
|
|
):
|
|
return int(model["id"])
|
|
raise LookupError(
|
|
"자동 확정 기본값과 일치하는 지표면 모델이 없습니다: "
|
|
f"filter={selection['source_filter']}, method={selection['method']}"
|
|
)
|
|
|
|
|
|
async def confirm_surface_selection(
|
|
connection: aiomysql.Connection,
|
|
project_id: UUID,
|
|
model_id: int,
|
|
selection: dict[str, Any],
|
|
) -> dict[str, Any]:
|
|
"""모델 확정, 선택값 스냅샷, WF1 완료를 현재 트랜잭션에서 처리한다."""
|
|
await confirm_surface_model(connection, project_id, model_id)
|
|
normalized = await merge_surface_confirmation_params(
|
|
connection,
|
|
str(project_id),
|
|
selection,
|
|
)
|
|
async with connection.cursor() as cursor:
|
|
await complete_stage(cursor, str(project_id), 1)
|
|
return normalized
|
|
|
|
|
|
async def switch_design_terrain(
|
|
connection: aiomysql.Connection,
|
|
project_id: UUID,
|
|
terrain: AnalysisTerrain | None,
|
|
) -> tuple[int, AnalysisTerrain] | None:
|
|
"""노선 [확인] 때 고른 분석 지형을 설계 지형으로(PLAN 133-2) — 확정 모델 · stage 1 params.
|
|
|
|
뒤 단계(종 · 횡단 · 배수 · 유토 · 3D)는 모두 stage 1 params · 확정 모델을 읽으므로 여기 한 곳만
|
|
바꾼다. 수치지형도로 갈 때 라이다 확정값을 적어 둠(되돌릴 자리). WF1 완료 전이는 안 건드림.
|
|
바뀌면 (새 확정 모델 id, 이전 설계 지형) · 그대로면 None. 트랜잭션은 부르는 쪽.
|
|
"""
|
|
params = await get_surface_confirmation_params(connection, str(project_id))
|
|
target = analysis_terrain_params(params, terrain)
|
|
if not terrain_changes(params, target):
|
|
return None
|
|
model_id = await find_surface_model_for_selection(connection, project_id, target)
|
|
await confirm_surface_model(connection, project_id, model_id)
|
|
previous = design_terrain(params)
|
|
extra = (
|
|
{LAS_STASH_KEY: {key: params[key] for key in ("source_filter", "method", "smooth")}}
|
|
if previous == "las"
|
|
else None
|
|
)
|
|
await merge_surface_confirmation_params(connection, str(project_id), target, extra)
|
|
return model_id, previous
|