Merge remote-tracking branch 'origin/main_laptop_1' into sub_laptop_1

2026-09-06 22:28:36 +09:00
parent 36a135f436 5967d64b63
commit 81eb82a710
2 changed files with 86 additions and 49 deletions
@@ -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(
@@ -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():