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

resources/tester/test_224_3_recommend_store.py

@@ -96,7 +96,16 @@ def test_no_overwrite(tmp_path: Path) -> None:
     assert not write_stored(tmp_path, "k", "p", {}, {"n": 2}, False)
     data = json.loads((store_dir(tmp_path) / "k.json").read_text(encoding="utf-8"))
     assert data["result"] == {"n": 1} and data["default"] is True
-    assert set(data) == {"version", "key", "pool", "default", "created_at", "request", "result"}
+    assert set(data) == {
+        "version",
+        "source",
+        "key",
+        "pool",
+        "default",
+        "created_at",
+        "request",
+        "result",
+    }
 
 
 def test_survives_reset_and_goes_with_data_swap(tmp_path: Path) -> None:
@@ -117,6 +126,8 @@ def test_survives_reset_and_goes_with_data_swap(tmp_path: Path) -> None:
 class _Calls:
     def __init__(self) -> None:
         self.count = 0
+        self.inputs: list[dict] = []
+        self.standard: dict | None = None
 
 
 @pytest.fixture
@@ -141,20 +152,26 @@ def wired(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> _Calls:
     async def surface_params(_connection, _pid):
         return {}
 
+    async def project_standard(_pid):
+        return calls.standard
+
     class _Pool:
         @asynccontextmanager
         async def acquire(self):
             yield None
 
-    def generate(expected, terrain, streams, *, pipes, **_inputs):
+    def generate(expected, terrain, streams, *, pipes, **inputs):
         assert pipes is None  # 관은 키 밖 — 덮개 없이 셈해 남김
         calls.count += 1
+        calls.inputs.append(inputs)
         return {
             "planned": [[x, y] for x, y in expected],
             "metrics": {
                 "length_m": 38.0,
                 "offset_avg_m": 0.0,
                 "profile": {
+                    "cut_m3": 10.0,
+                    "fill_m3": 4.0,
                     "stations_m": [0.0, 38.0],
                     "plan_m": [1.0, 1.0],
                     "ground_m": [0.0, 0.0],
@@ -167,6 +184,7 @@ def wired(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> _Calls:
     monkeypatch.setattr(replan, "_road_settings", road_settings)
     monkeypatch.setattr(replan, "_vertices_of", lambda _path: [list(p) for p in EXPECTED])
     monkeypatch.setattr(router, "get_db_pool", lambda: _Pool())
+    monkeypatch.setattr(router, "project_standard", project_standard)
     monkeypatch.setattr(surface, "get_surface_confirmation_params", surface_params)
     monkeypatch.setattr(surface, "analysis_terrain_params", lambda selection, _t: selection)
     monkeypatch.setattr(grade, "legal_grade_limit_pct", lambda *_a: 14.0)
@@ -225,7 +243,8 @@ def test_precompute_defaults_then_hits(wired: _Calls) -> None:
     ]
     assert sorted(item["request"]["mode"] for item in saved) == sorted(DEFAULT_MODES)
     assert all(item["default"] for item in saved)
-    # 화면 첫 누름(비중은 화면이 50/25/25 를 채워 보냄 · 추적은 종단 끔) — 셈 없이 저장본
+    # 화면 첫 누름(비중은 화면이 50/25/25 를 채워 보냄 · 다섯 갈래 종단 켬 · 420) — 셈 없이 저장본
+    assert all(item["request"]["with_profile"] for item in saved)
     hit = _ask(mode="weighted", weights={"expected": 50, "earthwork": 25, "grade": 25})
-    follow = _ask(mode="follow", with_profile=False)
+    follow = _ask(mode="follow")
     assert hit["stored"] and follow["stored"] and wired.count == 5

resources/tester/test_364_4_precompute_on_open.py

@@ -90,6 +90,7 @@ def test_모달이_여는_갈래_먼저(monkeypatch) -> None:
 
     async def compute(_pid, request):
         asked.append(request.mode)
+        assert request.with_profile  # 다섯 갈래 모두 종단 켬(420)
         return {}
 
     monkeypatch.setattr(router.replan, "_project_paths", paths)

resources/tester/test_420_1_cross_earth.py

@@ -0,0 +1,243 @@
+"""420 추천 횡단 토공 · 저장본 판 · [재계산] 서버(요구 254).
+
+    ./venv/Scripts/python.exe -m pytest resources/tester/test_420_1_cross_earth.py -q
+
+횡단 토공 = 공식 횡단 단면적 평균단면법(저장 정본 ±6 %) · 차 = 바뀐 측점만 · 결과 표 절 · 성 =
+횡단 · [재계산] = 잠금 안 지우고 새로 셈(셈 수 1 → 2) · 옛 판 · 옛 소스 지문 미리 셈 때 지움 ·
+표준 횡단면 = 저장 패널값(키 · 엔진에 감) · `initial_choice` with_profile 되살림.
+"""
+
+import asyncio
+import json
+from pathlib import Path
+from uuid import uuid4
+
+import numpy as np
+import pytest
+from test_224_3_recommend_store import EXPECTED, STREAMS, _ask, _key, _terrain, wired  # noqa: F401
+
+from B05_Profile import B05_Profile_Router_RouteInitial as router
+from B05_Profile.B05_Profile_Engine_RouteInitial import (
+    Terrain,
+    read_initial_choice,
+    write_initial_choice,
+)
+from B05_Profile.B05_Profile_Engine_RouteInitial_CrossEarth import (
+    attach_cross_earth,
+    cross_earthwork,
+    cross_earthwork_delta,
+)
+from B05_Profile.B05_Profile_Engine_RouteInitial_Profile import _station_areas, route_sections
+from B05_Profile.B05_Profile_Engine_RouteInitial_Store import (
+    ALGO_VERSION,
+    DEFAULT_MODES,
+    source_fingerprint,
+    store_dir,
+)
+
+ROOT = Path(__file__).resolve().parents[2]
+SUB2 = ROOT / "storage/1/3/cb15892a-5788-4a26-9316-20a15421f68b"
+
+
+def _slope() -> Terrain:
+    """옆경사 30 % 사면(노선 = x 방향) — 절 · 성 둘 다 생김."""
+    x = np.arange(0.0, 200.0, 2.0)
+    y = np.arange(-80.0, 80.0, 2.0)
+    xx, yy = np.meshgrid(x, y)
+    return Terrain(x, y, 0.05 * xx + 0.3 * yy, np.ones(xx.shape, dtype=bool))
+
+
+LINE = [(20.0, 0.0), (100.0, 0.0), (180.0, 0.0)]
+
+
+def test_횡단_토공_평균단면법() -> None:
+    terrain = _slope()
+    earth = cross_earthwork(LINE, None, None, terrain)
+    rows = earth["per_station"]
+    assert [r["chainage_m"] for r in rows] == [float(s) for s in range(0, 161, 20)]
+    # 직접 셈한 단면적 · Σ (A_i + A_i+1) / 2 · L
+    at, ground, sections, modes, widen = route_sections(LINE, terrain)
+    areas = [_station_areas(sections[i], modes[i], ground[i], widen[i]) for i in range(len(at))]
+    cut = sum((areas[i][0] + areas[i + 1][0]) / 2 * 20.0 for i in range(len(at) - 1))
+    fill = sum((areas[i][2] + areas[i + 1][2]) / 2 * 20.0 for i in range(len(at) - 1))
+    assert earth["cut"] == pytest.approx(cut, abs=0.1) and earth["cut"] > 0
+    assert earth["fill"] == pytest.approx(fill, abs=0.1) and earth["fill"] > 0
+    # 계획고를 1 m 올리면 성토가 늚 · 측점 수와 다른 계획고는 거절
+    up = cross_earthwork(LINE, None, np.asarray(ground) + 1.0, terrain)
+    assert up["fill"] > earth["fill"] and up["cut"] < earth["cut"]
+    with pytest.raises(ValueError):
+        cross_earthwork(LINE, None, [1.0, 2.0], terrain)
+
+
+def test_표준_횡단면이_단면적을_바꿈() -> None:
+    from config.config_system_design import STANDARD_CROSS_SECTION
+
+    terrain = _slope()
+    wide = json.loads(json.dumps(STANDARD_CROSS_SECTION))
+    for group in wide.values():
+        group["road_width_m"] += 2.0
+    base = cross_earthwork(LINE, None, None, terrain)
+    assert cross_earthwork(LINE, None, None, terrain, wide)["cut"] > base["cut"]
+
+
+def test_토공_차_바뀐_측점만() -> None:
+    terrain = _slope()
+    moved = [(20.0, 0.0), (100.0, 6.0), (180.0, 0.0)]
+    changed = [60.0, 80.0, 100.0]
+    same = cross_earthwork_delta(LINE, LINE, changed, terrain)
+    assert same["cut"] == 0 and same["fill"] == 0 and same["stations"] == 3
+    delta = cross_earthwork_delta(LINE, moved, changed, terrain)
+    old = cross_earthwork(LINE, changed, None, terrain)["per_station"]
+    new = cross_earthwork(moved, changed, None, terrain)["per_station"]
+    expect = 20.0 * sum(n["fill_m2"] - o["fill_m2"] for n, o in zip(new, old))
+    assert delta["fill"] == pytest.approx(expect, abs=0.2)
+    # 옛 결과를 넘기면 옛 단면적을 다시 안 셈 · 전체 어림 = 옛 합 + 차
+    whole = cross_earthwork(LINE, None, None, terrain)
+    reuse = cross_earthwork_delta(whole, moved, changed, terrain)
+    assert reuse["fill"] == pytest.approx(delta["fill"], abs=0.2)
+    assert reuse["fill_total"] == pytest.approx(whole["fill"] + reuse["fill"], abs=0.2)
+    assert cross_earthwork_delta(LINE, moved, [], terrain)["earth"] == 0
+
+
+@pytest.mark.skipif(not (SUB2 / "B05_Profile").is_dir(), reason="검증_sub2 storage 없음")
+def test_검증_sub2_저장_정본_6퍼센트() -> None:
+    """저장된 노선 측점 · 계획고 = 지반 — 공식 [저장] 단면적 합과 ±6 %(2 m 추천 지형 · 1 m 표본)."""
+    from B05_Profile.B05_Profile_Engine_RouteInitial import load_terrain_and_streams
+    from B05_Profile.B05_Profile_Engine_Solver import _MODELS_SUBDIR
+    from common_util.common_util_route_geometry import read_planned_route_csv
+    from common_util.common_util_surface_sampler import build_surface_sampler
+
+    route = read_planned_route_csv(SUB2 / "B05_Profile/route/planned_route_initial.csv")
+    curves = json.loads(
+        (SUB2 / "B05_Profile/route/planned_route_initial_curves.json").read_text("utf-8")
+    )
+    curves = curves.get("curves") if isinstance(curves, dict) else curves
+    line = [(v.x, v.y) for v in route.vertices]
+    files = sorted((SUB2 / "B06_Section/cross_sections").glob("cross_*.json"))
+    at = [json.loads(p.read_text("utf-8"))["chainage_m"] for p in files]
+    # 공식 [저장] 단면적(DB cross_sections.design · 2026-10-18 뜸 · 계획고 = 지반) 평균단면법 합
+    stored_cut, stored_fill = 5166.0, 23079.3
+    selection = {"source_filter": "classification", "method": "dtm", "smooth": True}
+    terrain, _ = load_terrain_and_streams(SUB2, selection)
+    surface = build_surface_sampler(SUB2 / _MODELS_SUBDIR, "classification", "dtm", True)
+    for ground in (terrain, surface):
+        earth = cross_earthwork(line, at, None, ground, design_curves=curves)
+        assert earth["cut"] == pytest.approx(stored_cut, rel=0.06)
+        assert earth["fill"] == pytest.approx(stored_fill, rel=0.06)
+
+
+def test_결과_절성_횡단_값() -> None:
+    terrain = _slope()
+    plain = {"planned": [list(p) for p in LINE], "curves": [], "metrics": {"cut_m3": 1.0}}
+    profiled = {
+        "planned": [list(p) for p in LINE],
+        "metrics": {"cut_m3": 1.0, "fill_m3": 2.0, "profile": {"cut_m3": 30.0, "fill_m3": 20.0}},
+    }
+    result = {**plain, "unsnapped": profiled}
+    attach_cross_earth(result, terrain)
+    earth = cross_earthwork(LINE, None, None, terrain)
+    assert result["metrics"]["cut_m3"] == earth["cut"] and result["metrics"]["band_cut_m3"] == 1.0
+    assert result["metrics"]["surplus_m3"] == pytest.approx(earth["cut"] - earth["fill"], abs=0.2)
+    shown = result["unsnapped"]["metrics"]
+    assert (shown["cut_m3"], shown["fill_m3"], shown["band_fill_m3"]) == (30.0, 20.0, 2.0)
+    attach_cross_earth(result, terrain)  # 두 번 붙여도 띠 가늠을 안 덮음
+    assert result["metrics"]["band_cut_m3"] == 1.0
+
+
+def test_표준_횡단면_키_엔진에_감(wired) -> None:  # noqa: F811
+    assert _key(standard={"a": 1}) != _key() and _key(standard={"a": 1}) == _key(standard={"a": 1})
+    wired.standard = {"road": {"width_m": 5.0}}
+    _ask(mode="earthwork")
+    assert wired.inputs[-1]["standard"] == wired.standard
+
+
+def test_재계산_잠금_안_지우고_새로(wired) -> None:  # noqa: F811
+    first = _ask(mode="earthwork")
+    _ask(mode="gentle")
+    _ask(mode="earthwork", criteria={"min_radius_m": 15.0})  # 다른 묶음
+    assert wired.count == 3 and first["stored"] is False
+    files = {p.name: p.read_text("utf-8") for p in store_dir(wired.root).glob("*.json")}
+    again = _ask(mode="earthwork", recompute=True)
+    assert again["stored"] is False and wired.count == 4  # 셈 1 → 2(earthwork)
+    now = {p.name: json.loads(p.read_text("utf-8")) for p in store_dir(wired.root).glob("*.json")}
+    # 같은 묶음 gentle 은 지워짐 · 다른 묶음은 그대로 · earthwork 파일은 새로 씀
+    assert len(now) == 2
+    other = [n for n, d in now.items() if d["request"]["criteria"]]
+    assert other and now[other[0]] == json.loads(files[other[0]])
+    fresh = next(d for n, d in now.items() if not d["request"]["criteria"])
+    assert "recompute" not in fresh["request"] and fresh["version"] == ALGO_VERSION
+    assert fresh["result"]["metrics"]["cut_m3"] == 10.0  # 횡단(균형 종단) 값
+    assert _ask(mode="earthwork")["stored"] and wired.count == 4
+
+
+def test_재계산_메모리_묶음도_버림(wired) -> None:  # noqa: F811
+    from B05_Profile.B05_Profile_Engine_RouteInitial import _POOLS
+    from B05_Profile.B05_Profile_Engine_RouteInitial_Store import request_fingerprint
+
+    _ask(mode="earthwork")
+    pool = request_fingerprint(
+        EXPECTED,
+        _terrain(),
+        STREAMS,
+        mode="earthwork",
+        weights=None,
+        max_grade=0.14,
+        min_radius_m=12.0,
+        stream_offset_m=None,
+        grade_smooth_m=None,
+        target_grade=None,
+        with_profile=True,
+        design_speed=20,
+    )[1]
+    _POOLS[bytes.fromhex(pool)] = "옛 묶음"
+    _ask(mode="earthwork")
+    assert bytes.fromhex(pool) in _POOLS
+    _ask(mode="earthwork", recompute=True)
+    assert bytes.fromhex(pool) not in _POOLS
+
+
+def test_옛_판_옛_지문_미리_셈_때_지움(wired) -> None:  # noqa: F811
+    folder = store_dir(wired.root)
+    folder.mkdir(parents=True)
+    old = {"version": ALGO_VERSION - 1, "source": source_fingerprint(), "request": {}}
+    stale = {"version": ALGO_VERSION, "source": "0" * 16, "request": {}}
+    (folder / "old.json").write_text(json.dumps(old), encoding="utf-8")
+    (folder / "stale.json").write_text(json.dumps(stale), encoding="utf-8")
+    (folder / "broken.json").write_text("{", encoding="utf-8")
+    asyncio.run(router.precompute_defaults(uuid4()))
+    names = {p.stem for p in folder.glob("*.json")}
+    assert not names & {"old", "stale", "broken"} and len(names) == len(DEFAULT_MODES)
+    assert wired.count == len(DEFAULT_MODES)
+
+
+def test_소스_지문이_바뀌면_다른_키(monkeypatch: pytest.MonkeyPatch) -> None:
+    import B05_Profile.B05_Profile_Engine_RouteInitial_Store as store
+
+    before = _key()
+    monkeypatch.setattr(store, "source_fingerprint", lambda: "f" * 16)
+    assert _key() != before
+
+
+def test_with_profile_되살림(tmp_path: Path) -> None:
+    assert read_initial_choice(tmp_path)["with_profile"] is True  # 없으면 켬
+    write_initial_choice(tmp_path, "gentle", None, None, False)
+    assert read_initial_choice(tmp_path)["with_profile"] is False
+    write_initial_choice(tmp_path, "gentle", None, None)
+    assert read_initial_choice(tmp_path)["with_profile"] is True
+    # 옛 파일(칸 없음) = 켬
+    path = tmp_path / "B05_Profile/route/initial_choice.json"
+    path.write_text(json.dumps({"logic": "gentle"}), encoding="utf-8")
+    assert read_initial_choice(tmp_path)["with_profile"] is True
+
+
+def test_적용이_종단_반영을_적음_소스() -> None:
+    source = (ROOT / "B05_Profile/B05_Profile_Router_Replan.py").read_text(encoding="utf-8")
+    call = source[source.index("write_initial_choice,\n") :][:600]
+    assert "request.balanced_profile" in call
+    initial = (ROOT / "B05_Profile/B05_Profile_Router_RouteInitial.py").read_text(encoding="utf-8")
+    # 지우기는 잠금 안
+    lock = initial.index("async with pool_lock(pool_key):")
+    assert lock < initial.index("drop_pool_results, project_root, pool_key") < lock + 400
+    assert 'model_dump(exclude={"recompute"})' in initial
+    # [적용] 균형 종단도 저장 패널값
+    assert "standard=standard" in initial[initial.index("async def balanced_profile_of") :]

resources/tester/test_route_initial.py

@@ -321,6 +321,7 @@ def test_initial_choice_roundtrip(tmp_path) -> None:
         "logic": "follow",
         "weights": {"expected": 100.0, "earthwork": 0.0, "grade": 0.0},
         "criteria": None,
+        "with_profile": True,
     }
     for logic in ("gradeline", "safe", "crossings", "cost"):  # 새 갈래도 적고 다시 읽음(23-5)
         write_initial_choice(tmp_path, logic, None, {"target_grade_pct": 10.0})
@@ -331,6 +332,7 @@ def test_initial_choice_roundtrip(tmp_path) -> None:
         "logic": "weighted",
         "weights": {"expected": 60.0, "earthwork": 20.0, "grade": 20.0},
         "criteria": None,
+        "with_profile": True,
     }
     # 사용자가 화면 칸에 겹쳐 넣은 값만(빈 칸은 안 담김) — 남으면 다시 열 때 그 칸만 채운다.
     write_initial_choice(
@@ -340,6 +342,7 @@ def test_initial_choice_roundtrip(tmp_path) -> None:
         "logic": "earthwork",
         "weights": {"expected": 0.0, "earthwork": 100.0, "grade": 0.0},
         "criteria": {"stream_offset_m": 30.0, "grade_smooth_m": 2.0},
+        "with_profile": True,
     }
     (tmp_path / "B05_Profile" / "route" / "initial_choice.json").write_text("{", encoding="utf-8")
     assert read_initial_choice(tmp_path)["logic"] == "follow"