Merge remote-tracking branch 'origin/dev' into sub_laptop_5

2026-10-26 03:26:54 +09:00
parent aff924404d a5a9040b41
commit e014b5dc04
3 changed files with 77 additions and 43 deletions
@@ -8,6 +8,7 @@ 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
@@ -26,10 +27,14 @@ def analyze_station(
    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()
@@ -46,6 +51,8 @@ def analyze_station(
            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}
@@ -485,52 +485,57 @@ def search(strata: Strata, params: SearchParams | None = None) -> SearchResult:
                        )
        return improved

-    step = max(1, int(p.coarse))
-    grid = np.stack(
-        np.meshgrid(
-            np.arange(0, nhs, step), np.arange(0, nys, step), np.arange(0, nxs, step), indexing="ij"
-        ),
-        -1,
-    ).reshape(-1, 3)
-    ids = (grid[:, 0] * nys + grid[:, 1]) * nxs + grid[:, 2]
-    seen.update(ids.tolist())
-    seeds = run(ids)
-    rounds = passes = 0
-    while seeds and (step > 1 or rounds < p.refine_rounds):
-        if step == 1:
-            rounds += 1
-        step = max(1, step // 2)
-        passes += 1
-        stage[:] = [min(40 + 8 * (passes - 1), 77), 8, f"고운 탐색 {passes}"]
-        cand = set()
-        for nid in seeds:
-            ih, rem = divmod(nid, nys * nxs)
-            iy, ix = divmod(rem, nxs)
-            for dh in (-step, 0, step):
-                for dy in (-step, 0, step):
-                    for dxs in (-step, 0, step):
-                        h2, y2, x2 = ih + dh, iy + dy, ix + dxs
-                        if 0 <= h2 < nhs and 0 <= y2 < nys and 0 <= x2 < nxs:
-                            cid = (h2 * nys + y2) * nxs + x2
-                            if cid not in seen:
-                                cand.add(cid)
-        if not cand:
+    try:
+        step = max(1, int(p.coarse))
+        grid = np.stack(
+            np.meshgrid(
+                np.arange(0, nhs, step),
+                np.arange(0, nys, step),
+                np.arange(0, nxs, step),
+                indexing="ij",
+            ),
+            -1,
+        ).reshape(-1, 3)
+        ids = (grid[:, 0] * nys + grid[:, 1]) * nxs + grid[:, 2]
+        seen.update(ids.tolist())
+        seeds = run(ids)
+        rounds = passes = 0
+        while seeds and (step > 1 or rounds < p.refine_rounds):
            if step == 1:
-                break
-            continue
-        seen.update(cand)
-        seeds = run(np.fromiter(cand, np.int64))
+                rounds += 1
+            step = max(1, step // 2)
+            passes += 1
+            stage[:] = [min(40 + 8 * (passes - 1), 77), 8, f"고운 탐색 {passes}"]
+            cand = set()
+            for nid in seeds:
+                ih, rem = divmod(nid, nys * nxs)
+                iy, ix = divmod(rem, nxs)
+                for dh in (-step, 0, step):
+                    for dy in (-step, 0, step):
+                        for dxs in (-step, 0, step):
+                            h2, y2, x2 = ih + dh, iy + dy, ix + dxs
+                            if 0 <= h2 < nhs and 0 <= y2 < nys and 0 <= x2 < nxs:
+                                cid = (h2 * nys + y2) * nxs + x2
+                                if cid not in seen:
+                                    cand.add(cid)
+            if not cand:
+                if step == 1:
+                    break
+                continue
+            seen.update(cand)
+            seeds = run(np.fromiter(cand, np.int64))

-    # 타원체 — 구 탐색이 찾은 임계 노드(F < refine_below)에서만(시간 1/3 · 노선 1 km 실측)
-    ell = tuple(r for r in p.ratios if r != 1.0)
-    limit = max(p.refine_below, 1.2 * best["governing"])
-    crit = np.unique(node_map[(node_map >= 0) & (f_min < limit)])
-    if ell and len(crit):
-        stage[:] = [85, 14, "타원체"]
-        run(crit, ell)
+        # 타원체 — 구 탐색이 찾은 임계 노드(F < refine_below)에서만(시간 1/3 · 노선 1 km 실측)
+        ell = tuple(r for r in p.ratios if r != 1.0)
+        limit = max(p.refine_below, 1.2 * best["governing"])
+        crit = np.unique(node_map[(node_map >= 0) & (f_min < limit)])
+        if ell and len(crit):
+            stage[:] = [85, 14, "타원체"]
+            run(crit, ell)

-    if pool:
-        pool.shutdown()
+    finally:
+        if pool:  # 끊기(InterruptedError) · 오류에도 일꾼을 닫음
+            pool.shutdown(cancel_futures=True)
    f_out = np.where(np.isfinite(f_min), f_min, np.nan)
    return SearchResult(
        f_min=f_out,
@@ -238,6 +238,28 @@ def test_지하수_측점_모형에_끼움():
    assert res["methods"]["bishop"]["min"]["F"] > 0  # 끼운 모형이 엔진에서 돎


+def test_측점_해석_groundwater_for_인자():
+    ground = ground_model.build_initial(ground_model.load_defaults(), [])
+    ground["groundwater"].update(wet_mode="ru", ru=0.3)
+    seen = []
+
+    def gw_for(cond, model):
+        seen.append(cond)
+        return ground_model.station_groundwater(ground, cond, model)
+
+    kw = {"conditions": ("dry", "wet"), "methods": ("bishop",), "noncircular": False}
+    base = analyze_station(CROSS, DESIGN, **kw)
+    out = analyze_station(CROSS, DESIGN, groundwater_for=gw_for, **kw)
+    assert seen == ["dry", "wet"]
+    F = {(c["condition"], c["side"]): c["min_F"]["bishop"] for c in out["cases"]}
+    F0 = {(c["condition"], c["side"]): c["min_F"]["bishop"] for c in base["cases"]}
+    for (cond, side), f in F.items():
+        if cond == "dry":
+            assert f == pytest.approx(F0[cond, side])  # 건기 = 수위 없음 그대로
+        else:
+            assert f > F0[cond, side]  # r_u 0.3 < 지표 포화
+
+
@pytest.fixture
 def project(tmp_path, monkeypatch):
    async def fake_project(_project_id):