"""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, surface: dict[str, Any] | None = None ) -> list[dict[str, float]] | None: """설계용 계획노선을 프로젝트 좌표계(m) 점 목록으로 돌려준다. 없으면 None. 읽기·좌표계 변환·트림·조밀화는 `load_design_route()` 한 곳에서 한다 — 배수유역·유입도 같은 함수를 쓰므로 여기만 트림되는 일이 없다. """ from common_util.common_util_route_geometry import load_design_route planned = load_design_route(project_root, surface) if planned is None: if surface: logger.warning( "자동 설계 체인 중단(설계 노선 없음): 계획노선을 읽지 못했거나 라이다 측량" " 범위와 겹치지 않습니다 — %s", project_root.name, ) return None return [{"x": v.x, "y": v.y} for v in planned.vertices] async def _prepare_drainage_pipes_and_reprofile( project_id: UUID, route_id: int, *, refresh: bool = False ) -> bool: """배수유역 분석 → 기본 관 지점 확정 저장 → 배관 정착 계획선으로 종횡단 재생성. 계획선 1차 로직(`design_pipe_anchored_profile`)은 **저장된 관 지점**을 변화점으로 쓴다. 자동 체인이 이 파일을 만들어 주지 않으면 초기 계획선이 2차 폴백(지반 추종 직선 분할)으로 산출돼 관 자리 틸팅·R이 통째로 빠진다(2026-08-08 사용자 보고). 같은 이유로 B05·B06에 들어왔을 때 배수유역도도 비어 있게 된다 — 여기서 미리 계산해 영구저장소에 남긴다. 실패해도 예외를 던지지 않는다 — 관 없이 만든 계획선이라도 남는 편이 낫다. """ from B04_PreProcess.B04_PreProcess_Router_Basins import put_pipe_points from B04_PreProcess.B04_PreProcess_Router_Watershed import get_primary_region from B06_Section.B06_Section_Repository import get_latest_section_options from B06_Section.B06_Section_Router import regenerate_sections from B06_Section.B06_Section_Schema import SectionRegenerateRequest from config.config_db import get_db_pool from config.config_system import SECTION_CROSS_HALF_WIDTH_M try: # 1) 배수유역 분석(30초 내외). 저장분이 있으면 그대로 쓴다. # 저장분은 노선이 바뀌었는지 보지 않으므로, 노선을 다시 푼 뒤에는 refresh로 부른다. region = await get_primary_region(project_id, refresh=refresh) if isinstance(region, JSONResponse): logger.error( "자동 설계 체인 배수유역 분석 실패(계획선 폴백 유지): project_id=%s status=%s", project_id, region.status_code, ) return False # 2) 기본 관(도로 × 상류 세류선) + 최대 간격 자동 보충을 정본으로 저장한다. pipes = await put_pipe_points(project_id, None) if isinstance(pipes, JSONResponse): logger.error( "자동 설계 체인 관 지점 확정 실패: project_id=%s status=%s", project_id, pipes.status_code, ) return False pipe_count = int(pipes.get("saved_count") or 0) # 3) 저장된 관 자리를 물려 계획선을 다시 산출한다. 반폭은 이미 쓰던 값을 유지한다. pool = get_db_pool() async with pool.acquire() as connection: stored_options = await get_latest_section_options(connection, project_id) half_width = float( (stored_options or {}).get("cross_half_width_m") or SECTION_CROSS_HALF_WIDTH_M ) regenerated = await regenerate_sections( project_id, route_id, SectionRegenerateRequest(cross_half_width_m=half_width) ) if isinstance(regenerated, JSONResponse): logger.error( "자동 설계 체인 배관 정착 계획선 재산출 실패: project_id=%s status=%s", project_id, regenerated.status_code, ) return False logger.info( "자동 설계 체인 배수유역·배관 정착 계획선 완료: project_id=%s route_id=%s 관=%d개", project_id, route_id, pipe_count, ) return True except Exception: logger.exception( "자동 설계 체인 배수유역·관 지점 단계 실패(계획선은 폴백으로 유지): project_id=%s", project_id, ) return False async def run_auto_design_chain( project_id: UUID, surface_model_id: int | None = None ) -> dict[str, Any] | None: """B05 기본 경로 계산·확정 → B06 기본 횡단 설계 확정을 기본값으로 이어 실행한다. WF1 자동 확정 직후 같은 백그라운드 태스크에서 호출된다. 어떤 단계가 실패해도 예외를 밖으로 던지지 않는다 — 로그와 각 단계의 workflow 상태 기록으로 남긴다. 성공하면 초기 분석 완료 메일에 실을 요약(노선 id·연장·측점 수)을 돌려주고, 건너뛰거나 실패하면 None을 돌려준다. """ 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_initial_snapshot import ( clear_designing, mark_designing, save_initial_snapshot, ) from common_util.common_util_storage import resolve_stored_project_path from common_util.common_util_surface_confirmation import get_surface_confirmation_params from config.config_db import get_db_pool pool = get_db_pool() project_root: Path | None = None 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 None # 2) 계획노선 CSV → BP/EP/경유점. 없으면 자동 경로를 세울 근거가 없다. project_root = Path(resolve_stored_project_path(stored_path)) # 이 체인이 끝나기 전에는 B05·B06에 들어가면 안 된다 — 반쯤 계산된 화면을 만지면 # 그 편집이 섞인 채 초기값이 찍힌다(2026-08-29 사용자 확정, CLAUDE.md 5장). mark_designing(project_root) # WF1이 **실제로 확정한** 선택값(stage 1 스냅샷)을 그대로 쓴다. config 기본값을 # 쓰면 LAS 없이 설계한 프로젝트에서 있지도 않은 모델을 가리켜 404로 체인이 # 끊긴다(2026-08-30 실사고). 노선 트림도 이 지표면을 기준으로 한다. async with pool.acquire() as connection: defaults = await get_surface_confirmation_params(connection, str(project_id)) points = _planned_route_points_in_project_crs(project_root, defaults) if not points: logger.warning("자동 설계 체인 중단(계획노선 CSV 없음): project_id=%s", project_id) return None # 3) B05 경로 계산 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 None 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 None # 4.5) 배수유역 분석 → 관 지점 확정 → 배관 정착 계획선 재산출. # B06 확정 전에 끝내야 확정 스냅샷에 배관 정착 계획선이 담긴다. await _prepare_drainage_pipes_and_reprofile(project_id, route_id) # 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 None logger.info( "자동 설계 체인 완료(B05·B06 기본값 확정): project_id=%s route_id=%s", project_id, route_id, ) # 초기값 스냅샷 — 여기가 [초기화]가 되돌릴 기준선이다(CLAUDE.md 5장). # 체인 규약대로 실패는 비치명적이다: 스냅샷이 없으면 [초기화]가 재계산으로 돈다. try: async with pool.acquire() as connection: await save_initial_snapshot(connection, project_root, route_id) except Exception: # noqa: BLE001 — 스냅샷 실패가 체인을 막지는 않는다 logger.exception("초기값 스냅샷 실패: project_id=%s", project_id) return { "route_id": route_id, "length_m": float(solve_result.total_length_m or 0.0), "cross_section_count": solve_result.cross_section_count, } except Exception: # 체인은 업로드·WF1 흐름의 부가 작업이다 — 어떤 예외도 밖으로 던지지 않는다. logger.exception("자동 설계 체인 실패: project_id=%s", project_id) finally: # 성공·실패·중단 어느 쪽이든 문은 연다 — 마커가 남으면 영영 못 들어간다. if project_root is not None: clear_designing(project_root) 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_initial_snapshot import ( clear_designing, discard_initial_snapshot, mark_designing, ) from common_util.common_util_storage import resolve_stored_project_path from common_util.common_util_workflow_state import get_workflow_state from config.config_db import get_db_pool pool = get_db_pool() project_root: Path | None = None try: async with pool.acquire() as connection: latest = await get_latest_route(connection, project_id) stored_path = await get_project_storage_relative_path(connection, project_id) if stored_path: project_root = Path(resolve_stored_project_path(stored_path)) mark_designing(project_root) # 지표면이 바뀌면 옛 초기값은 다른 지형 기준이라 더는 기준이 아니다. 여기서 # 다시 뜨지는 않는다 — 이 체인은 **사용자 입력을 유지한 채** 재계산하므로 그 # 결과를 찍으면 편집이 섞인 가짜 초기값이 된다(2026-08-29 사용자 확정). if discard_initial_snapshot(project_root): logger.info("재확정 체인: 초기값 스냅샷 무효화 (project_id=%s)", 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 # 노선이 바뀌었으므로 배수유역·관 지점도 새 노선 기준으로 다시 만든다 — 옛 관 자리로 # 계획선을 앉히면 전부 어긋나고, 저장분은 노선 변경을 스스로 알지 못한다. await _prepare_drainage_pipes_and_reprofile(project_id, new_route_id, refresh=True) 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) finally: if project_root is not None: clear_designing(project_root)