Aislo/resources/tester/test_532_5_e03_seam.py

# -*- 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}