Merge remote-tracking branch 'origin/sub_laptop_1' into main_laptop_1

resources/tester/test_364_4_precompute_on_open.py

@@ -0,0 +1,104 @@
+"""364-4 노선 추천 미리 셈 — B05 열 때 예약 · 겹쳐 셈 막음 · 모달 진행 표시.
+
+    ./venv/Scripts/python.exe -m pytest resources/tester/test_364_4_precompute_on_open.py -q
+
+옛 프로젝트 · 판이 바뀐 저장본은 사용자가 처음 누를 때 기다렸음(요구 239 · 토공 62.9 s).
+`GET /route/plan`(B05 열기)이 미리 셈을 예약 · 도는 동안 또 오면 끝난 뒤 한 번 더 · 진행은
+`GET /route/initial/progress` · 모달 상태줄이 그것을 붙임.
+"""
+
+import asyncio
+from pathlib import Path
+from uuid import uuid4
+
+from B05_Profile import B05_Profile_Router_RouteInitial as router
+
+ROOT = Path(__file__).resolve().parents[2]
+
+
+def test_겹쳐_셈_막음_끝난_뒤_한_번_더(monkeypatch) -> None:
+    calls: list[str] = []
+    seen: list[dict] = []
+    pid = uuid4()
+
+    async def precompute(project_id):
+        calls.append(str(project_id))
+        progress = router._RUNNING[str(project_id)]
+        progress.update(total=5, done=2, mode="earthwork")
+        seen.append(router.precompute_progress(project_id))
+        await asyncio.sleep(0.05)
+
+    monkeypatch.setattr(router, "precompute_defaults", precompute)
+
+    async def main() -> None:
+        first = asyncio.create_task(router.run_precompute(pid))
+        await asyncio.sleep(0.01)
+        # 도는 중 두 번 더 와도 셈은 한 번 더만
+        await router.run_precompute(pid)
+        await router.run_precompute(pid)
+        await first
+
+    asyncio.run(main())
+    assert calls == [str(pid), str(pid)]
+    assert seen[0] == {"running": True, "done": 2, "total": 5, "mode": "earthwork"}
+    assert router.precompute_progress(pid) == {
+        "running": False,
+        "done": 0,
+        "total": 0,
+        "mode": None,
+    }
+
+
+def test_실패해도_진행_비움(monkeypatch) -> None:
+    async def broken(_pid):
+        raise RuntimeError("셈 실패")
+
+    monkeypatch.setattr(router, "precompute_defaults", broken)
+    pid = uuid4()
+    try:
+        asyncio.run(router.run_precompute(pid))
+    except RuntimeError:
+        pass
+    assert not router.precompute_progress(pid)["running"]
+
+
+def test_b05_열기_예약_모달_진행_배선() -> None:
+    replan = (ROOT / "B05_Profile/B05_Profile_Router_Replan.py").read_text(encoding="utf-8")
+    plan = replan[replan.index("async def read_route_plan") :]
+    assert "schedule_precompute(project_id)" in plan[: plan.index("\n@router")]
+    source = (ROOT / "B05_Profile/B05_Profile_Router_RouteInitial.py").read_text(encoding="utf-8")
+    assert '"/{project_id}/route/initial/progress"' in source
+    # 체인 · B01 범위도 같은 한 줄(run_precompute)로
+    assert "run_precompute(project_id), task_name" in source
+    assert "await run_precompute(project_id)" in source
+    api = (ROOT / "B05_Profile/B05_Profile_Api_RouteInitial.ts").read_text(encoding="utf-8")
+    assert "/route/initial/progress" in api
+    initial = (ROOT / "B05_Profile/B05_Profile_UI_RouteEdit_Initial.ts").read_text(encoding="utf-8")
+    assert "progress: () => fetchPrecomputeProgress(projectId)" in initial
+    modal = (ROOT / "B05_Profile/B05_Profile_UI_RouteEdit_Recommend.ts").read_text(encoding="utf-8")
+    assert "미리 셈하는 중 ${p.done}/${p.total}" in modal
+
+
+def test_모달이_여는_갈래_먼저(monkeypatch) -> None:
+    """저장된 갈래(추적 아님)를 먼저 셈 — 모달이 열자마자 그 갈래를 물음."""
+    from B05_Profile.B05_Profile_Engine_RouteInitial_Store import DEFAULT_MODES
+
+    asked: list[str] = []
+
+    async def paths(_pid):
+        return (ROOT, None)
+
+    async def compute(_pid, request):
+        asked.append(request.mode)
+        return {}
+
+    monkeypatch.setattr(router.replan, "_project_paths", paths)
+    monkeypatch.setattr(
+        router,
+        "read_initial_choice",
+        lambda _root: {"logic": "gentle", "weights": None, "criteria": None},
+    )
+    monkeypatch.setattr(router, "compute_initial", compute)
+    asyncio.run(router.precompute_defaults(uuid4()))
+    assert asked[0] == "gentle" and asked[-1] == "follow"
+    assert sorted(asked) == sorted(DEFAULT_MODES)

resources/tester/test_364_5_crowded_curves.py

@@ -0,0 +1,150 @@
+"""364-5 한 측점을 다투는 가까운 두 곡선 — `B05_Profile_Engine_RouteInitial_Station._crowded`.
+
+    ./venv/Scripts/python.exe -m pytest resources/tester/test_364_5_crowded_curves.py -q -s
+
+자료 = 브레인 검증 판(요구 239) 다섯 갈래의 맞추기 전 선(`fixtures/b05_364_5_unsnapped_req239.json`)
++ 검증_sub1 지형. 측점 밖 곡선 수 · 넘은 곳 · 최대 기울기를 맞추기 전 · 옛 맞춤 · 새 맞춤으로 찍고
+이웃 MC 20 m 안 측점 밖 쌍 0 · 측점 밖 합계가 옛 맞춤의 1/4 이하 · 이탈 한도 안을 봄.
+"""
+
+import json
+from pathlib import Path
+
+import numpy as np
+import pytest
+from scipy.interpolate import RegularGridInterpolator
+from shapely.geometry import LineString
+
+from B05_Profile.B05_Profile_Engine_RouteInitial import Terrain
+from B05_Profile.B05_Profile_Engine_RouteInitial_Metrics import (
+    SAMPLE_M,
+    _resample,
+    route_metrics,
+    smoothed_ground,
+)
+from B05_Profile.B05_Profile_Engine_RouteInitial_Station import (
+    MC_TOL_M,
+    STATION_M,
+    curve_mc_chainages,
+    off_station_count,
+    snap_curves_to_stations,
+)
+from common_util.common_util_route_geometry import read_planned_route_csv
+from common_util.common_util_route_polyline import build_planned_polyline
+from config.config_system_design import ROUTE_STATION_SNAP_MAX_GAP_M
+
+HERE = Path(__file__).resolve().parent
+PROJECT = HERE.parents[1] / "storage/1/3/7e420ed6-f940-4d12-a1a5-39cdcf70a391"
+SURFACE = PROJECT / "B05_Profile/route/cost_surface_classification_dtm_smooth.npz"
+CASES = json.loads((HERE / "fixtures/b05_364_5_unsnapped_req239.json").read_text(encoding="utf-8"))
+
+
+def _ls(vertices: list) -> LineString:
+    return LineString([xy for i, xy in enumerate(vertices) if not i or xy != vertices[i - 1]])
+
+
+def _outline(case: dict):
+    nodes = case["nodes"]
+    return build_planned_polyline(
+        [(n["x"], n["y"]) for n in nodes],
+        min_radius_m=case["criteria"]["min_radius_m"],
+        simplify=False,
+        curve_flags=[n["radius_m"] is not None for n in nodes],
+        radii=[n["radius_m"] for n in nodes],
+    )
+
+
+def _excess(terrain: Terrain, max_grade: float, smooth: float):
+    """`generate_initial_route.grade_excess` 와 같은 잣대(20 m 고른 지반 기울기 − 상한)."""
+    surface = RegularGridInterpolator(
+        (terrain.y, terrain.x),
+        smoothed_ground(terrain, smooth),
+        bounds_error=False,
+        fill_value=None,
+    )
+
+    def excess(points):
+        xy, chain = _resample(points, SAMPLE_M)
+        z = surface(xy[:, ::-1])
+        span = min(int(round(STATION_M / SAMPLE_M)), len(z) - 1)
+        if span < 1:
+            return 0.0
+        run = np.maximum(chain[span:] - chain[:-span], 1e-9)
+        return max(0.0, float((np.abs(z[span:] - z[:-span]) / run).max()) - max_grade)
+
+    return excess
+
+
+def _off_mcs(outline) -> list[float]:
+    return [
+        mc
+        for mc in curve_mc_chainages(outline)
+        if abs(mc - STATION_M * round(mc / STATION_M)) > MC_TOL_M
+    ]
+
+
+def test_req239_five_modes_before_after() -> None:
+    if not SURFACE.is_file():
+        pytest.skip("검증_sub1 storage 없음")
+    cached = np.load(SURFACE, allow_pickle=False)
+    terrain = Terrain(cached["x"], cached["y"], cached["z"], cached["valid_mask"])
+    expected = [
+        (v.x, v.y)
+        for v in read_planned_route_csv(PROJECT / "B05_Profile/route/expected_route.csv").vertices
+    ]
+    old_total = new_total = 0
+    for case in CASES:
+        cr = case["criteria"]
+        before = _outline(case)
+        follow = case["mode"] == "follow"
+        excess = (
+            None if follow else _excess(terrain, cr["max_grade_pct"] / 100, cr["grade_smooth_m"])
+        )
+        after = snap_curves_to_stations(before, cr["min_radius_m"], excess=excess)
+        assert after is not None
+
+        def grade(outline):
+            m = route_metrics(
+                outline.vertices,
+                expected,
+                terrain,
+                [],
+                outline.curves,
+                max_grade=cr["max_grade_pct"] / 100,
+                min_radius_m=cr["min_radius_m"],
+                stream_offset_m=cr["stream_offset_m"],
+                road_width_m=cr["road_width_m"],
+                bound_pct=None,
+                grade_smooth_m=cr["grade_smooth_m"],
+            )
+            return m["over_limit"]["total"], m["max_grade_pct"]
+
+        (o0, g0), (o1, g1) = grade(before), grade(after)
+        off = off_station_count(after)
+        print(
+            f"{case['mode']}: 측점 밖 {case['old']['off']}/{case['old']['curves']}(옛)"
+            f" → {off}/{len(after.curves)} · 넘은 곳 {o0} → {o1}"
+            f" · 최대 기울기 {g0:.1f} → {g1:.1f} %(맞추기 전 → 새)"
+        )
+        old_total += case["old"]["off"]
+        new_total += off
+        # 다투는 쌍 없음 — 이웃 측점 밖 곡선 MC 가 20 m 안이면 한 측점을 다툼
+        mcs = _off_mcs(after)
+        assert all(b - a >= STATION_M for a, b in zip(mcs, mcs[1:])), mcs
+        gap = _ls(after.vertices).hausdorff_distance(_ls(before.vertices))
+        assert gap <= ROUTE_STATION_SNAP_MAX_GAP_M + 0.05
+        assert after.vertices[0] == pytest.approx(before.vertices[0], abs=1e-3)
+        assert after.vertices[-1] == pytest.approx(before.vertices[-1], abs=1e-3)
+    print(f"측점 밖 곡선 합 {old_total} → {new_total}")
+    assert new_total * 4 <= old_total
+
+
+def test_crowded_pair_drops_one() -> None:
+    """11 m 사이 두 곡선 · 이탈 한도 1 m(측점 옮김 막힘) — 옛 맞춤은 둘 다 그 자리(44.8 · 55.8)."""
+    points = [(0.0, 0.0), (45.0, 0.0), (54.0, 7.0), (58.0, 70.0)]
+    before = build_planned_polyline(
+        points, min_radius_m=12.0, simplify=False, curve_flags=[True] * 4
+    )
+    after = snap_curves_to_stations(before, 12.0, max_gap_m=1.0)
+    assert after is not None and len(after.curves) == 1
+    assert _ls(after.vertices).hausdorff_distance(_ls(before.vertices)) <= 1.0 + 0.05