"""B06 종횡단 생성 FastAPI 라우터.""" import asyncio import json import logging from pathlib import Path from typing import Any from uuid import UUID import aiomysql from fastapi import APIRouter, Depends from fastapi.responses import JSONResponse from B03_FileInput.B03_FileInput_Repository import get_project_storage_relative_path from B05_Profile.B05_Profile_Engine_Grade import GradeDesignOptions, resolve_grade_options from B05_Profile.B05_Profile_Engine_Grade_Profile import rebuild_alignment_profile from B05_Profile.B05_Profile_Engine_Sections import ( cross_filename, prune_stale_cross_files, run_section_generation, ) from B05_Profile.B05_Profile_Engine_Sections_Core import SectionGenerationOptions from B06_Section.B06_Section_Engine_Design import compute_cross_design from B06_Section.B06_Section_Repository import ( count_cross_sections, create_longitudinal_section, delete_sections_for_route, get_confirmed_route_context, get_cross_section_designs, get_latest_section_options, get_longitudinal_section, get_project_standard_cross_section, get_route_generation_source, insert_cross_sections, list_recent_company_projects, update_cross_section_design, ) from B06_Section.B06_Section_Schema import ( CompanyStandardListResponse, CompanyStandardProject, CompanyStandardResponse, CrossDesignPreviewRequest, CrossDesignPreviewResponse, CrossDesignRequest, CrossDesignResponse, HaulEquipmentLimit, SectionContextResponse, SectionDetailResponse, SectionOptionDefaults, SectionRegenerateRequest, SectionSummaryResponse, ) from common_util.common_util_auth import verify_session from common_util.common_util_route_profile import design_elevation_from_longitudinal from common_util.common_util_storage import resolve_stored_project_path from common_util.common_util_surface_confirmation import get_surface_confirmation_params from common_util.common_util_workflow_state import get_workflow_state from config.config_db import get_db_pool from config.config_system import ( EARTHWORK_CONVERSION_FACTORS, EARTHWORK_HAUL_EQUIPMENT_LIMITS_M, FOREST_ROAD_MIN_WIDTH_M, NATURAL_SPOIL_MIN_GROUND_SLOPE, SECTION_VERTICAL_EXAGGERATION, STANDARD_CROSS_SECTION, STANDARD_ROCK_BOUNDARY_DEFAULT_OFFSET_M, STANDARD_ROCK_BOUNDARY_STEP_M, ) logger = logging.getLogger(__name__) router = APIRouter(prefix="/api/projects", tags=["B06 Profile Cross"]) @router.get("/{project_id}/sections/context", response_model=SectionContextResponse) async def get_section_context(project_id: UUID) -> SectionContextResponse | JSONResponse: """최신 확정 경로와 B04 확정값, config 기반 생성 기본값을 반환한다.""" pool = get_db_pool() try: async with pool.acquire() as connection: route_context = await get_confirmed_route_context(connection, project_id) surface_params = await get_surface_confirmation_params(connection, str(project_id)) # 임도 종류(projects.road_type) — B05가 계획선 법정 기준을 정하는 데 쓴다. async with connection.cursor() as cursor: await cursor.execute( "SELECT road_type FROM projects WHERE id = %s", (str(project_id),) ) row = await cursor.fetchone() road_type = row[0] if row else None defaults = SectionGenerationOptions() return SectionContextResponse( project_id=str(project_id), route_id=route_context["route_id"] if route_context else None, filter_key=surface_params["source_filter"] if route_context else None, method=surface_params["method"] if route_context else None, smooth=bool(surface_params["smooth"]) if route_context else None, crs_epsg=route_context["crs_epsg"] if route_context else None, road_type=road_type, defaults=SectionOptionDefaults( station_interval_m=defaults.station_interval_m, cross_half_width_m=defaults.cross_half_width_m, cross_sample_interval_m=defaults.cross_sample_interval_m, long_sample_interval_m=defaults.long_sample_interval_m, vertical_exaggeration=SECTION_VERTICAL_EXAGGERATION, ), standard_cross_section=STANDARD_CROSS_SECTION, rock_boundary_default_offset_m=STANDARD_ROCK_BOUNDARY_DEFAULT_OFFSET_M, rock_boundary_step_m=STANDARD_ROCK_BOUNDARY_STEP_M, earthwork_conversion=EARTHWORK_CONVERSION_FACTORS, haul_equipment_limits=[ HaulEquipmentLimit(key=key, max_distance_m=limit) for key, limit in EARTHWORK_HAUL_EQUIPMENT_LIMITS_M ], natural_spoil_min_ground_slope=NATURAL_SPOIL_MIN_GROUND_SLOPE, ) except Exception: logger.exception("B06 종횡단 컨텍스트 조회 실패: project_id=%s", project_id) return JSONResponse( status_code=500, content={"status": "error", "message": "종횡단 컨텍스트 조회 중 오류가 발생했습니다."}, ) @router.get("/{project_id}/sections/road-widths") async def get_forest_road_min_widths(project_id: UUID) -> dict[str, dict[str, float]]: """B05 측점 가로선에 적용할 임도 등급별 법정 최소너비를 반환한다.""" return {"forest_road_min_width_m": FOREST_ROAD_MIN_WIDTH_M} # 아래 두 엔드포인트는 `/sections/{route_id}`(int)보다 먼저 선언해 라우팅 충돌을 막는다. @router.get("/{project_id}/sections/company-standards", response_model=CompanyStandardListResponse) async def list_company_standards( project_id: UUID, session: dict[str, Any] = Depends(verify_session) ) -> CompanyStandardListResponse: """같은 회사의 최근 프로젝트 5개를 반환한다(설계값 보유 여부 무관).""" company_id = session.get("company_id") if company_id is None: return CompanyStandardListResponse(projects=[]) pool = get_db_pool() async with pool.acquire() as connection: rows = await list_recent_company_projects(connection, company_id, project_id) return CompanyStandardListResponse( projects=[ CompanyStandardProject(project_id=str(row["project_id"]), name=row["name"]) for row in rows ] ) @router.get( "/{project_id}/sections/company-standards/{source_project_id}", response_model=CompanyStandardResponse, ) async def get_company_standard( project_id: UUID, source_project_id: UUID, session: dict[str, Any] = Depends(verify_session), ) -> CompanyStandardResponse | JSONResponse: """특정 프로젝트의 표준횡단 설정값을 미리보기용으로 반환한다(회사 스코프 강제).""" company_id = session.get("company_id") if company_id is None: return JSONResponse( status_code=404, content={"status": "error", "message": "설계값을 찾을 수 없습니다."} ) pool = get_db_pool() async with pool.acquire() as connection: standard = await get_project_standard_cross_section( connection, company_id, source_project_id ) if standard is None: return JSONResponse( status_code=404, content={"status": "error", "message": "설계값을 찾을 수 없습니다."} ) return CompanyStandardResponse( project_id=str(source_project_id), standard_cross_section=standard ) @router.get("/{project_id}/sections/{route_id}", response_model=SectionSummaryResponse) async def get_sections(project_id: UUID, route_id: int) -> SectionSummaryResponse | JSONResponse: """경로의 종단면 요약을 조회한다.""" pool = get_db_pool() try: async with pool.acquire() as connection: longitudinal = await get_longitudinal_section(connection, project_id, route_id) cross_section_count = await count_cross_sections(connection, route_id) data = longitudinal.get("data") if longitudinal else None length_m = data.get("length_m") if isinstance(data, dict) else None return SectionSummaryResponse( project_id=str(project_id), route_id=route_id, longitudinal=longitudinal, length_m=length_m, cross_section_count=cross_section_count, ) except Exception: logger.exception("B06 종횡단 조회 실패: project_id=%s", project_id) return JSONResponse( status_code=500, content={"status": "error", "message": "종횡단 조회 중 오류가 발생했습니다."}, ) def _read_section_detail(project_root: Path, longitudinal_file_path: str) -> dict: """검증된 프로젝트 루트 안의 종단 및 횡단 JSON을 읽는다.""" root = project_root.resolve() longitudinal_path = (root / longitudinal_file_path).resolve() if root not in longitudinal_path.parents: raise ValueError("종단면 파일 경로가 프로젝트 저장소를 벗어났습니다.") if not longitudinal_path.is_file(): raise FileNotFoundError("종단면 상세 파일을 찾을 수 없습니다.") stage_root = longitudinal_path.parent.parent cross_dir = stage_root / "cross_sections" if not cross_dir.is_dir(): raise FileNotFoundError("횡단면 상세 파일을 찾을 수 없습니다.") longitudinal = json.loads(longitudinal_path.read_text(encoding="utf-8")) stations = longitudinal.get("stations") if isinstance(longitudinal, dict) else None valid_names = prune_stale_cross_files(cross_dir, stations if isinstance(stations, list) else []) cross_sections = [ json.loads(path.read_text(encoding="utf-8")) for path in sorted(cross_dir.glob("cross_*.json")) if not valid_names or path.name in valid_names ] if not isinstance(longitudinal, dict) or not all( isinstance(section, dict) for section in cross_sections ): raise ValueError("종횡단 상세 파일 형식이 올바르지 않습니다.") return {"longitudinal": longitudinal, "cross_sections": cross_sections} def _read_balloon_offsets(row: dict[str, Any]) -> dict[str, list[float]] | None: """확정 시 저장해 둔 유토곡선 balloon 위치를 꺼낸다. 없거나 형태가 깨졌으면 None.""" data = row.get("data") if isinstance(data, (str, bytes)): try: data = json.loads(data) except (ValueError, TypeError): return None if not isinstance(data, dict): return None offsets = (data.get("mass_haul") or {}).get("balloon_offsets") if not isinstance(offsets, dict): return None cleaned: dict[str, list[float]] = {} for key, value in offsets.items(): if isinstance(value, (list, tuple)) and len(value) == 2: try: cleaned[str(key)] = [float(value[0]), float(value[1])] except (TypeError, ValueError): continue return cleaned or None @router.get("/{project_id}/sections/{route_id}/detail", response_model=SectionDetailResponse) async def get_section_detail( project_id: UUID, route_id: int ) -> SectionDetailResponse | JSONResponse: """경로의 SVG 렌더링용 종단·횡단 원시 샘플을 반환한다.""" pool = get_db_pool() try: async with pool.acquire() as connection: longitudinal = await get_longitudinal_section(connection, project_id, route_id) if not longitudinal: return JSONResponse( status_code=404, content={"status": "error", "message": "종횡단 상세 결과가 없습니다."}, ) stored_path = await get_project_storage_relative_path(connection, project_id) designs = await get_cross_section_designs(connection, route_id) project_root = Path(resolve_stored_project_path(stored_path)) detail = await asyncio.to_thread( _read_section_detail, project_root, str(longitudinal["longitudinal_file_path"]), ) # DB에만 있는 잠정 설계 지정을 chainage 근사로 각 횡단에 얹어 화면 복원을 돕는다. for record in designs: for section in detail["cross_sections"]: if abs(float(section.get("chainage_m", 0.0)) - record["chainage_m"]) < 0.01: section["design"] = record["design"] break # 지정값이 없는 측점은 기본값(토사/좌절토)으로 즉석 계산해 프리뷰로 채운다. # (미저장 프리뷰: 실제 저장은 사용자가 카드를 조작하거나 확정할 때 이뤄진다.) await asyncio.to_thread( _attach_default_designs, detail["longitudinal"], detail["cross_sections"] ) return SectionDetailResponse(**detail, balloon_offsets=_read_balloon_offsets(longitudinal)) except FileNotFoundError as exc: return JSONResponse(status_code=404, content={"status": "error", "message": str(exc)}) except (OSError, ValueError, json.JSONDecodeError) as exc: logger.warning( "B06 종횡단 상세 파일 조회 실패: project_id=%s route_id=%s error=%s", project_id, route_id, exc, ) return JSONResponse( status_code=500, content={"status": "error", "message": "종횡단 상세 파일을 읽지 못했습니다."}, ) except Exception: logger.exception( "B06 종횡단 상세 조회 실패: project_id=%s route_id=%s", project_id, route_id ) return JSONResponse( status_code=500, content={"status": "error", "message": "종횡단 상세 조회 중 오류가 발생했습니다."}, ) def _regeneration_grade_options(stage_params: dict[str, Any] | None) -> GradeDesignOptions | None: """재생성 시 종단 계획선 기준을 확정 당시 저장 파라미터(stage 2)로 재구성한다. grade_options 없이 재생성하면 종단 계획선(profile_alignment)이 통째로 소실돼 계획 횡단도선·유토곡선·테이블이 사라진다(2026-08-06 사용자 보고 — 옛 재계산 버튼 사고의 원인). B04 파일입력→B06 자동 계산 파이프라인이 쓰는 같은 엔진 경로를 재활용한다. """ if not stage_params: return None options = stage_params.get("options") or {} grade_class = options.get("grade_class") if not grade_class: return None stored = { key: stage_params.get(key) for key in ( "max_grade_pct", "min_vertical_radius_m", "min_tangent_length_m", "balance_segment_length_m", "start_elevation_offset_m", "end_elevation_offset_m", ) if stage_params.get(key) is not None } # 설계속도는 B05가 저장해 둔 값을 그대로 따른다 — 없으면 종류별 기본(20). speed = options.get("design_speed_kph") return resolve_grade_options( str(grade_class), terrain_type=str(options.get("terrain_type") or "normal"), paved=bool(options.get("paved", False)), design_speed_kph=int(speed) if speed is not None else None, stored=stored, ) def _regeneration_options( stored_options: dict[str, Any] | None, stage_params: dict[str, Any] | None, cross_half_width_m: float, ) -> SectionGenerationOptions: """DB 저장 옵션(단일 소스) → stage 2 params → config 순으로 유지하고 반폭만 교체한다.""" defaults = SectionGenerationOptions() stored = stored_options or {} params = stage_params or {} def pick(key: str, default: float) -> float: return stored.get(key) or params.get(key) or default return SectionGenerationOptions( station_interval_m=pick("station_interval_m", defaults.station_interval_m), cross_half_width_m=cross_half_width_m, cross_sample_interval_m=pick("cross_sample_interval_m", defaults.cross_sample_interval_m), long_sample_interval_m=pick("long_sample_interval_m", defaults.long_sample_interval_m), include_endpoint=defaults.include_endpoint, ) @router.post("/{project_id}/sections/{route_id}/regenerate", response_model=SectionDetailResponse) async def regenerate_sections( project_id: UUID, route_id: int, request: SectionRegenerateRequest ) -> SectionDetailResponse | JSONResponse: """표시 옵션의 횡단 반폭으로 종횡단을 재생성해 저장하고 상세를 반환한다.""" pool = get_db_pool() try: async with pool.acquire() as connection: source = await get_route_generation_source(connection, project_id, route_id) if not source: return JSONResponse( status_code=404, content={"status": "error", "message": "재생성할 경로가 없습니다."}, ) surface_params = await get_surface_confirmation_params(connection, str(project_id)) stored_path = await get_project_storage_relative_path(connection, project_id) stored_options = await get_latest_section_options(connection, project_id) async with connection.cursor(aiomysql.DictCursor) as cursor: workflow = await get_workflow_state(cursor, str(project_id)) route_stage = next( (stage for stage in workflow["stages"] if stage["stage_no"] == 2), None, ) project_root = Path(resolve_stored_project_path(stored_path)) crs_epsg = source["crs_epsg"] stage_params = route_stage.get("params") if route_stage else None sections = await asyncio.to_thread( run_section_generation, project_root, source["route_data_path"], surface_params["source_filter"], surface_params["method"], bool(surface_params["smooth"]), options=_regeneration_options( stored_options, stage_params, request.cross_half_width_m, ), # 계획선까지 함께 재계산·저장 — 없으면 계획 횡단도선·유토곡선이 사라진다. grade_options=_regeneration_grade_options(stage_params), crs=f"EPSG:{crs_epsg}" if crs_epsg is not None else None, ) await connection.begin() try: await delete_sections_for_route(connection, route_id) await create_longitudinal_section( connection, project_id=project_id, route_id=route_id, data=sections["longitudinal"]["data"], longitudinal_file_path=sections["longitudinal"]["file_path"], ) await insert_cross_sections( connection, project_id=project_id, route_id=route_id, sections=sections["cross_sections"], ) await connection.commit() except Exception: await connection.rollback() raise result = sections["result"] return SectionDetailResponse( longitudinal=result["longitudinal"], cross_sections=result["cross_sections"] ) except FileNotFoundError as exc: return JSONResponse(status_code=404, content={"status": "error", "message": str(exc)}) except ValueError as exc: return JSONResponse(status_code=400, content={"status": "error", "message": str(exc)}) except Exception: logger.exception("B06 종횡단 재생성 실패: project_id=%s route_id=%s", project_id, route_id) return JSONResponse( status_code=500, content={"status": "error", "message": "종횡단 재생성 처리 중 오류가 발생했습니다."}, ) def _read_cross_design_inputs( project_root: Path, longitudinal_file_path: str, chainage_m: float ) -> tuple[list[dict], float | None, bool]: """측점 하나의 지반 샘플·계획고·포장 제안 여부를 파일에서 읽는다 (경로 이탈 검증 포함).""" root = project_root.resolve() longitudinal_path = (root / longitudinal_file_path).resolve() if root not in longitudinal_path.parents: raise ValueError("종단면 파일 경로가 프로젝트 저장소를 벗어났습니다.") if not longitudinal_path.is_file(): raise FileNotFoundError("종단면 상세 파일을 찾을 수 없습니다.") longitudinal = json.loads(longitudinal_path.read_text(encoding="utf-8")) design_elevation = design_elevation_from_longitudinal(longitudinal, chainage_m) pavement_suggested = _pavement_suggestions(longitudinal).get(round(chainage_m, 3), False) cross_dir = longitudinal_path.parent.parent / "cross_sections" cross_path = (cross_dir / cross_filename(chainage_m)).resolve() if cross_dir.resolve() not in cross_path.parents or not cross_path.is_file(): raise FileNotFoundError("해당 측점의 횡단 상세 파일을 찾을 수 없습니다.") cross = json.loads(cross_path.read_text(encoding="utf-8")) samples = cross.get("samples") if isinstance(cross, dict) else None if not isinstance(samples, list): raise ValueError("횡단 상세 파일 형식이 올바르지 않습니다.") return samples, design_elevation, pavement_suggested def _pavement_suggestions(longitudinal: dict[str, Any]) -> dict[float, bool]: """측점별 포장 제안(B05 solve가 법정 경사 기준으로 판정) 매핑을 만든다.""" stations = longitudinal.get("stations") mapping: dict[float, bool] = {} if isinstance(stations, list): for station in stations: suggested = station.get("pavement_suggested") if isinstance(suggested, bool): mapping[round(float(station.get("chainage_m", 0.0)), 3)] = suggested return mapping def _default_section_modes(longitudinal: dict[str, Any]) -> dict[float, str]: """측점별 기본 단면유형 매핑: 상단측(uphill_side)이 절토측이 되는 편절편성. B05 solve가 자동 판정하고 사용자가 3D 램프로 바꾼 값(확정 시 정본 병합)을 그대로 소비한다. 판정 불가 측점은 매핑에서 빠지고 호출부가 좌절토로 폴백한다. """ stations = longitudinal.get("stations") mapping: dict[float, str] = {} if isinstance(stations, list): for station in stations: side = station.get("uphill_side") if side in ("left", "right"): mapping[round(float(station.get("chainage_m", 0.0)), 3)] = f"{side}_cut" return mapping def _attach_default_designs( longitudinal: dict[str, Any], cross_sections: list[dict[str, Any]] ) -> None: """지정 설계가 없는 횡단에 기본값(리핑암 + 암반 경계 0.5m + 상단측 절토) 프리뷰를 얹는다. detail 조회가 이미 읽어온 samples와 종단 계획선을 그대로 써서 추가 파일 I/O 없이 전 측점 프리뷰를 만든다(미저장). 계산 불가 측점은 건너뛴다. 기본 지반을 토사가 아니라 **리핑암 + 지표 아래 0.5m 암반 경계**로 두는 이유(2026-08-03 사용자 확정): 산지 절토는 대부분 표토 아래에서 암이 나오므로, 전량 토사 가정은 물량이 낙관적으로 나온다. 지표 0.5m까지 토사·그 아래 리핑암인 2단 단면이 안전한 출발값이고, 발파암이 있으면 사용자가 B06에서 측점별로 고친다. """ default_modes = _default_section_modes(longitudinal) pavement = _pavement_suggestions(longitudinal) for section in cross_sections: if section.get("design"): continue try: chainage_m = float(section.get("chainage_m", 0.0)) suggested = pavement.get(round(chainage_m, 3), False) design = compute_cross_design( section.get("samples", []), design_elevation_from_longitudinal(longitudinal, chainage_m), ground_type="ripping_rock", section_mode=default_modes.get(round(chainage_m, 3), "left_cut"), paved=suggested, rock_boundary_offset_m=STANDARD_ROCK_BOUNDARY_DEFAULT_OFFSET_M, ) design["status"] = "provisional" design["pavement_suggested"] = suggested section["design"] = design except (ValueError, KeyError): continue #: 계획선 프리뷰 응답에 실을 설계 필드 — 유토곡선 계산(면적·지반유형·자연방토)과 #: 계획고 변경 감지에 쓰는 것만. 나머지(설계선 좌표 등)는 B06이 상세를 받을 때 온다. _PREVIEW_DESIGN_FIELDS = ( "ground_type", "roadbed_width_m", "cut_area_m2", "cut_soil_area_m2", "cut_rock_area_m2", "cut_rock_kind", "fill_area_m2", "fill_ground_slope", "design_elevation_m", ) def _recompute_designs_for_alignment( longitudinal: dict[str, Any], cross_sections: list[dict[str, Any]], stored_designs: list[dict[str, Any]], standard: dict[str, Any] | None, rock_boundary_offsets: dict[str, float] | None = None, ) -> None: """새 계획선 기준으로 전 측점 횡단 설계를 다시 계산해 얹는다(메모리 프리뷰). 사용자가 이미 고른 지반유형·단면유형·측구·암 경계는 **그대로 유지**하고 계획고만 새 선형 값으로 바꿔 넣는다 — 계획선을 끌었다고 측점 선택이 초기화되면 안 된다. 지정이 없는 측점은 기본값(리핑암 + 암반 경계 0.5m + 상단측 절토)으로 계산한다. `rock_boundary_offsets`(B06 세션값)가 오면 DB 저장분보다 우선한다. """ default_modes = _default_section_modes(longitudinal) pavement = _pavement_suggestions(longitudinal) stored_by_chainage = { round(float(record["chainage_m"]), 3): (record.get("design") or {}) for record in stored_designs } # 세션 오프셋 키는 프론트 표기(소수 자릿수)가 제각각일 수 있어 수치로 정규화한다. session_offsets: dict[float, float] = {} for raw_key, offset in (rock_boundary_offsets or {}).items(): try: session_offsets[round(float(raw_key), 3)] = float(offset) except (TypeError, ValueError): continue for section in cross_sections: chainage_m = float(section.get("chainage_m", 0.0)) key = round(chainage_m, 3) stored = stored_by_chainage.get(key) or {} suggested = pavement.get(key, False) rock_offset = session_offsets.get( key, stored.get("rock_boundary_offset_m", STANDARD_ROCK_BOUNDARY_DEFAULT_OFFSET_M) ) try: design = compute_cross_design( section.get("samples", []), design_elevation_from_longitudinal(longitudinal, chainage_m), ground_type=str(stored.get("ground_type") or "ripping_rock"), section_mode=str(stored.get("section_mode") or default_modes.get(key, "left_cut")), ditch_side=stored.get("ditch_side"), ditch_type=str(stored.get("ditch_type") or "standard"), paved=bool(stored.get("paved", suggested)), standard=standard, rock_boundary_offset_m=rock_offset, two_stage_slope=bool(stored.get("two_stage_slope", True)), ditch_enabled=stored.get("ditch_enabled"), ) except (ValueError, KeyError): continue design["status"] = "provisional" design["pavement_suggested"] = suggested # 표시 설정(측점 개별 반폭)은 계산과 무관 — 재계산이 지우면 안 된다(2026-08-06). if stored.get("display_half_width_m") is not None: design["display_half_width_m"] = stored["display_half_width_m"] section["design"] = design @router.post( "/{project_id}/sections/{route_id}/cross-design/preview", response_model=CrossDesignPreviewResponse, ) async def preview_cross_designs( project_id: UUID, route_id: int, request: CrossDesignPreviewRequest ) -> CrossDesignPreviewResponse | JSONResponse: """계획선 편집 델타로 계획선과 전 측점 횡단 설계를 다시 계산해 돌려준다(저장 없음). B05에서 계획고를 끄는 동안 횡단 단면적·유토곡선이 함께 움직이게 하는 프리뷰 경로다. 영속화는 각 페이지의 임시저장·확정이 맡는다. """ pool = get_db_pool() try: async with pool.acquire() as connection: longitudinal_row = await get_longitudinal_section(connection, project_id, route_id) if not longitudinal_row: return JSONResponse( status_code=404, content={"status": "error", "message": "종횡단 상세 결과가 없습니다."}, ) stored_path = await get_project_storage_relative_path(connection, project_id) designs = await get_cross_section_designs(connection, route_id) project_root = Path(resolve_stored_project_path(stored_path)) detail = await asyncio.to_thread( _read_section_detail, project_root, str(longitudinal_row["longitudinal_file_path"]), ) def rebuild() -> None: alignment, profile = rebuild_alignment_profile(detail["longitudinal"], request.edits()) detail["longitudinal"]["profile_alignment"] = alignment detail["longitudinal"]["design_profiles"] = [profile] _recompute_designs_for_alignment( detail["longitudinal"], detail["cross_sections"], designs, request.standard_cross_section, request.rock_boundary_offsets, ) await asyncio.to_thread(rebuild) # 계획고를 끄는 동안 오가는 값이라 기본은 **유토곡선이 실제로 읽는 필드만** 싣는다. # 지반선 샘플·설계선 좌표(design_line)·선형 구조를 다 담으면 1MB가 넘어 편집이 굼떠진다 # (2026-08-03 사용자 지적). 계획선 자체는 화면이 이미 같은 규칙으로 계산해 들고 있다. # `full_designs`(B06 진입 시 stale 일괄 재계산)일 때만 설계 전체를 1회 싣는다 — # 측점별 순차 호출 N번보다 왕복 1번이 훨씬 싸다(2026-08-04 사용자 확인). return CrossDesignPreviewResponse( designs=[ { "chainage_m": section.get("chainage_m"), "design": section["design"] if request.full_designs else { key: section["design"].get(key) for key in _PREVIEW_DESIGN_FIELDS if key in section["design"] }, } for section in detail["cross_sections"] if section.get("design") ], ) except FileNotFoundError as exc: return JSONResponse(status_code=404, content={"status": "error", "message": str(exc)}) except (OSError, ValueError, json.JSONDecodeError) as exc: return JSONResponse(status_code=400, content={"status": "error", "message": str(exc)}) except Exception: logger.exception( "B06 횡단 설계 프리뷰 실패: project_id=%s route_id=%s", project_id, route_id ) return JSONResponse( status_code=500, content={"status": "error", "message": "횡단 설계 프리뷰 계산 중 오류가 발생했습니다."}, ) def _compute_default_designs( project_root: Path, longitudinal_file_path: str, chainages: list[float], standard: dict[str, Any] | None = None, ) -> list[tuple[float, dict[str, Any]]]: """미지정 측점들을 기본값(리핑암 + 암반 경계 0.5m/상단측 절토)으로 계산한 목록을 만든다. standard가 오면(확정 요청의 패널 편집값) 그 값으로 표준단면 기하를 계산한다. 계획고 부재 등으로 계산 불가한 측점은 조용히 건너뛴다(확정을 막지 않기 위함). """ root = project_root.resolve() longitudinal_path = (root / longitudinal_file_path).resolve() if root not in longitudinal_path.parents or not longitudinal_path.is_file(): return [] longitudinal = json.loads(longitudinal_path.read_text(encoding="utf-8")) cross_dir = longitudinal_path.parent.parent / "cross_sections" default_modes = _default_section_modes(longitudinal) pavement = _pavement_suggestions(longitudinal) results: list[tuple[float, dict[str, Any]]] = [] for chainage_m in chainages: cross_path = cross_dir / cross_filename(chainage_m) if not cross_path.is_file(): continue try: cross = json.loads(cross_path.read_text(encoding="utf-8")) samples = cross.get("samples") if not isinstance(samples, list): continue suggested = pavement.get(round(chainage_m, 3), False) design = compute_cross_design( samples, design_elevation_from_longitudinal(longitudinal, chainage_m), ground_type="ripping_rock", section_mode=default_modes.get(round(chainage_m, 3), "left_cut"), paved=suggested, standard=standard, rock_boundary_offset_m=STANDARD_ROCK_BOUNDARY_DEFAULT_OFFSET_M, ) design["status"] = "provisional" design["pavement_suggested"] = suggested results.append((chainage_m, design)) except (ValueError, KeyError, OSError, json.JSONDecodeError): continue return results @router.post("/{project_id}/sections/{route_id}/cross-design", response_model=CrossDesignResponse) async def compute_cross_section_design( project_id: UUID, route_id: int, request: CrossDesignRequest ) -> CrossDesignResponse | JSONResponse: """측점 표준횡단 설계(지반유형·단면유형)를 즉시 계산해 잠정치로 저장한다.""" pool = get_db_pool() try: async with pool.acquire() as connection: longitudinal = await get_longitudinal_section(connection, project_id, route_id) if not longitudinal: return JSONResponse( status_code=404, content={"status": "error", "message": "종횡단 상세 결과가 없습니다."}, ) stored_path = await get_project_storage_relative_path(connection, project_id) project_root = Path(resolve_stored_project_path(stored_path)) samples, design_elevation, pavement_suggested = await asyncio.to_thread( _read_cross_design_inputs, project_root, str(longitudinal["longitudinal_file_path"]), request.chainage_m, ) design = compute_cross_design( samples, design_elevation, ground_type=request.ground_type, section_mode=request.section_mode, ditch_side=request.ditch_side, ditch_type=request.ditch_type, paved=request.paved, standard=request.standard_cross_section, rock_boundary_offset_m=request.rock_boundary_offset_m, two_stage_slope=request.two_stage_slope, ditch_enabled=request.ditch_enabled, ) # B06 지정 시점은 잠정치. B08 도면 확정 시 동일 엔진으로 재계산해 confirmed로 승격한다. design["status"] = "provisional" # 법정 근거 문구 표기용 — 사용자가 포장을 바꿔도 제안 여부는 그대로 남긴다. design["pavement_suggested"] = pavement_suggested async with pool.acquire() as connection: # 표시 설정(측점 개별 반폭)은 계산 입력이 아니다 — 저장분에서 이월해 재계산이 # 지우지 않게 한다(2026-08-06). stored_designs = await get_cross_section_designs(connection, route_id) for record in stored_designs: if abs(float(record["chainage_m"]) - request.chainage_m) < 0.01: stored_design = record.get("design") if ( isinstance(stored_design, dict) and stored_design.get("display_half_width_m") is not None ): design["display_half_width_m"] = stored_design["display_half_width_m"] break await connection.begin() try: updated = await update_cross_section_design( connection, route_id=route_id, chainage_m=request.chainage_m, design=design, project_id=project_id, ) await connection.commit() except Exception: await connection.rollback() raise if not updated: return JSONResponse( status_code=404, content={"status": "error", "message": "해당 측점 횡단 레코드가 없습니다."}, ) return CrossDesignResponse(chainage_m=request.chainage_m, design=design) except FileNotFoundError as exc: return JSONResponse(status_code=404, content={"status": "error", "message": str(exc)}) except ValueError as exc: return JSONResponse(status_code=400, content={"status": "error", "message": str(exc)}) except Exception: logger.exception( "B06 측점 횡단 설계 계산 실패: project_id=%s route_id=%s", project_id, route_id ) return JSONResponse( status_code=500, content={"status": "error", "message": "횡단 설계 계산 중 오류가 발생했습니다."}, )