"""B03 업로드 이후 자동 설계 체인 — WF1 확정 다음을 잇는다. WF1(지표면 분석·자동 확정)이 끝나면 사용자가 화면에 없어도 서버가 이어서 ① B05 기본 경로 계산·확정(계획노선 CSV 기반) ② B06 기본 횡단 설계 확정까지 기본값으로 진행해 영구저장소에 남긴다(2026-08-04 사용자 확정). 이후 사용자가 대시보드에서 B05/B06에 들어오면 저장본을 바로 로딩해 검토·수정만 하면 된다. 원칙: - **수동 이력 보호**: 프로젝트에 경로가 하나라도 있으면 체인을 건너뛴다 — 사용자가 이미 작업한 것을 자동 계산이 덮어쓰면 안 된다. - **단계별 실패 격리**: 각 단계는 해당 라우터가 자기 workflow stage 전이(실패 기록)를 책임진다. 체인은 실패한 단계에서 멈추고 뒤 단계로 오류를 전파하지 않는다 — 사용자는 그 페이지에서 수동으로 이어서 진행할 수 있다. - 라우터 함수를 직접 호출한다(HTTP 재진입 없음). solve/confirm 엔드포인트는 인증 의존성이 없는 순수 함수 시그니처라 서버 내부 호출이 가능하다. """ import logging from pathlib import Path from typing import Any from uuid import UUID from fastapi.responses import JSONResponse logger = logging.getLogger(__name__) def _planned_route_points_in_project_crs(project_root: Path) -> list[dict[str, float]] | None: """계획노선 CSV를 읽어 프로젝트 좌표계(m) 점 목록으로 돌려준다. 없으면 None. B04 `/planned-route` 조회와 같은 규칙 — CSV가 제 좌표계(crs_epsg)를 적어 두었고 프로젝트 좌표계와 다르면 한 번 옮긴다. """ from B04_PreProcess.B04_PreProcess_Engine_Extent import project_epsg_from_prj from common_util.common_util_route_geometry import ( find_planned_route_file, read_planned_route_csv, ) route_file = find_planned_route_file(project_root / "B03_FileInput" / "input") planned = read_planned_route_csv(route_file) if route_file else None if planned is None or len(planned.vertices) < 2: return None target_epsg = project_epsg_from_prj(project_root) points = [(float(vertex.x), float(vertex.y)) for vertex in planned.vertices] source_epsg = f"EPSG:{planned.epsg}" if planned.epsg else target_epsg if source_epsg.upper() != target_epsg.upper(): from pyproj import Transformer transformer = Transformer.from_crs(source_epsg, target_epsg, always_xy=True) points = [transformer.transform(x, y) for x, y in points] return [{"x": x, "y": y} for x, y in points] async def run_auto_design_chain(project_id: UUID, surface_model_id: int | None = None) -> None: """B05 기본 경로 계산·확정 → B06 기본 횡단 설계 확정을 기본값으로 이어 실행한다. WF1 자동 확정 직후 같은 백그라운드 태스크에서 호출된다. 어떤 단계가 실패해도 예외를 밖으로 던지지 않는다 — 로그와 각 단계의 workflow 상태 기록으로 남긴다. """ from B03_FileInput.B03_FileInput_Repository import get_project_storage_relative_path from B05_Profile.B05_Profile_Repository import get_latest_route from B05_Profile.B05_Profile_Router import confirm_latest_route, solve_route from B05_Profile.B05_Profile_Schema import RoutePoint, RouteSolveRequest from B06_Section.B06_Section_Router_Confirm import confirm_sections from common_util.common_util_storage import resolve_stored_project_path from common_util.common_util_surface_confirmation import surface_confirmation_defaults from config.config_db import get_db_pool pool = get_db_pool() try: # 1) 수동 이력 보호 — 경로가 이미 있으면(사용자 작업 또는 이전 자동 실행) 건너뛴다. async with pool.acquire() as connection: existing = await get_latest_route(connection, project_id) stored_path = await get_project_storage_relative_path(connection, project_id) if existing: logger.info( "자동 설계 체인 건너뜀(기존 경로 있음): project_id=%s route_id=%s", project_id, existing.get("id"), ) return # 2) 계획노선 CSV → BP/EP/경유점. 없으면 자동 경로를 세울 근거가 없다. project_root = Path(resolve_stored_project_path(stored_path)) points = _planned_route_points_in_project_crs(project_root) if not points: logger.warning("자동 설계 체인 중단(계획노선 CSV 없음): project_id=%s", project_id) return # 3) B05 경로 계산 — WF1 자동 확정과 같은 config 기본값을 쓴다. defaults = surface_confirmation_defaults() request = RouteSolveRequest( filter_key=str(defaults["source_filter"]), method=str(defaults["method"]), smooth=bool(defaults["smooth"]), surface_model_id=surface_model_id, bp=RoutePoint(**points[0]), ep=RoutePoint(**points[-1]), cp=[ RoutePoint(**point, order=index) for index, point in enumerate(points[1:-1], start=1) ], ) solve_result: Any = await solve_route(project_id, request) if isinstance(solve_result, JSONResponse): logger.error( "자동 설계 체인 중단(B05 경로 계산 실패): project_id=%s status=%s", project_id, solve_result.status_code, ) return route_id = int(solve_result.route_id) logger.info( "자동 설계 체인 B05 경로 계산 완료: project_id=%s route_id=%s length=%.1fm", project_id, route_id, float(solve_result.total_length_m or 0.0), ) # 4) B05 경로 확정 — 데이터만 CONFIRMED, stage 2는 IN_PROGRESS(사용자 검토 대기)로 # 남긴다. 완료 전이는 B06 [확정]이 stage 2·3을 함께 처리한다(2026-08-08 재정의). confirm_result = await confirm_latest_route(project_id, None, mark_stage_complete=False) if isinstance(confirm_result, JSONResponse): logger.error( "자동 설계 체인 중단(B05 경로 확정 실패): project_id=%s status=%s", project_id, confirm_result.status_code, ) return # 5) B06 횡단 설계 확정 — 미지정 측점을 기본값으로 채워 저장. stage 3은 # IN_PROGRESS(스텝바 노란 표시)로 남겨 사용자 검토·확정을 기다린다. sections_result = await confirm_sections( project_id, route_id, None, mark_stage_complete=False ) if isinstance(sections_result, JSONResponse): logger.error( "자동 설계 체인 중단(B06 횡단 확정 실패): project_id=%s route_id=%s status=%s", project_id, route_id, sections_result.status_code, ) return logger.info( "자동 설계 체인 완료(B05·B06 기본값 확정): project_id=%s route_id=%s", project_id, route_id, ) except Exception: # 체인은 업로드·WF1 흐름의 부가 작업이다 — 어떤 예외도 밖으로 던지지 않는다. logger.exception("자동 설계 체인 실패: project_id=%s", project_id) async def run_redesign_chain( project_id: UUID, surface_model_id: int, selection: dict[str, Any], ) -> None: """B04 재확정 후 — **사용자 입력을 유지한 채** 새 지표면 기준으로 B05·B06 재계산·저장. 관리자가 B04에서 다른 지표면 모델로 재확정하면(2026-08-04 사용자 확정) 그 값을 기준으로 다음 페이지들도 함께 갱신돼야 한다. 이때 일반 사용자가 이미 쓰던 설정은 버리지 않는다: - B05: 저장된 stage 2 params(제어점 BP/EP/CP·회피/금지원·경사 옵션·측점 간격 등)를 그대로 쓰고 **지표면(filter/method/smooth/model id)만** 새 확정값으로 바꾼다. - B06: 옛 경로의 측점별 설계(지반유형·단면유형·측구·암 경계)를 chainage 매칭으로 새 경로에 이월하고, 표준단면 설정(data.options)도 함께 넘긴다. 나머지 미지정 측점은 확정 시 기본값으로 채워진다. 경로가 아예 없으면 신규 자동 체인(계획노선 CSV 기본값)으로 되돌아간다. """ from B03_FileInput.B03_FileInput_Repository import get_project_storage_relative_path from B05_Profile.B05_Profile_Repository import get_latest_route from B05_Profile.B05_Profile_Router import confirm_latest_route, solve_route from B05_Profile.B05_Profile_Schema import RouteSolveRequest from B06_Section.B06_Section_Repository import ( get_cross_section_designs, get_longitudinal_section, update_cross_section_design, ) from B06_Section.B06_Section_Router_Confirm import confirm_sections from B06_Section.B06_Section_Schema import SectionConfirmRequest from common_util.common_util_workflow_state import get_workflow_state from config.config_db import get_db_pool _ = get_project_storage_relative_path # 시그니처 정렬용 (사용 안 함) pool = get_db_pool() try: async with pool.acquire() as connection: latest = await get_latest_route(connection, project_id) if not latest: logger.info( "재확정 체인 → 기존 경로 없음, 신규 자동 체인으로: project_id=%s", project_id ) await run_auto_design_chain(project_id, surface_model_id=surface_model_id) return # 1) 사용자 입력 회수 — 마지막 경로 계산의 stage 2 params가 정본이다. old_route_id = int(latest["id"]) async with pool.acquire() as connection: import aiomysql async with connection.cursor(aiomysql.DictCursor) as cursor: state = await get_workflow_state(cursor, str(project_id)) old_longitudinal = await get_longitudinal_section(connection, project_id, old_route_id) old_designs = await get_cross_section_designs(connection, old_route_id) stage2 = next( (s for s in (state or {}).get("stages", []) if int(s.get("stage_no", -1)) == 2), None ) params = (stage2 or {}).get("params") or {} points = params.get("points") or {} options = params.get("options") or {} if not points.get("bp") or not points.get("ep"): logger.warning( "재확정 체인 중단(stage 2 params에 제어점 없음): project_id=%s", project_id ) return # 2) B05 재계산 — 지표면 관련 값만 새 확정 선택으로 교체, 나머지는 사용자 저장분. request = RouteSolveRequest( filter_key=str(selection.get("source_filter") or params.get("filter_key")), method=str(selection.get("method") or params.get("method") or "dtm"), smooth=bool(selection.get("smooth", params.get("smooth", False))), surface_model_id=surface_model_id, algorithm=str(params.get("algorithm") or "dijkstra"), bp=points["bp"], ep=points["ep"], cp=points.get("cp") or [], ap=points.get("ap") or [], fp=points.get("fp") or [], station_interval_m=params.get("station_interval_m"), cross_half_width_m=params.get("cross_half_width_m"), cross_sample_interval_m=params.get("cross_sample_interval_m"), long_sample_interval_m=params.get("long_sample_interval_m"), **{key: options.get(key) for key in ("grade_class",) if options.get(key)}, paved=bool(options.get("paved", False)), terrain_type=str(options.get("terrain_type") or "normal"), main_direction=str(options.get("main_direction") or "auto"), min_curve_radius_m=options.get("min_curve_radius_m"), max_uphill_grade=options.get("max_uphill_grade"), max_downhill_grade=options.get("max_downhill_grade"), min_uphill_grade=options.get("min_uphill_grade"), min_downhill_grade=options.get("min_downhill_grade"), weights=options.get("weights"), allow_avoid_pass_through=bool(options.get("allow_avoid_pass_through", False)), max_grade_pct=params.get("max_grade_pct"), min_vertical_radius_m=params.get("min_vertical_radius_m"), min_tangent_length_m=params.get("min_tangent_length_m"), balance_segment_length_m=params.get("balance_segment_length_m"), start_elevation_offset_m=params.get("start_elevation_offset_m"), end_elevation_offset_m=params.get("end_elevation_offset_m"), ) solve_result: Any = await solve_route(project_id, request) if isinstance(solve_result, JSONResponse): logger.error( "재확정 체인 중단(B05 재계산 실패): project_id=%s status=%s", project_id, solve_result.status_code, ) return new_route_id = int(solve_result.route_id) logger.info( "재확정 체인 B05 재계산 완료: project_id=%s %s→%s", project_id, old_route_id, new_route_id, ) # 3) 옛 측점별 사용자 설계를 chainage 매칭으로 새 경로에 이월(비치명적). carried = 0 try: async with pool.acquire() as connection: await connection.begin() try: for record in old_designs: design = record.get("design") if not isinstance(design, dict): continue await update_cross_section_design( connection, route_id=new_route_id, chainage_m=float(record["chainage_m"]), design=design, project_id=project_id, ) carried += 1 await connection.commit() except Exception: await connection.rollback() raise except Exception: logger.exception( "재확정 체인 — 옛 설계 이월 실패(계속 진행): project_id=%s", project_id ) logger.info("재확정 체인 설계 이월: project_id=%s %d건", project_id, carried) # 4) B05 확정 → B06 확정(옛 표준단면 설정 이월, 미지정 측점 기본값 채움). # 재확정 후에도 stage 2·3은 IN_PROGRESS로 남겨 사용자 재검토를 받는다. confirm_result = await confirm_latest_route(project_id, None, mark_stage_complete=False) if isinstance(confirm_result, JSONResponse): logger.error( "재확정 체인 중단(B05 확정 실패): project_id=%s status=%s", project_id, confirm_result.status_code, ) return old_options = ((old_longitudinal or {}).get("data") or {}).get("options") or {} section_request = ( SectionConfirmRequest(standard_cross_section=old_options["standard_cross_section"]) if old_options.get("standard_cross_section") else None ) sections_result = await confirm_sections( project_id, new_route_id, section_request, mark_stage_complete=False ) if isinstance(sections_result, JSONResponse): logger.error( "재확정 체인 중단(B06 확정 실패): project_id=%s status=%s", project_id, sections_result.status_code, ) return logger.info( "재확정 체인 완료: project_id=%s route %s→%s (설계 %d건 이월)", project_id, old_route_id, new_route_id, carried, ) except Exception: logger.exception("재확정 체인 실패: project_id=%s", project_id)