Aislo/resources/tester/test_533_3_e04_engine.py

"""E04 엔진 잇기(PLAN 533-3) — 복사본 측점 → 533-1 합성 지형 → 533-2 탐색 → 저장본 꼴 ·
탐색 노드 = 구간 · 탐색 폭 안 · 진행률 · 끊기 · 고리를 달아도 결과 불변.

./venv/Scripts/python.exe -m pytest resources/tester/test_533_3_e04_engine.py -q
"""

from __future__ import annotations

import math

import numpy as np
import pytest

from E04_Stab3D import E04_Stab3D_Run as run
from E04_Stab3D.E04_Stab3D_Engine_Column import Strata
from E04_Stab3D.E04_Stab3D_Engine_Search import SearchParams, search

# 노선 = x 축(동쪽) · 왼쪽(북) = 오르막 0.6 · 측점 20 m
CHAINAGES = [0.0, 20.0, 40.0, 60.0, 80.0, 100.0]
SLOPE = 0.6


def _ground(y):
    return 100.0 + SLOPE * y


def _cross(ch: float) -> dict:
    offsets = np.arange(-25.0, 25.01, 0.5)
    samples = [{"offset_m": o, "z": _ground(o), "valid": True} for o in offsets]
    # 노면 |o| ≤ 3 평평(100) · 왼쪽 1:1 깎기 · 오른쪽 1:1.5 쌓기 · 바깥 = 지반
    line = []
    for o in offsets:
        if abs(o) <= 3:
            z = 100.0
        elif o > 0:
            z = min(100.0 + (o - 3), _ground(o))
        else:
            z = max(100.0 - (-o - 3) / 1.5, _ground(o))
        line.append({"offset_m": o, "elevation_m": z})
    return {
        "chainage_m": ch,
        "frame": {"left_xy": [0.0, 1.0]},
        "samples": samples,
        "design": {
            "design_line": line,
            "ground_type": "soil",
            "road_edges": {"left": {"offset_m": 3.0}, "right": {"offset_m": -3.0}},
        },
    }


STATIONS = [
    {"chainage_m": ch, "label": f"No.{i}", "center_x": ch, "center_y": 0.0}
    for i, ch in enumerate(CHAINAGES)
]

SETTINGS = {
    "start_m": 20.0,
    "end_m": 60.0,
    "condition": "dry",
    "shape": "both",
    "ellipsoid_ratio": 2.0,
    "box_half_width_m": 12.0,
    "cell_m": 1.0,
    "spacing_m": 3.0,
    "h_min_m": 2.0,
    "h_max_m": 10.0,
    "h_step_m": 2.0,
    "dr_m": 1.0,
    "max_depth_m": 6.0,
    "v_min_m3": 5.0,
    "v_max_m3": 3000.0,
}


@pytest.fixture
def root(tmp_path, monkeypatch):
    models = tmp_path / "B04_PreProcess" / "models"
    models.mkdir(parents=True)
    xs = np.arange(-60.0, 160.0, 1.0)
    ys = np.arange(-60.0, 60.0, 1.0)
    xx, yy = np.meshgrid(xs, ys)
    z = _ground(yy)
    np.savez(models / "dtm_classification.npz", x=xs, y=ys, z=z, valid_mask=np.ones_like(z, bool))
    monkeypatch.setattr(run.source, "cross_section", lambda r, ch: _cross(float(ch)))
    return tmp_path


def test_engine_raster_and_best(root):
    seen: list[tuple[int, str]] = []
    raw = run.run_engine(
        root, STATIONS[0:5], None, SETTINGS, lambda p, m: seen.append((p, m)), lambda: False
    )
    grid = raw["grid"]
    f = raw["f"]
    assert f.shape == (grid["ny"], grid["nx"]) == raw["z"].shape
    assert grid["dx"] == 1.0
    # 격자 = 구간 측점 ± (탐색 폭 + 깊이 + 10)
    assert grid["x0"] == pytest.approx(0.0 - 28.0, abs=1.0)
    finite = np.isfinite(f)
    assert finite.sum() > 50
    best = raw["best"]
    iy = round((best["y"] - grid["y0"]) / grid["dx"])
    ix = round((best["x"] - grid["x0"]) / grid["dx"])
    # 전체 최소 = 래스터 최소(그 활동체가 덮는 셀 중 하나) · 중심 = 탐색 띠 안(x 20~60 · |y| ≤ 12)
    assert np.nanmin(f) == pytest.approx(best["f"], rel=1e-6)
    assert 20.0 - 1e-6 <= best["x"] <= 60.0 + 1e-6 and abs(best["y"]) <= 12.0 + 1e-6
    assert 0 <= ix < grid["nx"] and 0 <= iy < grid["ny"]
    assert best["r"] > 0 and best["ratio"] in (1.0, 2.0) and 0 <= best["azimuth_deg"] < 360
    assert best["f"] == best["bishop"] or best["method"] == 1  # 다스림 = Bishop(잘림 없음)
    assert raw["reference_fs"] == 1.5  # 마스터 기본 건기
    # 활동체 셀 = 엔진 덩이 — 모두 그 셀 최소(= 전체 최소)
    cells = np.asarray(best["cells"])
    assert len(cells) and np.allclose(f.ravel()[cells], best["f"])
    # 합성 지표 — 노면 위(측점 40 · o=0) = 설계 100 · 깎기 끝 밖(o=+20 · 원지형) = 지반
    z = raw["z"]
    jy0 = round((0.0 - grid["y0"]) / grid["dx"])
    jx40 = round((40.0 - grid["x0"]) / grid["dx"])
    assert z[jy0, jx40] == pytest.approx(100.0, abs=0.05)
    jyw = round((20.0 - grid["y0"]) / grid["dx"])
    assert z[jyw, jx40] == pytest.approx(_ground(20.0), abs=0.05)
    # 진행률 — 0~100 · 단계 이름 · 마지막이 탐색 끝 무렵
    assert seen and all(0 <= p <= 100 for p, _ in seen) and seen[-1][0] >= 80


def test_engine_stop(root):
    calls = {"n": 0}

    def stop():
        calls["n"] += 1
        return calls["n"] > 3

    with pytest.raises(InterruptedError):
        run.run_engine(root, STATIONS[0:5], None, SETTINGS, lambda p, m: None, stop)


def test_wet_lower_than_dry(root):
    dry = run.run_engine(root, STATIONS[0:5], None, SETTINGS, lambda p, m: None, lambda: False)
    wet = run.run_engine(
        root,
        STATIONS[0:5],
        None,
        {**SETTINGS, "condition": "wet"},
        lambda p, m: None,
        lambda: False,
    )
    assert wet["best"]["f"] < dry["best"]["f"]
    assert wet["reference_fs"] == 1.3  # 우기 깎기 1.2 · 쌓기 1.3 중 큰 값


def test_hooks_do_not_change_search():
    """533-2 탐색에 단 진행률 · 끊기 고리 — 결과 불변(sub5 시험 꼴 작은 비탈)."""
    x = np.arange(-15.0, 35.0, 1.0)
    y = np.arange(-15.0, 15.0, 1.0)
    xx, yy = np.meshgrid(x, y)
    zz = np.clip(xx * 0.5, 0, 10)
    mat = [{"id": "m", "name": "m", "gamma_t": 19.0, "gamma_sat": 19.0, "c": 10.0, "phi": 25.0}]
    strata = Strata(x, y, zz, np.zeros((0,) + zz.shape), mat)
    base = dict(box=(-5, 22, -1, 1), spacing=2.0, h_min=2, h_max=14, h_step=2, dr=1.0,
                max_depth=10, v_min=20, v_max=2e4, ratios=(1.0,))  # fmt: skip
    plain = search(strata, SearchParams(**base))
    ticks: list[int] = []
    hooked = search(
        strata,
        SearchParams(**base, on_progress=lambda p, m: ticks.append(p), should_stop=lambda: False),
    )
    assert hooked.best["governing"] == plain.best["governing"]
    assert np.array_equal(np.isnan(hooked.f_min), np.isnan(plain.f_min))
    assert np.nanmax(np.abs(hooked.f_min - plain.f_min)) == 0
    assert ticks and min(ticks) >= 0 and max(ticks) <= 100
    with pytest.raises(InterruptedError):
        search(strata, SearchParams(**base, should_stop=lambda: True))
    assert math.isfinite(plain.best["governing"])


def test_node_mask_section_and_route_ends(root):
    """노드 자리 = 구간 누가 안 · 탐색 폭 안 · 노선 시점 앞(x < 0) · 종점 뒤(x > 100)는 뺌."""
    x = np.arange(-30.0, 131.0, 1.0)
    y = np.arange(-20.0, 21.0, 1.0)
    grid = {"x": x, "y": y}
    mask = run._node_mask(STATIONS, grid, 0.0, 100.0, 12.0)
    xx, yy = np.meshgrid(x, y)
    assert not mask[xx < -0.5].any() and not mask[xx > 100.5].any()
    assert not mask[np.abs(yy) > 12.0].any()
    assert mask[(np.abs(yy) <= 12.0) & (xx >= 1) & (xx <= 99)].all()
    part = run._node_mask(STATIONS, grid, 20.0, 60.0, 12.0)
    assert not part[(xx < 19.5) | (xx > 60.5)].any() and part[(yy == 0) & (xx == 40)].all()


def test_ground_order_and_label_map(tmp_path):
    """물성 = E02 작업본 → 초기값 → 마스터 기본값(성토 φ 28) · 층 물성 = label_map 짝."""
    import json

    from E04_Stab3D import E04_Stab3D_Engine_Grid as grid_engine

    assert {m["id"]: m["phi"] for m in run.read_ground(tmp_path)["materials"]}["fill"] == 28.0
    init = tmp_path / "initial_snapshot" / "E02_StabGround" / "ground.json"
    init.parent.mkdir(parents=True)
    init.write_text(json.dumps({"tag": "initial"}), encoding="utf-8")
    assert run.read_ground(tmp_path) == {"tag": "initial"}
    work = tmp_path / "E02_StabGround" / "ground.json"
    work.parent.mkdir(parents=True)
    work.write_text(json.dumps({"tag": "work"}), encoding="utf-8")
    assert run.read_ground(tmp_path) == {"tag": "work"}

    z = np.full((2, 2), 10.0)
    grid = {
        "x": np.arange(2.0),
        "y": np.arange(2.0),
        "z": z,
        "z_orig": z,
        "rock_top": z - 3,
        "rock_kind": np.ones((2, 2), int),
        "q": np.zeros((2, 2)),
    }
    ids = [m["id"] for m in grid_engine.strata_from_grid(grid).materials]
    assert ids == ["fill", "weathered_soil", "weathered_rock", "soft_rock"]
    mapped = grid_engine.strata_from_grid(grid, label_map={"soil": "colluvium", "fill": "fill"})
    assert [m["id"] for m in mapped.materials][:2] == ["fill", "colluvium"]


def _embankment(run_m, gamma, phi, c, half=20.0):
    """Scoops3D 검증 제방(H 10 m) — 탐색 상자 없이 격자 전체 · y ±20 m."""
    x = np.arange(-25, run_m + 35 + 1e-9, 1.0)
    y = np.arange(-half, half + 1e-9, 1.0)
    xx, _ = np.meshgrid(x, y)
    z = np.clip(xx / run_m * 10.0, 0, 10.0)
    mat = {"id": "m", "name": "m", "gamma_t": gamma, "gamma_sat": gamma, "c": c, "phi": phi}
    return Strata(x, y, z, np.zeros((0,) + z.shape), [mat])


@pytest.mark.parametrize(
    ("run_m", "gamma", "phi", "c", "expected"),
    [
        (10 / math.tan(math.radians(30)), 19, 20, 8.64, 1.32),  # 표 6.8 제방 30°
        (20.0, 20, 19.6, 3.0, 1.04),  # Donald & Giam 1a
        (10 / math.tan(math.radians(60)), 19, 20, 8.64, 0.75),  # 표 6.11 제방 60°
    ],
)
def test_default_settings_reproduce_scoops3d(run_m, gamma, phi, c, expected):
    """567-2 — 실사용 기본 설정 그대로(노드 2 m · 거친 4 배 · 높이 2~30 m · dr 1 m · 부피 10~10,000 m³ ·
    구 + 타원체) · 좁힌 상자 없이 → 문헌값 ±2 %(실측 +0.0 · −0.1 · −0.2 %)."""
    best = search(_embankment(run_m, gamma, phi, c), SearchParams(workers=8)).best
    assert best["governing"] == pytest.approx(expected, rel=0.02)
    assert best["columns"] == len(best["cells"]) >= 100 and best["few_columns"] is False


def test_few_columns_warning():
    """567-2 — 임계 활동체 기둥 < 100(Scoops3D limcol) → few_columns 경고."""
    st = _embankment(10 / math.tan(math.radians(60)), 19, 20, 8.64, half=8.0)
    p = SearchParams(box=(-9, -3, -1, 1), h_min=4, h_max=10, v_min=10, v_max=60, ratios=(1.0,))
    r = search(st, p)
    assert r.best["columns"] < 100 and r.best["few_columns"] is True
    assert r.stats["columns"] == r.best["columns"] and r.stats["few_columns"] is True