"""D 예정노선 2단계 분석(PLAN 530-2) — 합성 지형 두 구간 · 경계 · 누가 · 저장본 · 복도 인자. ./venv/Scripts/python.exe -m pytest resources/tester/test_530_2_plan_analyze.py -q """ import numpy as np import pytest from B05_Profile.B05_Profile_Engine_RouteInitial import _POOLS, Terrain from B05_Profile.B05_Profile_Engine_RouteInitial_Search import CORRIDOR_M, SearchGrid from D03_PlanAnalysis import D03_PlanAnalysis_Analyze as analyze CRITERIA = {"max_grade": 0.14, "min_radius_m": 12.0, "design_speed": 20} SEGMENTS = [ {"year": 2026, "start": (60.0, 200.0), "end": (560.123456789, 230.987654321)}, {"year": 2027, "start": (560.123456789, 230.987654321), "end": (1040.0, 160.0)}, ] def _hill() -> Terrain: x = np.arange(0.0, 1102.0, 2.0) y = np.arange(0.0, 402.0, 2.0) xx, yy = np.meshgrid(x, y) z = 30.0 * np.exp(-((xx - 300.0) ** 2 + (yy - 210.0) ** 2) / 80.0**2) + 0.02 * xx return Terrain(x, y, z, np.ones(xx.shape, dtype=bool)) def hill_loader(root, selection, models_dir, cache_dir): """나란히 셈 프로세스가 이름으로 부르는 지형(모듈 맨 위 함수).""" return _hill(), [] @pytest.fixture(scope="module") def hill() -> Terrain: return _hill() @pytest.fixture def run(tmp_path, hill): _POOLS.clear() calls = [] def loader(root, selection, models_dir, cache_dir): calls.append((models_dir.name, cache_dir.name)) return hill, [] def go(criteria=CRITERIA, corridor_m=150.0): return analyze.analyze_plan( tmp_path, SEGMENTS, criteria, loader=loader, corridor_m=corridor_m, workers=1 ) go.calls = calls yield go _POOLS.clear() def test_boundary_chainage_and_line(run) -> None: results = run() # 연도 구간마다 제 DEM · 제 비용면 캐시 assert run.calls == [("models_2026", "cost_2026"), ("models_2027", "cost_2027")] for seg, result in zip(SEGMENTS, results): assert set(result["routes"]) == set(analyze.PLAN_MODES) for route in result["routes"].values(): assert "error" not in route # 시종점 그대로 — 연도 경계가 정확히 맞물림 assert route["planned"][0] == result["start"] == pytest.approx(seg["start"], abs=1e-4) assert route["planned"][-1] == result["end"] == pytest.approx(seg["end"], abs=1e-4) assert results[0]["routes"]["gradeline"]["planned"][-1] == results[1]["start"] rows = analyze.chain(results, {2026: "earthwork", 2027: None}, 1234.5) assert [r["mode"] for r in rows] == ["earthwork", "gradeline"] assert rows[0]["start_chainage_m"] == 1234.5 assert rows[1]["start_chainage_m"] == rows[0]["end_chainage_m"] for result, row in zip(results, rows): length = result["routes"][row["mode"]]["metrics"]["length_m"] assert row["end_chainage_m"] - row["start_chainage_m"] == pytest.approx(length, abs=1e-3) line = analyze.joined_line(results, rows) first = results[0]["routes"]["earthwork"]["planned"] second = results[1]["routes"]["gradeline"]["planned"] assert len(line) == len(first) + len(second) - 1 # 경계점 한 번만(꺾인 채) assert line[len(first) - 1] == results[0]["end"] == results[1]["start"] def test_same_input_reads_store(run, monkeypatch) -> None: first = run() assert [r["stored"] for r in first] == [False, False] def boom(*args, **kwargs): raise AssertionError("저장본이 있는데 다시 셈") monkeypatch.setattr(analyze, "generate_initial_route", boom) again = run() assert [r["stored"] for r in again] == [True, True] assert again[1]["routes"] == first[1]["routes"] # 입력(복도 · 기준)이 바뀌면 다시 셈 with pytest.raises(AssertionError, match="다시 셈"): run(corridor_m=120.0) with pytest.raises(AssertionError, match="다시 셈"): run(criteria={**CRITERIA, "max_grade": 0.12}) def test_parallel_years_match_sequential(tmp_path) -> None: """연도 구간 나란히(프로세스 둘) = 차례로 — 갈래 넷 노선 좌표 · 잰 값 차 0(531-6).""" _POOLS.clear() seq = analyze.analyze_plan( tmp_path / "a", SEGMENTS, CRITERIA, loader=hill_loader, corridor_m=150.0, workers=1 ) _POOLS.clear() steps = [] par = analyze.analyze_plan( tmp_path / "b", SEGMENTS, CRITERIA, loader=hill_loader, corridor_m=150.0, workers=2, on_progress=lambda p, m: steps.append(m), ) assert [r["routes"] for r in par] == [r["routes"] for r in seq] assert [r["fingerprint"] for r in par] == [r["fingerprint"] for r in seq] assert "나란히 2" in steps[0] and steps[-1] == "분석 끝" # 저장본은 제 프로세스에서 지문만 보고 그대로 again = analyze.analyze_plan( tmp_path / "b", SEGMENTS, CRITERIA, loader=hill_loader, corridor_m=150.0, workers=2 ) assert [r["stored"] for r in again] == [True, True] _POOLS.clear() def test_near_streams_keeps_every_nearest_line(hill) -> None: """계류 거르기 — 지형 상자 안 점의 계류 거리 · 건너기가 통째와 같고 먼 선은 빠짐(531-6).""" import shapely from shapely.geometry import LineString, MultiLineString rng = np.random.default_rng(7) streams = [] for _ in range(80): # 지형(1.1 × 0.4 km) 둘레 5 km 에 흩은 계류 start = rng.uniform(-2000.0, 3000.0, 2) streams.append([tuple(start + step) for step in np.cumsum(rng.normal(0, 60, (6, 2)), 0)]) kept = analyze.near_streams(hill, streams) assert 0 < len(kept) < len(streams) points = shapely.points(rng.uniform((0.0, 0.0), (1100.0, 400.0), (4000, 2))) whole, part = MultiLineString(streams), MultiLineString(kept) assert np.array_equal(shapely.distance(points, whole), shapely.distance(points, part)) route = LineString([(0.0, 0.0), (1100.0, 400.0)]) assert route.intersection(whole).equals(route.intersection(part)) assert analyze.near_streams(hill, []) == [] def test_corridor_argument_keeps_b_default(hill) -> None: line = [SEGMENTS[0]["start"], SEGMENTS[0]["end"]] base = SearchGrid(line, hill, [], stream_offset_m=10.0) wide = SearchGrid(line, hill, [], stream_offset_m=10.0, corridor_m=180.0) assert base.corridor_m == CORRIDOR_M == 100.0 assert wide.cells > base.cells def test_criteria_from_project() -> None: from B05_Profile.B05_Profile_Engine_Grade import ( legal_grade_limit_pct, legal_plan_radius_min_m, ) got = analyze.plan_criteria("work", 99) # 못 고르는 속도 → 종류 기본 assert got["design_speed"] == 20 assert got["max_grade"] == legal_grade_limit_pct("work", "special", False, 20) / 100.0 assert got["min_radius_m"] == legal_plan_radius_min_m(20, "special")