diff --git a/B05_Profile/B05_Profile_Router_Confirm.py b/B05_Profile/B05_Profile_Router_Confirm.py index 8e87d7b6..ac8d637e 100644 --- a/B05_Profile/B05_Profile_Router_Confirm.py +++ b/B05_Profile/B05_Profile_Router_Confirm.py @@ -90,14 +90,16 @@ async def sync_uphill_overrides_into_designs( 양성(both_fill)은 측구가 없으므로 건드리지 않는다. """ # 지연 import — B06 라우터 모듈 로드는 이 함수가 실제 불릴 때만 필요하다. - from B06_Section.B06_Section_Engine_Design import compute_cross_design - from B06_Section.B06_Section_Repository import ( - get_cross_section_designs, - update_cross_section_design, + from B06_Section.B06_Section_Engine_Design import compute_cross_design, curve_widening_args + from B06_Section.B06_Section_Repository import get_cross_section_designs + from B06_Section.B06_Section_Repository_Bulk import merge_cross_section_designs + from B06_Section.B06_Section_Router_Design import ( + ford_drop_at, + ford_surface_drops, + pavement_suggestions, + read_cross_design_inputs, + resolve_longitudinal_path, ) - from B06_Section.B06_Section_Engine_Design import curve_widening_args - from B06_Section.B06_Section_Router import _read_cross_design_inputs - from B06_Section.B06_Section_Router_Design import ford_drop_at, ford_surface_drops if not overrides: return @@ -110,7 +112,8 @@ async def sync_uphill_overrides_into_designs( options = longitudinal["data"].get("options") if isinstance(options, dict): stored_standard = options.get("standard_cross_section") - ford_drops = ford_surface_drops(Path(project_root)) + # 다시 계산할 측점만 먼저 고른다 — 파일·계산은 아래에서 **스레드 한 번**에 몰아 한다. + jobs: list[tuple[float, dict[str, Any], str, str]] = [] for record in designs: chainage = round(float(record["chainage_m"]), 3) side = by_chainage.get(chainage) @@ -121,39 +124,55 @@ async def sync_uphill_overrides_into_designs( if mode == "both_fill": continue next_mode = f"{side}_cut" if mode in ("left_cut", "right_cut") else mode - next_ditch = side - if next_mode == mode and design.get("ditch_side") == next_ditch: + if next_mode == mode and design.get("ditch_side") == side: continue - samples, design_elevation, pavement_suggested, cross_record = await asyncio.to_thread( - _read_cross_design_inputs, project_root, longitudinal_file_path, float(chainage) - ) - next_design = compute_cross_design( - samples, - design_elevation, - ground_type=str(design.get("ground_type", "soil")), - section_mode=str(next_mode), - ditch_side=next_ditch, - ditch_type=str(design.get("ditch_type", "standard")), - paved=bool(design.get("paved", False)), - standard=stored_standard, - rock_boundary_offset_m=design.get("rock_boundary_offset_m"), - two_stage_slope=bool(design.get("two_stage_slope", True)), - ditch_enabled=design.get("ditch_enabled"), - surface_drop_m=ford_drop_at(float(chainage), ford_drops), - **curve_widening_args(cross_record), - ) - next_design["status"] = design.get("status", "provisional") - next_design["pavement_suggested"] = design.get("pavement_suggested", pavement_suggested) - # 개별 표시 반폭 등 계산과 무관한 표시 설정은 그대로 이월한다. - if design.get("display_half_width_m") is not None: - next_design["display_half_width_m"] = design["display_half_width_m"] - await update_cross_section_design( - connection, - route_id=route_id, - chainage_m=float(chainage), - design=next_design, - project_id=project_id, - ) + jobs.append((chainage, design, str(next_mode), side)) + if not jobs: + return + + def _recompute() -> list[tuple[float, dict[str, Any]]]: + """측점마다 종단 정본을 다시 열던 것을 한 번으로 줄인다(측점당 13.8ms 였다).""" + root = Path(project_root) + longitudinal_path = resolve_longitudinal_path(root, longitudinal_file_path) + longitudinal_json = json.loads(longitudinal_path.read_text(encoding="utf-8")) + preloaded = (longitudinal_path, longitudinal_json, pavement_suggestions(longitudinal_json)) + ford_drops = ford_surface_drops(root) + out: list[tuple[float, dict[str, Any]]] = [] + for chainage, design, next_mode, side in jobs: + samples, design_elevation, pavement_suggested, cross_record = read_cross_design_inputs( + root, longitudinal_file_path, float(chainage), preloaded + ) + next_design = compute_cross_design( + samples, + design_elevation, + ground_type=str(design.get("ground_type", "soil")), + section_mode=next_mode, + ditch_side=side, + ditch_type=str(design.get("ditch_type", "standard")), + paved=bool(design.get("paved", False)), + standard=stored_standard, + rock_boundary_offset_m=design.get("rock_boundary_offset_m"), + two_stage_slope=bool(design.get("two_stage_slope", True)), + ditch_enabled=design.get("ditch_enabled"), + surface_drop_m=ford_drop_at(float(chainage), ford_drops), + **curve_widening_args(cross_record), + ) + next_design["status"] = design.get("status", "provisional") + next_design["pavement_suggested"] = design.get("pavement_suggested", pavement_suggested) + # 개별 표시 반폭 등 계산과 무관한 표시 설정은 그대로 이월한다. + if design.get("display_half_width_m") is not None: + next_design["display_half_width_m"] = design["display_half_width_m"] + out.append((float(chainage), next_design)) + return out + + # 쓰기도 한 문장으로 — 행마다 내면 원격 DB 왕복이 측점 수만큼 난다(건당 24.5ms). + await merge_cross_section_designs( + connection, + route_id=route_id, + entries=await asyncio.to_thread(_recompute), + replace=True, + project_id=project_id, + ) def _merge_irregular_into_longitudinal( diff --git a/B06_Section/B06_Section_Router_Design.py b/B06_Section/B06_Section_Router_Design.py index af631349..35eb3830 100644 --- a/B06_Section/B06_Section_Router_Design.py +++ b/B06_Section/B06_Section_Router_Design.py @@ -251,19 +251,37 @@ def default_section_modes(longitudinal: dict[str, Any]) -> dict[float, str]: return mapping -def read_cross_design_inputs( - project_root: Path, longitudinal_file_path: str, chainage_m: float -) -> tuple[list[dict], float | None, bool, dict[str, Any]]: - """(지반 샘플, 계획고, 포장 제안, 측점 기록) — 측점 기록은 곡선부 확폭 입력을 담고 있다.""" +def resolve_longitudinal_path(project_root: Path, longitudinal_file_path: str) -> Path: + """종단 정본 파일 경로를 검증해 돌려준다 — 저장소 밖 경로를 막는다.""" root = project_root.resolve() - longitudinal_path = (root / longitudinal_file_path).resolve() - if root not in longitudinal_path.parents: + path = (root / longitudinal_file_path).resolve() + if root not in path.parents: raise ValueError("종단면 파일 경로가 프로젝트 저장소를 벗어났습니다.") - if not longitudinal_path.is_file(): + if not path.is_file(): raise FileNotFoundError("종단면 상세 파일을 찾을 수 없습니다.") - longitudinal = json.loads(longitudinal_path.read_text(encoding="utf-8")) + return path + + +def read_cross_design_inputs( + project_root: Path, + longitudinal_file_path: str, + chainage_m: float, + preloaded: tuple[Path, dict[str, Any], dict[float, bool]] | None = None, +) -> tuple[list[dict], float | None, bool, dict[str, Any]]: + """(지반 샘플, 계획고, 포장 제안, 측점 기록) — 측점 기록은 곡선부 확폭 입력을 담고 있다. + + `preloaded` 는 (종단 경로, 종단 내용, 포장 제안표)다. 여러 측점을 잇달아 볼 때 + 종단 정본을 측점마다 다시 읽지 않게 넘긴다 — 그 재읽기가 측점당 13.8ms 였다 + (2026-09-06 실측, 측구 방향 역반영 루프). + """ + if preloaded is not None: + longitudinal_path, longitudinal, pavement = preloaded + else: + longitudinal_path = resolve_longitudinal_path(project_root, longitudinal_file_path) + longitudinal = json.loads(longitudinal_path.read_text(encoding="utf-8")) + pavement = pavement_suggestions(longitudinal) design_elevation = design_elevation_from_longitudinal(longitudinal, chainage_m) - suggested = pavement_suggestions(longitudinal).get(round(chainage_m, 3), False) + suggested = pavement.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():