Aislo/E03_StabSection/E03_StabSection_Engine_Station.py

"""E03 측점 하나 해석 — 좌 · 우 비탈 × 조건(건기 · 우기 · 지진) × 방법.

조건 = KDS 11 70 05 표 4.3-3: 건기 u = 0 · 우기 지하수위 = 지표 · 지진 = 건기 수위 + kh.
판정(기준안전율 대조)은 화면 · 보고서 몫 — 여기는 값만 냄.
"""

import time

from E03_StabSection.E03_StabSection_Engine_Model import section_model_from_b06
from E03_StabSection.E03_StabSection_Engine_Search import METHODS, analyze
from E03_StabSection.E03_StabSection_Types import groundwater as norm_groundwater

#: 활동면 최소 깊이(m) — 지표를 얇게 긁는 면(표토 박리)은 사면 붕괴로 보지 않음.
MIN_DEPTH_M = 0.3


def analyze_station(
    cross: dict,
    design: dict,
    *,
    materials=None,
    layers=None,
    ground=None,
    conditions=("dry", "wet"),
    seismic_kh: float = 0.0,
    methods=METHODS,
    noncircular: bool = True,
    min_depth: float = MIN_DEPTH_M,
    case_workers: int = 1,
    groundwater_for=None,
    supports=None,
    **model_kw,
) -> dict:
    """측점 하나 — case_workers > 1 이면 비탈 × 조건 경우를 프로세스로 나눠 돎(측점 하나만 다시 셀 때).

    groundwater_for(condition, model) → 엔진 지하수 꼴 — 주면 조건마다 모형 지하수를 갈아 끼움
    (E02 `station_groundwater` 꼴) · None 이면 모형 기본(우기 = 지표 수위).
    supports = 네일 · 앵커(`Engine_Support` 꼴) — 보강재마다 "side"(없으면 머리 x 가 있는 쪽 = 노면 가운데
    기준 좌 +)의 비탈 경우에만 넣음 · 결과 cases[i].result.support_forces · None 이면 지금 그대로.

    서버 프로세스 안에서는 1(차례로) 그대로 둠 — 자식이 main.py 를 다시 읽음(`Worker` 머리글).
    """
    t0 = time.perf_counter()
    out = {"chainage_m": cross.get("chainage_m"), "label": cross.get("label"), "cases": []}
    jobs = []
    for cond in conditions:
        model = section_model_from_b06(
            cross,
            design,
            materials=materials,
            layers=layers,
            ground=ground,
            wet=cond == "wet",
            seismic_kh=seismic_kh if cond == "seismic" else 0.0,
            **model_kw,
        )
        if groundwater_for is not None:
            model["groundwater"] = norm_groundwater(groundwater_for(cond, model))
        out.setdefault("areas", model["areas"])
        for side in model["sides"]:
            own = _side_supports(supports, side["side"], design)
            jobs.append((cond, {**model, "supports": own} if own else model, side))
    kw = {"methods": methods, "noncircular": noncircular, "min_depth": min_depth}
    if case_workers > 1 and len(jobs) > 1:
        from concurrent.futures import ProcessPoolExecutor

        with ProcessPoolExecutor(max_workers=min(case_workers, len(jobs))) as pool:
            futures = [
                pool.submit(analyze, m, direction=sd["direction"], limits=sd["limits"], **kw)
                for _, m, sd in jobs
            ]
            results = [f.result() for f in futures]
    else:
        results = [
            analyze(m, direction=sd["direction"], limits=sd["limits"], **kw) for _, m, sd in jobs
        ]
    for (cond, _m, side), res in zip(jobs, results):
        mins = {m: (res["methods"][m]["min"] or {}).get("F") for m in methods}
        if supports:
            res.setdefault("support_forces", [])  # 보강재 없는 쪽 비탈 = 빈 칸
        out["cases"].append({"condition": cond, **side, "min_F": mins, "result": res})
    out["elapsed_s"] = round(time.perf_counter() - t0, 3)
    return out


def _side_supports(supports, side: str, design: dict) -> list[dict]:
    """그 비탈 보강재 — "side" 칸 · 없으면 머리 x 가 노면 가운데(좌 · 우 노면 끝 가운데)보다 + 면 좌."""
    if not supports:
        return []
    edges = design.get("road_edges") or {}
    mid = 0.5 * (
        float((edges.get("left") or {}).get("offset_m", 2.0))
        + float((edges.get("right") or {}).get("offset_m", -2.0))
    )
    return [
        s
        for s in supports
        if (s.get("side") or ("left" if float(s["x"]) >= mid else "right")) == side
    ]


def analyze_stations(items: list[dict], workers: int | None = None, **kw) -> list[dict]:
    """전 측점 — 측점마다 따로 도니 프로세스로 나눠 돎. items = [{"cross", "design", "ground"?}, …] · 순서 그대로."""
    import os
    from concurrent.futures import ProcessPoolExecutor

    workers = workers or max(1, (os.cpu_count() or 2) - 2)
    if workers == 1 or len(items) < 2:
        return [
            analyze_station(it["cross"], it["design"], ground=it.get("ground"), **kw)
            for it in items
        ]
    with ProcessPoolExecutor(max_workers=workers) as pool:
        futures = [
            pool.submit(analyze_station, it["cross"], it["design"], ground=it.get("ground"), **kw)
            for it in items
        ]
        return [f.result() for f in futures]