diff --git a/B05_Profile/B05_Profile_Engine_Solver.py b/B05_Profile/B05_Profile_Engine_Solver.py index 3157d0fc8..416265f54 100644 --- a/B05_Profile/B05_Profile_Engine_Solver.py +++ b/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: diff --git a/B05_Profile/B05_Profile_Router_RouteInitial.py b/B05_Profile/B05_Profile_Router_RouteInitial.py index 136a0a038..d4992f325 100644 --- a/B05_Profile/B05_Profile_Router_RouteInitial.py +++ b/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 ) diff --git a/common_util/common_util_surface_confirmation.py b/common_util/common_util_surface_confirmation.py index e48d1649b..df8a5a170 100644 --- a/common_util/common_util_surface_confirmation.py +++ b/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_") diff --git a/resources/tester/test_224_3_recommend_store.py b/resources/tester/test_224_3_recommend_store.py index 0567f6656..588e15b9a 100644 --- a/resources/tester/test_224_3_recommend_store.py +++ b/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)) diff --git a/resources/tester/test_509_5_recommend_terrain.py b/resources/tester/test_509_5_recommend_terrain.py index b2ab71452..c7d022618 100644 --- a/resources/tester/test_509_5_recommend_terrain.py +++ b/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"] diff --git a/resources/tester/test_763_1_confirmed_terrain.py b/resources/tester/test_763_1_confirmed_terrain.py new file mode 100644 index 000000000..92475801b --- /dev/null +++ b/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