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

B05_Profile/B05_Profile_Engine_Solver.py

@@ -19,7 +19,11 @@ from B05_Profile.B05_Profile_Engine_Geometry import (
     resample_polyline_2d,
     single_segment_dijkstra,
 )
-from common_util.common_util_surface_sampler import check_surface_keys
+from common_util.common_util_surface_sampler import (
+    DtmGridSampler,
+    check_surface_keys,
+    implicit_grid_npz,
+)
 from config.config_system import (
     FOREST_ROAD_MAX_GRADE,
     FOREST_ROAD_MIN_CURVE_R_M,
@@ -71,7 +75,7 @@ def _sample_surface_on_grid(
     y_coords: np.ndarray,
     dtm_z: np.ndarray,
 ) -> np.ndarray:
-    """확정 지표면 모델의 표고를 DTM 격자에 샘플링한다(실패 시 DTM으로 폴백)."""
+    """확정 지표면 표고를 DTM 격자에 샘플링 — 파일 없음 · 깨짐은 까닭 그대로 던짐(763-1)."""
     check_surface_keys(filter_key, method)
     if method == "dtm":
         return dtm_z
@@ -114,24 +118,14 @@ def _sample_surface_on_grid(
             return _finalize(spline(y_coords, x_coords).ravel())
 
         if method == "implicit":
-            from scipy.interpolate import RBFInterpolator
-
-            d = np.load(models_dir / f"implicit_{filter_key}.npz")
-            centers = np.asarray(d["centers_xy"], dtype=np.float64)
-            cz = np.asarray(d["center_z"], dtype=np.float64)
-            smoothing = float(d["smoothing"][0]) if "smoothing" in d else 0.0
-            interp = RBFInterpolator(
-                centers,
-                cz,
-                neighbors=min(64, len(centers)),
-                smoothing=smoothing,
-                kernel="thin_plate_spline",
-            )
-            out = np.empty(len(query), dtype=np.float64)
-            for s in range(0, len(query), 50_000):
-                e = min(s + 50_000, len(query))
-                out[s:e] = interp(query[s:e])
-            return _finalize(out)
+            # 굳힌 1 m 격자(641-1) 마디 그대로 — B05 종횡단 표본기와 같은 값 · RBF 다시 안 셈(763-1)
+            grid = DtmGridSampler.open_cached(implicit_grid_npz(models_dir, filter_key))
+            ix = np.abs(grid.x[None, :] - np.asarray(x_coords)[:, None]).argmin(axis=1)
+            iy = np.abs(grid.y[None, :] - np.asarray(y_coords)[:, None]).argmin(axis=1)
+            off = max(np.abs(grid.x[ix] - x_coords).max(), np.abs(grid.y[iy] - y_coords).max())
+            if off > 1e-3:
+                raise ValueError(f"굳힌 격자 축이 DTM 과 다름: implicit_{filter_key}_grid.npz")
+            return _finalize(grid.z[np.ix_(iy, ix)])
 
         if method == "meshfree":
             from scipy.interpolate import griddata
@@ -140,8 +134,11 @@ def _sample_surface_on_grid(
             pts = np.asarray(d["points"], dtype=np.float64)
             z = griddata(pts[:, :2], pts[:, 2], query, method="linear")
             return _finalize(z)
-    except Exception:
-        return dtm_z
+    except FileNotFoundError as error:
+        name = Path(error.filename).name if error.filename else f"{method}_{filter_key}"
+        raise FileNotFoundError(f"지표면 모델 파일이 없습니다: {name}") from error
+    except Exception as error:  # noqa: BLE001 — 깨진 파일 = 까닭 그대로 409
+        raise ValueError(f"지표면 모델을 읽지 못함({method}_{filter_key}): {error}") from error
 
     return dtm_z
 
@@ -157,6 +154,9 @@ def _source_npz_paths(models_dir: Path, filter_key: str, method: str, smooth: bo
     if method != "dtm":
         candidates.append(models_dir / f"{method}_{filter_key}{suffix}.npz")
         candidates.append(models_dir / f"{method}_{filter_key}.npz")
+    # 비용면이 읽는 굳힌 격자 — 서명 전에 지음(763-1)
+    if method == "implicit" and candidates[-1].exists():
+        candidates.append(implicit_grid_npz(models_dir, filter_key))
     seen: set[Path] = set()
     out: list[Path] = []
     for p in candidates:

B05_Profile/B05_Profile_Router_RouteInitial.py

@@ -75,9 +75,10 @@ async def compute_initial(
     )
     from B05_Profile.B05_Profile_Router_RouteInitial_Initial import initial_profile_of
     from common_util.common_util_surface_confirmation import (
+        NOT_CONFIRMED,
         analysis_terrain_params,
+        confirmed_surface_params,
         design_terrain,
-        get_surface_confirmation_params,
     )
 
     weights = request.weights.model_dump() if request.weights else None
@@ -117,7 +118,9 @@ async def compute_initial(
     target_pct = criteria.target_grade_pct if criteria else None
     pool = get_db_pool()
     async with pool.acquire() as connection:
-        selection = await get_surface_confirmation_params(connection, str(project_id))
+        selection = await confirmed_surface_params(connection, str(project_id))
+    if selection is None:  # 확정 전 — 기본값으로 셈하지 않음(763-1)
+        return JSONResponse(status_code=409, content={"status": "error", "message": NOT_CONFIRMED})
     # 화면은 늘 분석 지형을 실음 — 설계 지형과 같으면 설계 지형(미리 셈 갈래 판정 · 505-1)
     picked = None if request.terrain == design_terrain(selection) else request.terrain
     selection = analysis_terrain_params(selection, picked)
@@ -320,16 +323,19 @@ async def precompute_defaults(project_id: UUID) -> None:
 
 
 async def precompute_terrains(project_id: UUID, project_root: Path) -> list[Any]:
-    """미리 셈 지형(509-5) — 설계 지형 먼저 · 추천 창이 고를 수 있는 다른 지형."""
+    """미리 셈 지형(509-5) — 설계 지형 먼저 · 추천 창이 고를 수 있는 다른 지형 · 확정 전 = []."""
     from B05_Profile.B05_Profile_Router_Terrain import _MODELS_SUBDIR, choosable_terrains
     from common_util.common_util_surface_confirmation import (
+        confirmed_surface_params,
         design_terrain,
-        get_surface_confirmation_params,
     )
 
     pool = get_db_pool()
     async with pool.acquire() as connection:
-        params = await get_surface_confirmation_params(connection, str(project_id))
+        params = await confirmed_surface_params(connection, str(project_id))
+        if params is None:  # 확정 전 — 셀 지형 없음(763-1)
+            logger.info("추천 미리 셈 건너뜀(지표면 확정 전): project_id=%s", project_id)
+            return []
         sources = await choosable_terrains(
             connection, project_id, params, Path(project_root) / _MODELS_SUBDIR
         )

common_util/common_util_surface_confirmation.py

@@ -75,6 +75,30 @@ async def get_surface_confirmation_params(
     return resolved
 
 
+#: 확정 전 추천 요청(763-1) — 기본값으로 슬쩍 셈하지 않음
+NOT_CONFIRMED = "지표면 확정 전 — B04 [확정] 뒤 추천이 생김"
+
+
+async def confirmed_surface_params(
+    connection: aiomysql.Connection,
+    project_id: str,
+) -> dict[str, Any] | None:
+    """확정된 지표면 params — stage 1 COMPLETE · params 에 source_filter 가 있을 때만 · 아니면 None
+    (추천 노선이 config 기본값으로 셈하지 않게 · 763-1)."""
+    async with connection.cursor(aiomysql.DictCursor) as cursor:
+        await cursor.execute(
+            "SELECT state, params FROM project_workflow_stages "
+            "WHERE project_id = %s AND stage_no = 1",
+            (project_id,),
+        )
+        row = await cursor.fetchone()
+    if not row or row.get("state") != "COMPLETE":
+        return None
+    if not _decode_params(row.get("params")).get("source_filter"):
+        return None
+    return await get_surface_confirmation_params(connection, project_id)
+
+
 def _is_sheet(params: dict[str, Any]) -> bool:
     return str(params.get("source_filter") or "").startswith("sheet_")
 

resources/tester/test_224_3_recommend_store.py

@@ -185,7 +185,7 @@ def wired(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> _Calls:
     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, "confirmed_surface_params", surface_params)
     monkeypatch.setattr(surface, "analysis_terrain_params", lambda selection, _t: selection)
     monkeypatch.setattr(grade, "legal_grade_limit_pct", lambda *_a: 14.0)
     monkeypatch.setattr(engine, "load_terrain_and_streams", lambda *_a: (_terrain(), STREAMS))

resources/tester/test_509_5_recommend_terrain.py

@@ -85,7 +85,7 @@ def test_미리_셈_지형_차례_설계_먼저(monkeypatch) -> None:
         return ["las", "sheet"]
 
     monkeypatch.setattr(router, "get_db_pool", lambda: _Pool())
-    monkeypatch.setattr(surface, "get_surface_confirmation_params", params)
+    monkeypatch.setattr(surface, "confirmed_surface_params", params)
     monkeypatch.setattr(terrain_router, "choosable_terrains", choosable)
     assert asyncio.run(router.precompute_terrains(uuid4(), ROOT)) == ["sheet", "las"]
 

resources/tester/test_763_1_confirmed_terrain.py

@@ -0,0 +1,143 @@
+"""763-1 추천노선 = 확정 지표면에서만 — 확정 전 409 · 비용면 = 굳힌 격자 마디 · DTM 대체 없음.
+
+    ./venv/Scripts/python.exe -m pytest resources/tester/test_763_1_confirmed_terrain.py -q
+
+거울: 비용면 implicit 높이 = `implicit_grid_npz`(B05 종횡단 표본기와 같은 파일) 마디 · 유효 칸 차 0.
+"""
+
+import asyncio
+import json
+from contextlib import asynccontextmanager
+from pathlib import Path
+from uuid import uuid4
+
+import numpy as np
+import pytest
+from test_224_3_recommend_store import _ask, wired  # noqa: F401 — 픽스처
+
+from B05_Profile import B05_Profile_Engine_Solver as solver
+from B05_Profile import B05_Profile_Router_RouteInitial as router
+from B05_Profile.B05_Profile_Engine_RouteInitial_Store import store_dir
+from common_util import common_util_surface_confirmation as surface
+from common_util.common_util_surface_sampler import implicit_grid_npz
+
+
+class _Cursor:
+    def __init__(self, row):
+        self.row = row
+
+    async def __aenter__(self):
+        return self
+
+    async def __aexit__(self, *_a):
+        return False
+
+    async def execute(self, *_a):
+        return None
+
+    async def fetchone(self):
+        return self.row
+
+
+class _Connection:
+    def __init__(self, row):
+        self.row = row
+
+    def cursor(self, *_a):
+        return _Cursor(self.row)
+
+
+def _confirmed(row):
+    return asyncio.run(surface.confirmed_surface_params(_Connection(row), "p"))
+
+
+def test_확정_판정_단계_완료와_필터_칸() -> None:
+    picked = {"source_filter": "classification", "method": "implicit", "smooth": False}
+    assert _confirmed(None) is None
+    assert _confirmed({"state": "IN_PROGRESS", "params": json.dumps(picked)}) is None
+    assert _confirmed({"state": "STALE", "params": json.dumps(picked)}) is None
+    assert _confirmed({"state": "COMPLETE", "params": None}) is None
+    assert _confirmed({"state": "COMPLETE", "params": json.dumps({"method": "dtm"})}) is None
+    got = _confirmed({"state": "COMPLETE", "params": json.dumps(picked)})
+    assert got["source_filter"] == "classification" and got["method"] == "implicit"
+
+
+def test_확정_전_추천_409_저장본_없음(wired, monkeypatch) -> None:  # noqa: F811
+    async def none(_c, _pid):
+        return None
+
+    monkeypatch.setattr(surface, "confirmed_surface_params", none)
+    for precompute in (True, False):
+        answer = _ask(precompute, mode="gentle")
+        assert answer.status_code == 409
+        assert json.loads(answer.body)["message"] == surface.NOT_CONFIRMED
+    assert wired.count == 0
+    assert not store_dir(wired.root).exists() or not list(store_dir(wired.root).glob("*"))
+
+
+def test_확정_전_미리_셈_지형_없음(monkeypatch) -> None:
+    class _Pool:
+        @asynccontextmanager
+        async def acquire(self):
+            yield None
+
+    async def none(_c, _pid):
+        return None
+
+    monkeypatch.setattr(router, "get_db_pool", lambda: _Pool())
+    monkeypatch.setattr(surface, "confirmed_surface_params", none)
+    assert asyncio.run(router.precompute_terrains(uuid4(), Path("."))) == []
+
+
+def _models(tmp_path: Path) -> Path:
+    """1 m dtm(가운데만 유효) · implicit 중심점(기울면) — 굳힌 격자와 비용면을 견줌."""
+    x = np.arange(0.0, 41.0)
+    y = np.arange(0.0, 31.0)
+    xx, yy = np.meshgrid(x, y)
+    valid = (xx > 3) & (xx < 37) & (yy > 3) & (yy < 27)
+    z = (0.05 * xx + 0.02 * yy).astype(np.float32)
+    np.savez(tmp_path / "dtm_classification.npz", x=x, y=y, z=z, valid_mask=valid)
+    rng = np.random.default_rng(7)
+    centers = rng.uniform(0, 40, size=(300, 2))
+    cz = 0.05 * centers[:, 0] + 0.02 * centers[:, 1] + 0.3 * np.sin(centers[:, 0] / 5)
+    np.savez(
+        tmp_path / "implicit_classification.npz",
+        centers_xy=centers,
+        center_z=cz,
+        smoothing=np.array([0.1]),
+    )
+    return tmp_path
+
+
+def test_비용면_implicit_굳힌_격자_마디_그대로(tmp_path) -> None:
+    models = _models(tmp_path)
+    x, y, z, valid, *_rest = solver._build_cost_surface(models, "classification", "implicit", False)
+    with np.load(implicit_grid_npz(models, "classification")) as grid:
+        gx, gy, gz, gvalid = grid["x"], grid["y"], grid["z"], grid["valid_mask"]
+    ix = [int(np.flatnonzero(gx == v)[0]) for v in x]
+    iy = [int(np.flatnonzero(gy == v)[0]) for v in y]
+    expected = np.asarray(gz, dtype=np.float64)[np.ix_(iy, ix)]
+    inside = np.asarray(gvalid)[np.ix_(iy, ix)]
+    assert inside.sum() > 50
+    assert np.abs(z[inside] - expected[inside]).max() == 0.0
+    # 서명이 굳힌 격자를 봄 — 서명 전에 지어 둠(첫 셈 뒤 열쇠가 안 바뀜)
+    names = [p.name for p in solver._source_npz_paths(models, "classification", "implicit", False)]
+    assert "implicit_classification_grid.npz" in names
+
+
+def test_implicit_파일_없으면_까닭_그대로(tmp_path) -> None:
+    models = _models(tmp_path)
+    (models / "implicit_classification.npz").unlink()
+    with pytest.raises(FileNotFoundError) as caught:
+        solver._build_cost_surface(models, "classification", "implicit", False)
+    assert "implicit_classification" in str(caught.value) and str(tmp_path) not in str(caught.value)
+
+
+def test_깨진_모델은_DTM_대체_없이_오류(tmp_path) -> None:
+    models = _models(tmp_path)
+    (models / "tin_classification.npz").write_bytes(b"broken")
+    with pytest.raises(ValueError, match="지표면 모델을 읽지 못함"):
+        solver._build_cost_surface(models, "classification", "tin", False)
+    source = (Path(solver.__file__)).read_text(encoding="utf-8")
+    assert "return dtm_z\n\n    return dtm_z" not in source
+    assert "except Exception:\n        return dtm_z" not in source