fix(763-1): 추천노선 = 확정 지표면에서만 — 확정 전 409 · 비용면 = 굳힌 격자 마디 · 조용한 DTM 대체 걷음
- `confirmed_surface_params()` — stage 1 COMPLETE · params 에 source_filter 있을 때만 · 아니면 None - compute_initial = 확정 전 409 「지표면 확정 전 — B04 [확정] 뒤 추천이 생김」 · 미리 셈 지형 = [](건너뜀 로그) - Solver implicit 갈래 = `implicit_grid_npz` 마디 그대로(RBF 다시 안 셈 · 짓기 16.2 s → 0.02 s · 유효 칸 최대차 0.00003 m) - 모델 파일 없음 · 깨짐 = 까닭 그대로 FileNotFoundError · ValueError(409) · 서명에 굳힌 격자(서명 전에 지음) - 시험 224_3 · 351_1 · 509_5 의 확정 갈아 끼우기를 새 함수로 옮김(라우터가 새 함수를 부름) Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
This commit is contained in:
@@ -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))
|
||||
|
||||
@@ -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"]
|
||||
|
||||
|
||||
@@ -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
|
||||
Reference in New Issue
Block a user