eomsangdonandClaude Opus 5.5
1fcdf5aa4c
- 도엽 3D 면 — 결과 옆 dtm_sheet_{방식}.key.json(등고선 이름:mtime:크기 · 노선 · CRS · 격자 · 여유 · 스무딩 · 화면 상자 · ALGO) 같으면 건너뜀
- 지표면 모델 — (방식 · 필터 · 파일) + 파일 지문 같으면 행 id 살려 고쳐 씀 · 이번 결과에 없는 옛 행만 지움
- 비용면 · 계류 메모리 기억(비용면 서명 · 계류 파일 지문 · 좌표계) · 추천 열쇠 = 서명(배열 통째 해시 없앰)
- 추천 저장본 종류(갈래 · 지형)당 최근 2벌 · 꺼내면 최근으로
- 횡단 상세 GET — 측점 파일 안 지움(valid_cross_names) · 본문 해시 ETag 304
- 시험 test_343_2_persist_keys 13건
Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_014utrjfoW831Vc5wJUd13kL
342 lines
14 KiB
Python
342 lines
14 KiB
Python
"""343-2 도엽 3D 면 열쇠 · 지표면 모델 id 유지 · 비용면 · 추천 노선 열쇠 · 횡단 상세 ETag.
|
|
|
|
같은 입력 두 번째 = 다시 안 셈 · 열쇠 바뀌면 다시 셈 · 지울 때.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import asyncio
|
|
import json
|
|
import os
|
|
import time
|
|
from pathlib import Path
|
|
|
|
import numpy as np
|
|
import pytest
|
|
from pyproj import Transformer
|
|
|
|
import B04_PreProcess.B04_PreProcess_Engine_SheetSurface as sheet
|
|
from B04_PreProcess.B04_PreProcess_Engine_SheetMethods import SHEET_METHOD_BUILDERS
|
|
|
|
EPSG = 5186
|
|
X0, Y0 = 200000.0, 500000.0
|
|
ROUTE = np.array([[X0 - 50.0, Y0], [X0 + 50.0, Y0]], dtype=np.float64)
|
|
|
|
|
|
@pytest.fixture()
|
|
def sheet_project(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> dict:
|
|
processed = tmp_path / "B04_PreProcess" / "processed"
|
|
models = tmp_path / "B04_PreProcess" / "models"
|
|
processed.mkdir(parents=True)
|
|
models.mkdir(parents=True)
|
|
to_wgs84 = Transformer.from_crs(f"EPSG:{EPSG}", "EPSG:4326", always_xy=True)
|
|
features = []
|
|
for half, elev in ((450.0, 100.0), (300.0, 110.0), (150.0, 120.0)):
|
|
ring = [
|
|
list(to_wgs84.transform(x, y))
|
|
for x, y in (
|
|
(X0 - half, Y0 - half),
|
|
(X0 + half, Y0 - half),
|
|
(X0 + half, Y0 + half),
|
|
(X0 - half, Y0 + half),
|
|
(X0 - half, Y0 - half),
|
|
)
|
|
]
|
|
features.append(
|
|
{
|
|
"type": "Feature",
|
|
"properties": {"등고수치": elev},
|
|
"geometry": {"type": "LineString", "coordinates": ring},
|
|
}
|
|
)
|
|
(processed / "도엽_등고선.geojson").write_text(
|
|
json.dumps({"type": "FeatureCollection", "features": features}, ensure_ascii=False),
|
|
encoding="utf-8",
|
|
)
|
|
calls: list[str] = []
|
|
real = SHEET_METHOD_BUILDERS["tin_sheet"]
|
|
|
|
def counting(*args, **kwargs): # type: ignore[no-untyped-def]
|
|
calls.append("tin_sheet")
|
|
return real(*args, **kwargs)
|
|
|
|
monkeypatch.setitem(SHEET_METHOD_BUILDERS, "tin_sheet", counting)
|
|
return {"root": tmp_path, "processed": processed, "models": models, "calls": calls}
|
|
|
|
|
|
def _build(project: dict, route: np.ndarray = ROUTE) -> list[dict]:
|
|
return sheet.build_sheet_surface_model(
|
|
project["root"], project["processed"], project["models"], route, EPSG, ["tin_sheet"]
|
|
)
|
|
|
|
|
|
def test_도엽_3D_면_같은_입력_두번째는_안_셈(sheet_project: dict) -> None:
|
|
first = _build(sheet_project)
|
|
assert sheet_project["calls"] == ["tin_sheet"]
|
|
assert (sheet_project["models"] / "dtm_sheet_tin_sheet.key.json").is_file()
|
|
started = time.perf_counter()
|
|
second = _build(sheet_project)
|
|
assert time.perf_counter() - started < 0.5
|
|
assert sheet_project["calls"] == ["tin_sheet"]
|
|
assert second == first
|
|
assert first[0]["generation_params"]["file_stamp"]
|
|
|
|
|
|
@pytest.mark.parametrize("change", ["route", "contour", "algo", "key_gone", "file_gone"])
|
|
def test_도엽_3D_면_열쇠가_바뀌면_다시_셈(
|
|
sheet_project: dict, monkeypatch: pytest.MonkeyPatch, change: str
|
|
) -> None:
|
|
_build(sheet_project)
|
|
route = ROUTE
|
|
if change == "route":
|
|
route = ROUTE + np.array([5.0, 0.0])
|
|
elif change == "contour":
|
|
path = sheet_project["processed"] / "도엽_등고선.geojson"
|
|
stat = path.stat()
|
|
os.utime(path, ns=(stat.st_atime_ns, stat.st_mtime_ns + 5_000_000_000))
|
|
elif change == "algo":
|
|
monkeypatch.setattr(sheet, "SHEET_SURFACE_ALGO", sheet.SHEET_SURFACE_ALGO + 1)
|
|
elif change == "key_gone":
|
|
(sheet_project["models"] / "dtm_sheet_tin_sheet.key.json").unlink()
|
|
else:
|
|
(sheet_project["models"] / "dtm_sheet_tin_sheet_preview.glb").unlink()
|
|
_build(sheet_project, route)
|
|
assert sheet_project["calls"] == ["tin_sheet", "tin_sheet"]
|
|
|
|
|
|
# ── 지표면 모델 id 유지(R2) ─────────────────────────────────────────────────────
|
|
|
|
|
|
class _FakeCursor:
|
|
def __init__(self, db: "_FakeDb") -> None:
|
|
self.db = db
|
|
self.rowcount = 0
|
|
self.lastrowid = 0
|
|
self._rows: list[tuple] = []
|
|
|
|
async def __aenter__(self) -> "_FakeCursor":
|
|
return self
|
|
|
|
async def __aexit__(self, *_: object) -> None:
|
|
return None
|
|
|
|
async def execute(self, sql: str, params: tuple = ()) -> None:
|
|
text = " ".join(sql.split())
|
|
self.db.log.append(text)
|
|
if text.startswith("SELECT id, model_type"):
|
|
self._rows = [
|
|
(row["id"], row["model_type"], row["path"], json.dumps(row["params"]))
|
|
for row in self.db.models.values()
|
|
]
|
|
elif text.startswith("INSERT INTO surface_models"):
|
|
self.db.next_id += 1
|
|
self.lastrowid = self.db.next_id
|
|
self.db.models[self.lastrowid] = {
|
|
"id": self.lastrowid,
|
|
"model_type": params[1],
|
|
"path": params[7],
|
|
"params": json.loads(params[8]),
|
|
}
|
|
elif text.startswith("UPDATE surface_models"):
|
|
self.db.models[params[-1]]["params"] = json.loads(params[4])
|
|
elif text.startswith("DELETE FROM surface_models WHERE id IN"):
|
|
for model_id in params:
|
|
self.db.models.pop(model_id, None)
|
|
elif text.startswith("DELETE FROM surface_models WHERE id ="):
|
|
self.db.models.pop(params[0], None)
|
|
elif text.startswith("DELETE FROM surface_models WHERE project_id"):
|
|
for model_id in [k for k, v in self.db.models.items() if v["path"] == params[1]]:
|
|
self.db.models.pop(model_id)
|
|
elif text.startswith("INSERT INTO terrain_layers"):
|
|
self.db.next_id += 1
|
|
self.lastrowid = self.db.next_id
|
|
|
|
async def fetchall(self) -> list[tuple]:
|
|
return self._rows
|
|
|
|
async def fetchone(self) -> tuple | None:
|
|
return None
|
|
|
|
|
|
class _FakeDb:
|
|
def __init__(self) -> None:
|
|
self.log: list[str] = []
|
|
self.models: dict[int, dict] = {}
|
|
self.next_id = 100
|
|
|
|
def cursor(self, *_: object) -> _FakeCursor:
|
|
return _FakeCursor(self)
|
|
|
|
|
|
def _model(path: str, stamp: str) -> dict:
|
|
return {
|
|
"model_type": "dtm",
|
|
"model_file_path": path,
|
|
"resolution_m": 1.0,
|
|
"generation_params": {"source_filter": "csf", "file_stamp": stamp},
|
|
"layers": [
|
|
{
|
|
"layer_name": "dtm_csf_preview",
|
|
"geometry_type": "MESH",
|
|
"file_path": "B04_PreProcess/models/dtm_csf_preview.glb",
|
|
"file_format": "glb",
|
|
}
|
|
],
|
|
}
|
|
|
|
|
|
def _save(db: _FakeDb, models: list[dict]) -> list[int]:
|
|
from B04_PreProcess import B04_PreProcess_Repository as repo
|
|
|
|
async def fake_input(*_a): # type: ignore[no-untyped-def]
|
|
return {"crs_epsg": 5186}
|
|
|
|
async def fake_cloud(*_a, **_k): # type: ignore[no-untyped-def]
|
|
return 9
|
|
|
|
repo_input, repo_cloud = repo.get_input_file, repo.create_processed_point_cloud
|
|
repo.get_input_file, repo.create_processed_point_cloud = fake_input, fake_cloud
|
|
try:
|
|
return asyncio.run(
|
|
repo.save_surface_analysis_to_db(
|
|
db, # type: ignore[arg-type]
|
|
project_id="00000000-0000-0000-0000-000000000001", # type: ignore[arg-type]
|
|
input_file_id=1,
|
|
analysis_result={
|
|
"processed": {
|
|
"processed_file_path": "B04_PreProcess/processed/structured.npz",
|
|
"converted_file_path": None,
|
|
"point_count": 1,
|
|
"bounds": {},
|
|
"statistics": {},
|
|
},
|
|
"models": models,
|
|
},
|
|
source_filters=["csf"],
|
|
)
|
|
)
|
|
finally:
|
|
repo.get_input_file, repo.create_processed_point_cloud = repo_input, repo_cloud
|
|
|
|
|
|
def test_파일_지문이_같으면_모델_id_유지() -> None:
|
|
db = _FakeDb()
|
|
path = "B04_PreProcess/models/dtm_csf.npz"
|
|
first = _save(db, [_model(path, "1:2"), _model("B04_PreProcess/models/tin_csf.npz", "3:4")])
|
|
second = _save(db, [_model(path, "1:2")])
|
|
assert second == [first[0]]
|
|
# 이번 결과에 없는 옛 행은 지움 · 같은 파일 행은 하나
|
|
assert list(db.models) == [first[0]]
|
|
third = _save(db, [_model(path, "9:9")])
|
|
assert third != second and list(db.models) == third
|
|
|
|
|
|
def test_도엽_한_방식_요청도_id_유지() -> None:
|
|
from B04_PreProcess import B04_PreProcess_Repository as repo
|
|
|
|
db = _FakeDb()
|
|
path = "B04_PreProcess/models/dtm_sheet_tin_sheet.npz"
|
|
entry = {**_model(path, "5:6"), "generation_params": {"source_filter": "sheet_tin_sheet"}}
|
|
entry["generation_params"]["file_stamp"] = "5:6"
|
|
first = asyncio.run(repo.save_sheet_surface_models(db, "p", [entry])) # type: ignore[arg-type]
|
|
second = asyncio.run(repo.save_sheet_surface_models(db, "p", [entry])) # type: ignore[arg-type]
|
|
assert first == second and len(db.models) == 1
|
|
|
|
|
|
# ── 비용면 · 계류 메모리 기억(R17) · 추천 노선 열쇠(R18) ──────────────────────────
|
|
|
|
|
|
def test_비용면_계류_메모리_기억(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None:
|
|
import B04_PreProcess.B04_PreProcess_Router_Watershed as watershed
|
|
import B05_Profile.B05_Profile_Engine_RouteInitial as engine
|
|
import B05_Profile.B05_Profile_Engine_Solver as solver
|
|
import common_util.common_util_crs as crs
|
|
|
|
loads: list[int] = []
|
|
signature = {"value": "sig-1"}
|
|
grid = np.arange(4.0)
|
|
|
|
def fake_load(*_a): # type: ignore[no-untyped-def]
|
|
loads.append(1)
|
|
return grid, grid, np.zeros((4, 4)), np.ones((4, 4), bool), None, None, 1.0
|
|
|
|
monkeypatch.setattr(engine, "_TERRAINS", {})
|
|
monkeypatch.setattr(solver, "_load_or_build_cost_surface", fake_load)
|
|
monkeypatch.setattr(solver, "_cost_surface_signature", lambda *_a: signature["value"])
|
|
monkeypatch.setattr(watershed, "_load_features", lambda *_a: [])
|
|
monkeypatch.setattr(crs, "resolve_project_crs", lambda _root: "EPSG:5186")
|
|
selection = {"source_filter": "csf", "method": "dtm", "smooth": False}
|
|
terrain, _ = engine.load_terrain_and_streams(tmp_path, selection)
|
|
again, _ = engine.load_terrain_and_streams(tmp_path, selection)
|
|
assert loads == [1] and again is terrain and terrain.signature
|
|
signature["value"] = "sig-2"
|
|
changed, _ = engine.load_terrain_and_streams(tmp_path, selection)
|
|
assert loads == [1, 1] and changed.signature != terrain.signature
|
|
# 추천 열쇠 = 서명 — 배열 통째 해시 안 함 · 서명이 다르면 다른 열쇠
|
|
key_a = engine._pool_key([(0.0, 0.0), (3.0, 3.0)], terrain, [], 0.14)
|
|
key_b = engine._pool_key([(0.0, 0.0), (3.0, 3.0)], changed, [], 0.14)
|
|
assert key_a != key_b
|
|
assert key_a == engine._pool_key([(0.0, 0.0), (3.0, 3.0)], terrain, [[(9.0, 9.0)]], 0.14)
|
|
|
|
|
|
def test_추천_저장본_종류당_최근_두_벌(tmp_path: Path) -> None:
|
|
from B05_Profile.B05_Profile_Engine_RouteInitial_Store import (
|
|
KEEP_PER_KIND,
|
|
read_stored,
|
|
store_dir,
|
|
write_stored,
|
|
)
|
|
|
|
assert KEEP_PER_KIND == 2
|
|
request = {"mode": "earthwork", "terrain": None}
|
|
write_stored(tmp_path, "a", "p", request, {"n": 1}, True)
|
|
time.sleep(0.02)
|
|
write_stored(tmp_path, "b", "p", request, {"n": 2}, False)
|
|
write_stored(tmp_path, "g", "p", {"mode": "gentle", "terrain": None}, {}, True)
|
|
time.sleep(0.02)
|
|
assert read_stored(tmp_path, "a") is not None # 꺼내면 최근으로
|
|
time.sleep(0.02)
|
|
write_stored(tmp_path, "c", "p", request, {"n": 3}, False)
|
|
names = sorted(path.stem for path in store_dir(tmp_path).glob("*.json"))
|
|
assert names == ["a", "c", "g"]
|
|
|
|
|
|
# ── 횡단 상세 — GET 은 안 지움 · ETag(R26) ───────────────────────────────────────
|
|
|
|
|
|
def test_횡단_상세_조회는_파일을_지우지_않음(tmp_path: Path) -> None:
|
|
from B05_Profile.B05_Profile_Engine_Sections import prune_stale_cross_files, valid_cross_names
|
|
|
|
for name in ("cross_00000m.json", "cross_00020m.json", "cross_00040m.json"):
|
|
(tmp_path / name).write_text("{}", encoding="utf-8")
|
|
stations = [{"chainage_m": 0.0}, {"chainage_m": 20.0}]
|
|
names = valid_cross_names(tmp_path, stations)
|
|
assert len(names) == 2 and (tmp_path / "cross_00040m.json").is_file()
|
|
assert prune_stale_cross_files(tmp_path, stations) == names
|
|
assert not (tmp_path / "cross_00040m.json").exists()
|
|
|
|
|
|
def test_횡단_상세_ETag_같으면_304() -> None:
|
|
from starlette.requests import Request
|
|
|
|
from B06_Section.B06_Section_Router import _etag_json_response
|
|
|
|
def request(etag: str | None) -> Request:
|
|
headers = [] if etag is None else [(b"if-none-match", etag.encode())]
|
|
return Request({"type": "http", "headers": headers})
|
|
|
|
body = b'{"cross_sections": []}'
|
|
first = _etag_json_response(request(None), body)
|
|
assert first.status_code == 200 and first.body == body
|
|
etag = first.headers["etag"]
|
|
assert _etag_json_response(request(etag), body).status_code == 304
|
|
assert _etag_json_response(request(etag), body + b" ").status_code == 200
|
|
|
|
|
|
def test_횡단_상세_GET_소스에_지우기_없음() -> None:
|
|
source = Path("B06_Section/B06_Section_Router.py").read_text(encoding="utf-8")
|
|
start = source.index("def _read_section_detail(")
|
|
end = source.index("\ndef ", start + 10)
|
|
assert "prune_stale_cross_files" not in source[start:end]
|
|
assert "unlink" not in source[start:end]
|