feat(532-5): E03 → E05 보고서 꼴 내보냄 · 지하수 갈래 지문 · r_u 범례 · 이음매 시험

- [해석] 끝 · [대표 단면 저장] 때 E03_StabSection/_derived/result2d.json(측점 줄 · 지배 비탈 · 조건별 최소 · 판정) · results/<대표 측점>.json · representative.json stations 덧붙임(overrides 그대로)
- 측점 지문에 E02 지하수 갈래 · 일꾼 job 에 지반 조건(sub4 analyze_station 고리 뒤 측점 수위로 이음)
- 측점 보기 water_rule — r_u 갈래면 범례에 r_u 값(수위선 없음)
- 시험 test_532_5_e03_seam — 벽 측점 임계면 · 벽 측점만 지문 · 지하수 갈래 지문 · 수위선 갈래 · 빠진 횡단 · E05 가 E03 저장본으로 빌더 돎

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

E03_StabSection/E03_StabSection_Router.py

@@ -64,6 +64,8 @@ PROJECT_ROOT = Path(__file__).resolve().parents[1]
 KIND_STATION = worker.KIND_STATION
 KIND_SUMMARY = "stab2d_summary"
 REPRESENTATIVE = Path("E03_StabSection") / "representative.json"
+#: E05 보고서가 읽는 꼴(`E05_StabReport_Build` · result2d.json · results/<측점>.json)
+REPORT_DIR = Path("E03_StabSection") / "_derived"
 ENGINE_FILES = (
     "E03_StabSection_Types.py",
     "E03_StabSection_Engine_Model.py",
@@ -132,6 +134,7 @@ def plan_run(root: Path, ground: dict[str, Any]) -> dict[str, Any]:
         "conditions": conditions,
         "seismic_kh": float(seismic.get("kh") or 0.0) if seismic.get("enabled") else 0.0,
         "road_surcharge_kpa": float(ground.get("road_surcharge_kpa") or 0.0),
+        "groundwater": ground.get("groundwater"),
     }
     resolved = {chainage_key(r["chainage_m"]): r for r in ground_model.resolve_stations(ground)}
     try:
@@ -158,7 +161,14 @@ def plan_run(root: Path, ground: dict[str, Any]) -> dict[str, Any]:
         )
     factors = ground.get("safety_factors") or {}
     overall = derived_key([item["fp"] for item in items], factors)
-    return {"source": source, "shared": shared, "items": items, "factors": factors, "fp": overall}
+    return {
+        "source": source,
+        "ground": ground,
+        "shared": shared,
+        "items": items,
+        "factors": factors,
+        "fp": overall,
+    }
 
 
 def _station_input(source: Path, item: dict[str, Any], sampler: Any) -> dict[str, Any]:
@@ -235,6 +245,7 @@ def run_all(root: Path, plan: dict[str, Any], report: Any) -> int:
             "root": str(root),
             "keep": max(3 * len(items), 6),
             "workers": WORKERS or max(1, (os.cpu_count() or 2) - 2),
+            "conditions_ground": plan["ground"],
             "kwargs": {
                 "materials": shared["materials"],
                 "conditions": list(shared["conditions"]),
@@ -269,9 +280,38 @@ def run_all(root: Path, plan: dict[str, Any], report: Any) -> int:
         "stations": rows,
     }
     write_derived(root, KIND_SUMMARY, plan["fp"], summary)
+    export_report(root, summary)
     return len(todo)
 
 
+def export_report(root: Path, summary: dict[str, Any]) -> None:
+    """E05 보고서 입력 — 요약(측점 줄) · 대표 단면 측점 해석 결과 · representative.json `stations`
+    (E03 은 `overrides` 만 읽음). [해석] 끝 · [대표 단면 저장] 때 씀."""
+    rows = [dict(r) for r in summary["stations"]]
+    overrides = _overrides(root)
+    view.mark_representative(rows, overrides)
+    folder = root / REPORT_DIR
+    atomic_write_json(
+        folder / "result2d.json",
+        {
+            "input_fingerprint": summary["fingerprint"],
+            "computed_at": summary["computed_at"],
+            "conditions": summary["conditions"],
+            "stations": [view.report_row(r) for r in rows],
+        },
+    )
+    reps = view.report_representatives(rows)
+    results = folder / "results"
+    for old in results.glob("*.json"):
+        old.unlink()
+    by_key = {r["key"]: r for r in rows}
+    for rep in reps:
+        saved = read_derived(root, KIND_STATION, by_key[rep["station_key"]]["fp"])
+        if saved and saved["value"]["out"]:
+            atomic_write_json(results / f"{rep['station_key']}.json", saved["value"]["out"])
+    atomic_write_json(root / REPRESENTATIVE, {"overrides": overrides, "stations": reps})
+
+
 def _latest_summary(root: Path, fp: str) -> tuple[dict[str, Any] | None, bool]:
     """(요약, 지금 입력과 같은지) — 같은 지문이 없으면 가장 최근 것(옛 결과)."""
     saved = read_derived(root, KIND_SUMMARY, fp)
@@ -398,13 +438,15 @@ async def section(project_id: UUID, station_key: str) -> dict[str, Any]:
         "walls": walls_mod.walls_of(entry),
     }
     dry = section_model_from_b06(cross, design, **kw)
-    water = {}
+    water, rules = {}, {}
     for cond in summary["conditions"]:
         # 수위선 = E02 고른 갈래(r_u · 없음 = 선 없음)
         model = section_model_from_b06(cross, design, wet=cond == "wet", **kw)
         gw = ground_model.station_groundwater(ground, cond, model)
         water[cond] = gw.get("line") if gw.get("mode") == "line" else None
+        rules[cond] = {"mode": gw.get("mode"), "ru": gw.get("ru")}
     out = view.section_view(row, value["out"], dry, water)
+    out["water_rule"] = rules
     out["fresh"] = fresh
     return out
 
@@ -413,12 +455,16 @@ async def section(project_id: UUID, station_key: str) -> dict[str, Any]:
 async def put_representative(project_id: UUID, payload: RepresentativeRequest) -> dict[str, Any]:
     """대표 단면 켬/끔(None = 자동) — 아는 측점 열쇠만."""
     root, _ = await e02._project(project_id)
-    known = {item["key"] for item in plan_run(root, _ground(root))["items"]}
+    plan = plan_run(root, _ground(root))
+    known = {item["key"] for item in plan["items"]}
     unknown = sorted(set(payload.overrides) - known)
     if unknown:
         raise HTTPException(status_code=422, detail=f"모르는 측점: {unknown[:5]}")
     overrides = {k: v for k, v in sorted(payload.overrides.items()) if v is not None}
     atomic_write_json(root / REPRESENTATIVE, {"overrides": overrides})
+    summary, _ = _latest_summary(root, plan["fp"])
+    if summary:
+        export_report(root, summary)
     return {"overrides": overrides}
 
 

E03_StabSection/E03_StabSection_UI_Page.ts

@@ -296,6 +296,15 @@ export function renderE03StabSection(root: HTMLElement): void {
     const draw = { view, analysis, toggles, picked };
     canvas.render(draw);
     legendHost.append(buildLegends(draw));
+    const rule = view.water_rule?.[condition];
+    if (rule?.mode === "ru")
+      legendHost.append(
+        el("p", {
+          className: "e03-muted",
+          text: t("E03_Water_Ru").replace("{v}", String(rule.ru)),
+          attrs: { "data-role": "water-ru" },
+        }),
+      );
     if (view.regions.some((r) => r.material_id === WALL_ID))
       legendHost.append(el("p", { className: "e03-muted", text: t("E03_Walls_Hint") }));
     resultHost.append(

E03_StabSection/E03_StabSection_View.py

@@ -130,6 +130,44 @@ def mark_representative(rows: list[dict[str, Any]], overrides: dict[str, bool])
         }
 
 
+#: 대표 단면 까닭 → E05 보고서 열쇠(`E05_StabReport_Build.REASONS`)
+REPORT_REASONS = {"fill_height": "fill_10m", "cut_height": "cut_15m", "lowest": "lowest_fs"}
+
+
+def report_row(row: dict[str, Any]) -> dict[str, Any]:
+    """E05 보고서가 읽는 측점 줄 — 지배 비탈(전 조건 최소 안전율 쪽) 역할 · 조건별 최소 · 판정."""
+    worst = None
+    for by_side in row["cases"].values():
+        for side, value in by_side.items():
+            if value["fs"] is not None and (worst is None or value["fs"] < worst[0]):
+                worst = (value["fs"], side)
+    roles = {s["side"]: s["role"] for s in row["sides"]}
+    values = [v for v in row["min"].values() if v is not None]
+    return {
+        "station_key": row["key"],
+        "chainage_m": row["chainage_m"],
+        "label": row.get("label"),
+        "side": roles.get(worst[1]) if worst else None,
+        "fs_min": min(values) if values else None,
+        "fs_by_condition": dict(row["min"]),
+        "method": PRIMARY,
+        "verdict": overall_verdict(row),
+        "walls": row.get("walls") or 0,
+    }
+
+
+def report_representatives(rows: list[dict[str, Any]]) -> list[dict[str, Any]]:
+    """E05 대표 단면 목록(켠 것만) — `mark_representative` 뒤 행."""
+    out = []
+    for row in rows:
+        rep = row.get("representative") or {}
+        if rep.get("on"):
+            auto = rep.get("auto") or []
+            reason = "user" if rep.get("user") or not auto else REPORT_REASONS.get(auto[0], auto[0])
+            out.append({"station_key": row["key"], "reason": reason, "enabled": True})
+    return out
+
+
 def summarize(rows: list[dict[str, Any]], conditions: list[str]) -> dict[str, Any]:
     """머리 요약 — 조건별 최소 안전율 · N.G 측점 수."""
     out: dict[str, Any] = {"stations": len(rows), "errors": sum(1 for r in rows if r["error"])}

E03_StabSection/E03_StabSection_Worker.py

@@ -3,7 +3,8 @@
 `python -m E03_StabSection.E03_StabSection_Worker <job.json>` — 서버(`Router.run_all`)가 띄움.
 측점마다 프로세스로 나눠 돌리고 끝날 때마다 파생 저장본을 쓰고 한 줄(JSON)을 찍음 → 서버 진행률.
 서버 프로세스에서 바로 프로세스 풀을 열면 자식마다 `main.py` 를 다시 읽어 이 모듈을 입구로 둠.
-job = {root, keep, kwargs, workers, items: [{fp, meta, layers, cross, design, walls, ground}]}.
+job = {root, keep, kwargs, workers, conditions_ground, items: [{fp, meta, layers, cross, design,
+walls, ground}]} · conditions_ground = E02 지반 조건(지하수 갈래 → 측점 수위).
 """
 
 from __future__ import annotations
@@ -22,7 +23,9 @@ KIND_STATION = "stab2d_station"
 logger = logging.getLogger(__name__)
 
 
-def _analyze(item: dict[str, Any], kwargs: dict[str, Any]) -> dict[str, Any]:
+def _analyze(
+    item: dict[str, Any], kwargs: dict[str, Any], conditions_ground: dict | None = None
+) -> dict[str, Any]:
     kw = {**kwargs, "layers": item["layers"], "ground": [tuple(p) for p in item["ground"]]}
     if item.get("walls"):
         kw["walls"] = item["walls"]
@@ -41,6 +44,7 @@ def _value(item: dict[str, Any], out: dict | None, error: str | None) -> dict[st
 def run_job(job: dict[str, Any], emit: Callable[[str], None]) -> None:
     """측점마다 해석 → 저장본 · emit(열쇠). 한 측점 실패는 그 줄에 까닭만 남기고 이어 감."""
     root, keep, kwargs = Path(job["root"]), int(job["keep"]), job["kwargs"]
+    conditions_ground = job.get("conditions_ground")
     items = job["items"]
 
     def save(item: dict[str, Any], out: dict | None, error: str | None) -> None:
@@ -51,13 +55,13 @@ def run_job(job: dict[str, Any], emit: Callable[[str], None]) -> None:
     if workers == 1:
         for item in items:
             try:
-                save(item, _analyze(item, kwargs), None)
+                save(item, _analyze(item, kwargs, conditions_ground), None)
             except Exception as error:  # noqa: BLE001
                 logger.exception("E03 측점 해석 실패: %s", item["meta"]["key"])
                 save(item, None, str(error))
         return
     with ProcessPoolExecutor(max_workers=workers) as pool:
-        futures = {pool.submit(_analyze, item, kwargs): item for item in items}
+        futures = {pool.submit(_analyze, item, kwargs, conditions_ground): item for item in items}
         for future in as_completed(futures):
             item = futures[future]
             try:

E03_StabSection/logic/E03_StabSection_Logic.ts

@@ -106,6 +106,8 @@ export interface SectionView {
   bottom_z: number | null;
   surcharge: { x0: number; x1: number; q: number }[];
   water: Partial<Record<Condition, Point[] | null>>;
+  /** 조건별 지하수 꼴(E02 갈래) — r_u 면 수위선 없음 */
+  water_rule?: Partial<Record<Condition, { mode: string; ru: number | null }>>;
   analyses: Analysis[];
 }
 

resources/tester/test_532_5_e03_seam.py

@@ -0,0 +1,216 @@
+# -*- coding: utf-8 -*-
+"""532-5 E03 이음매 — 2D 해석 길에 벽(532-4) · 지하수 갈래(E02) · 빠진 횡단 알림(532-1 덧).
+
+- 벽: 검증_sub2 복사본 벽 선 측점(351 m 기슭막이) — 길이 벽을 넘김 · 임계면이 벽을 안 지남 ·
+  측점 보기에 벽 폴리곤 · 목록 벽 수 · 벽 있는 측점만 지문이 바뀜
+- 지하수: E02 우기 갈래를 바꾸면 측점 지문이 바뀜 · 측점 보기 수위선 = 고른 갈래
+- 빠진 횡단: 횡단 파일이 없으면 result2d · stab/ground 의 missing_stations 에 그 측점
+
+    ./venv/Scripts/python.exe -m pytest resources/tester/test_532_5_e03_seam.py -q
+"""
+
+from __future__ import annotations
+
+import asyncio
+import json
+import shutil
+import sys
+import tempfile
+from pathlib import Path
+from uuid import UUID
+
+import numpy as np
+import pytest
+from shapely.geometry import LineString, Polygon
+
+ROOT = Path(__file__).resolve().parents[2]
+sys.path.insert(0, str(ROOT))
+
+from common_util import common_util_project_copy as cp  # noqa: E402
+from common_util.common_util_station_key import chainage_key  # noqa: E402
+from E01_StabRegister import E01_StabRegister_Source as src  # noqa: E402
+from E02_StabGround import E02_StabGround_Model as ground_model  # noqa: E402
+from E02_StabGround import E02_StabGround_Router as e02  # noqa: E402
+from E03_StabSection import E03_StabSection_Router as router  # noqa: E402
+from E03_StabSection import E03_StabSection_Worker as worker  # noqa: E402
+from E03_StabSection.E03_StabSection_Engine_Station import analyze_station  # noqa: E402
+
+SOURCE_ID = "cb15892a-5788-4a26-9316-20a15421f68b"  # 검증_sub2(읽기만)
+PID = "e" * 8 + "-" + "e" * 4 + "-" + "e" * 4 + "-" + "e" * 4 + "-" + "e" * 12
+WALL_KEY = "351.00"  # 검증_sub2 기슭막이 측점
+CORNER_M = 0.2  # 임계면이 벽 모서리를 스치는 길이 한도(test_532_4 와 같음)
+USER = {"role": "USER", "company_id": 1}
+
+
+async def _copy(target: Path) -> None:
+    import aiomysql
+
+    from config.config_db import DB_HOST, DB_NAME, DB_PASSWORD, DB_PORT, DB_USER
+
+    connection = await aiomysql.connect(
+        host=DB_HOST, port=DB_PORT, user=DB_USER, password=DB_PASSWORD, db=DB_NAME, autocommit=True
+    )
+    try:
+        await cp.copy_project(
+            connection, SOURCE_ID, scope="stability", new_id=PID, project_root=target
+        )
+    finally:
+        connection.close()
+
+
+@pytest.fixture(scope="module")
+def frozen():
+    folder = Path(tempfile.mkdtemp(prefix="aislo_e03_seam_"))
+    try:
+        asyncio.run(_copy(folder / "copy"))
+    except Exception as error:  # noqa: BLE001
+        shutil.rmtree(folder, ignore_errors=True)
+        pytest.skip(f"검증 원본 복사 실패: {error}")
+    root = folder / "copy"
+    if WALL_KEY not in router._walls(str(src.source_dir(root)), "seam"):
+        shutil.rmtree(folder, ignore_errors=True)
+        pytest.skip("검증_sub2 에 벽 측점이 없음")
+    yield root
+    shutil.rmtree(folder, ignore_errors=True)
+
+
+def _fast(cross, design, **kw):
+    """진짜 엔진 — Bishop 원호만(빠르게)."""
+    return analyze_station(cross, design, **{**kw, "methods": ("bishop",), "noncircular": False})
+
+
+@pytest.fixture
+def project(frozen, monkeypatch):
+    """벽 측점 + 벽 없는 측점 하나 · 진짜 엔진(빠르게) · 서버 안에서 차례로."""
+    for name in ("_derived", "E02_StabGround", "E03_StabSection", "initial_snapshot"):
+        shutil.rmtree(frozen / name, ignore_errors=True)
+    original = src.stations
+
+    def pick(root):
+        rows = original(root)
+        wall = [r for r in rows if chainage_key(r["chainage_m"]) == WALL_KEY]
+        return wall + [r for r in rows if chainage_key(r["chainage_m"]) != WALL_KEY][:1]
+
+    monkeypatch.setattr(router.src, "stations", pick)
+
+    async def fake_project(_pid):
+        return frozen, {"company_id": 1, "project_kind": "STABILITY"}
+
+    monkeypatch.setattr(e02, "_project", fake_project)
+    monkeypatch.setattr(worker, "analyze_station", _fast)
+    monkeypatch.setattr(router, "WORKERS", 1)
+    router._JOBS.clear()
+    return frozen
+
+
+def _analyze() -> dict:
+    async def go():
+        first = await router.analyze2d(UUID(PID))
+        while router._TASKS:
+            await asyncio.gather(*list(router._TASKS))
+        return first
+
+    return asyncio.run(go())
+
+
+def _save_ground(root: Path, ground: dict) -> None:
+    ground_model.ground_path(root).parent.mkdir(parents=True, exist_ok=True)
+    ground_model.ground_path(root).write_text(json.dumps(ground), encoding="utf-8")
+
+
+def test_벽_측점_임계면이_벽을_안_지남(project):
+    assert _analyze()["status"] == "accepted"
+    assert router._JOBS[PID]["error"] is None
+    body = asyncio.run(router.result2d(UUID(PID), USER))
+    walls = {r["key"]: r["walls"] for r in body["stations"]}
+    assert walls[WALL_KEY] >= 1 and sorted(walls.values())[0] == 0
+    sec = asyncio.run(router.section(UUID(PID), WALL_KEY))
+    polys = [Polygon(r["polygon"]) for r in sec["regions"] if r["material_id"] == "wall"]
+    assert polys and any(m["id"] == "wall" for m in sec["materials"])
+    crossed = 0
+    for a in sec["analyses"]:
+        line = LineString(a["critical"]["points"])
+        for poly in polys:
+            assert line.intersection(poly).length <= CORNER_M
+        crossed += 1
+    assert crossed == len(sec["analyses"]) > 0
+
+
+def test_벽_있는_측점만_지문이_바뀜(project, monkeypatch):
+    ground = router._ground(project)
+    with_walls = {it["key"]: it["fp"] for it in router.plan_run(project, ground)["items"]}
+    monkeypatch.setattr(router, "_walls", lambda source, fp: {})
+    plain = {it["key"]: it["fp"] for it in router.plan_run(project, ground)["items"]}
+    assert with_walls[WALL_KEY] != plain[WALL_KEY]
+    assert [with_walls[k] == plain[k] for k in plain if k != WALL_KEY] == [True]
+
+
+def test_지하수_갈래_바꾸면_지문_바뀜(project):
+    ground = router._ground(project)
+    before = router.plan_run(project, ground)
+    ground["groundwater"]["wet_mode"] = "ru"
+    after = router.plan_run(project, ground)
+    assert before["fp"] != after["fp"]
+    assert all(a["fp"] != b["fp"] for a, b in zip(before["items"], after["items"]))
+
+
+def test_수위선_고른_갈래(project):
+    _analyze()
+    key = next(r["key"] for r in asyncio.run(router.result2d(UUID(PID), USER))["stations"])
+    first = asyncio.run(router.section(UUID(PID), key))
+    kds, surface = first["water"]["wet"], first["surface"]
+    ground = router._ground(project)
+    ground["groundwater"].update(wet_mode="ru", ru=0.3)
+    _save_ground(project, ground)
+    assert asyncio.run(router.section(UUID(PID), key))["water"]["wet"] is None  # r_u = 선 없음
+    ground["groundwater"].update(wet_mode="depth_ratio", depth_ratio=0.5)
+    _save_ground(project, ground)
+    deep = asyncio.run(router.section(UUID(PID), key))["water"]["wet"]
+    assert kds and deep
+    # 깊이 비율 선 = 공사 뒤 지표 아래 · 비탈 쪽은 확실히 아래(연직 면 x 는 뺌 — 지표 값이 둘)
+    sx, sz = zip(*surface)
+    steps = {x for x in sx if sx.count(x) > 1}
+    below = [float(np.interp(x, sx, sz)) - z for x, z in deep if x not in steps]
+    assert min(below) >= -1e-6 and max(below) > 0.1
+
+
+def test_빠진_횡단_목록(project):
+    source = src.source_dir(project)
+    row = src._cross_rows(source)[3]
+    path = src.resolve_cross_file(source, row)
+    moved = path.with_suffix(".bak")
+    path.rename(moved)
+    try:
+        body = asyncio.run(router.result2d(UUID(PID), USER))
+        ground = asyncio.run(e02.get_ground(UUID(PID), USER))
+    finally:
+        moved.rename(path)
+    assert body["missing_stations"] == [float(row["chainage_m"])]
+    assert ground["missing_stations"] == [float(row["chainage_m"])]
+    assert asyncio.run(router.result2d(UUID(PID), USER))["missing_stations"] == []
+
+
+def test_E05_가_E03_저장본으로_보고서(project):
+    """E03 [해석] 이 쓴 E05 꼴(result2d.json · results/ · representative stations)로 E05 가 돎."""
+    from E05_StabReport import E05_StabReport_Build as build
+
+    _analyze()
+    inputs = build.gather(project)
+    rows = inputs["result2d"]["stations"]
+    assert len(rows) == 2 and WALL_KEY in {r["station_key"] for r in rows}
+    for r in rows:
+        assert r["side"] in ("cut", "fill") and r["verdict"] in ("ok", "ng")
+        assert r["fs_min"] == min(v for v in r["fs_by_condition"].values() if v is not None)
+    reps = inputs["representatives"]
+    assert reps and all(rep["result"] and rep["result"]["cases"] for rep in reps)
+    book = build.form()
+    jobs = build.build_program(book, inputs)
+    assert len([k for k, svg in jobs if svg.startswith("<svg")]) >= len(reps)
+
+    # 대표 단면 끔 → E05 목록에서 빠짐 · E03 overrides 는 그대로 읽힘
+    key = reps[0]["station_key"]
+    asyncio.run(
+        router.put_representative(UUID(PID), router.RepresentativeRequest(overrides={key: False}))
+    )
+    assert key not in [r["station_key"] for r in build.gather(project)["representatives"]]
+    assert router._overrides(project) == {key: False}

ui_template/ui_template_locale_e03.ts

@@ -74,6 +74,10 @@ export const ui_locales_e03 = {
   E03_Legend_Layers: ["지층", "Layers"],
   E03_Offset: ["거리", "Offset"],
   E03_Walls: ["벽 있음", "Walls"],
+  E03_Water_Ru: [
+    "지하수 = 간극수압비 r_u {v}(수위선 없음)",
+    "Groundwater = pore pressure ratio r_u {v} (no water table)",
+  ],
   E03_Walls_Hint: [
     "옹벽 · 석축(회색 빗금)은 강한 물체로 넣어 활동면이 지나지 않습니다.",
     "Retaining walls (gray hatch) are modeled as rigid bodies — slip surfaces do not cross them.",