eomsangdonandClaude Opus 5.5
f243524436
- 시험: 건기 그대로 · 우기 r_u 0.3 이 지표 포화보다 안전율 높음 Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
96 lines
3.8 KiB
Python
96 lines
3.8 KiB
Python
"""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,
|
|
**model_kw,
|
|
) -> dict:
|
|
"""측점 하나 — case_workers > 1 이면 비탈 × 조건 경우를 프로세스로 나눠 돎(측점 하나만 다시 셀 때).
|
|
|
|
groundwater_for(condition, model) → 엔진 지하수 꼴 — 주면 조건마다 모형 지하수를 갈아 끼움
|
|
(E02 `station_groundwater` 꼴) · 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"])
|
|
jobs += [(cond, model, side) for side in model["sides"]]
|
|
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}
|
|
out["cases"].append({"condition": cond, **side, "min_F": mins, "result": res})
|
|
out["elapsed_s"] = round(time.perf_counter() - t0, 3)
|
|
return out
|
|
|
|
|
|
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]
|