262 lines
11 KiB
Python
262 lines
11 KiB
Python
"""PLAN 637 — LAS 나중 올리기 공용(`common_util_lidar_upload`) · B05 창구(요구 546).
|
|
|
|
637-1 = 예정노선 길을 공용으로(동작 그대로 · `test_624_4_plan_lidar` 함께) —
|
|
갈음 · 짓기 · DB 넣기/고치기만.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
from pathlib import Path
|
|
from typing import Any
|
|
from uuid import UUID
|
|
|
|
import numpy as np
|
|
import pytest
|
|
|
|
from common_util import common_util_lidar_upload as lidar_upload
|
|
|
|
PID = UUID("00000000-0000-0000-0000-000000000637")
|
|
EPSG = 5186
|
|
|
|
|
|
def write_las(path: Path, x0: float, y0: float, size: float = 60.0, n: int = 6000) -> None:
|
|
"""지면점 LAS(비탈 0.3 · 좌표계 없음)."""
|
|
import laspy
|
|
|
|
rng = np.random.default_rng(7)
|
|
x, y = rng.uniform(x0, x0 + size, n), rng.uniform(y0, y0 + size, n)
|
|
header = laspy.LasHeader(point_format=3, version="1.2")
|
|
header.offsets = [round(x0), round(y0), 0]
|
|
header.scales = [0.01, 0.01, 0.01]
|
|
data = laspy.LasData(header)
|
|
data.x, data.y, data.z = x, y, 500 + 0.3 * (x - x0)
|
|
data.classification = np.full(len(x), 2, np.uint8)
|
|
path.parent.mkdir(parents=True, exist_ok=True)
|
|
data.write(path)
|
|
|
|
|
|
def test_갈음_짓기_지도_안_받음(tmp_path: Path, monkeypatch) -> None: # type: ignore[no-untyped-def]
|
|
import B04_PreProcess.B04_PreProcess_Engine as engine
|
|
|
|
monkeypatch.setattr(engine, "download_geodata", lambda *a, **k: pytest.fail("지도 받기"))
|
|
upload, las_dir = tmp_path / "up", tmp_path / "las"
|
|
write_las(las_dir / "old.las", 200000.0, 500000.0)
|
|
write_las(upload / "new.las", 200000.0, 500000.0)
|
|
got = lidar_upload.stage_las(upload, las_dir, EPSG, lambda *_: None)
|
|
assert [p.name for p in got] == ["new.las"] and not upload.exists() # 옛 LAS 갈음
|
|
again = lidar_upload.stage_las(upload, las_dir, EPSG, lambda *_: None)
|
|
assert [p.name for p in again] == ["new.las"] # 올림 없음 = 있는 것 그대로
|
|
seen: list[int] = []
|
|
result = lidar_upload.build_surface(tmp_path, got, lambda p, _m: seen.append(p))
|
|
picked = result["source_filters"][0]
|
|
assert (tmp_path / "B04_PreProcess" / "models" / f"dtm_{picked}.npz").is_file()
|
|
assert seen and min(seen) >= 10 and max(seen) <= 95
|
|
with pytest.raises(FileNotFoundError):
|
|
lidar_upload.stage_las(upload, tmp_path / "빈", EPSG, lambda *_: None)
|
|
|
|
|
|
def test_진행_기록_끊김(tmp_path: Path) -> None:
|
|
path = tmp_path / "job.json"
|
|
assert lidar_upload.saved_job(path) == lidar_upload.IDLE
|
|
lidar_upload.note_job(path, {"running": True, "percent": 42})
|
|
assert lidar_upload.saved_job(path) == {**lidar_upload.IDLE, "percent": 42, "interrupted": True}
|
|
|
|
|
|
class _Cursor:
|
|
def __init__(self, rows: list[tuple]) -> None:
|
|
self.rows = rows
|
|
|
|
async def __aenter__(self) -> "_Cursor":
|
|
return self
|
|
|
|
async def __aexit__(self, *_: Any) -> None:
|
|
return None
|
|
|
|
async def execute(self, *_: Any) -> None:
|
|
return None
|
|
|
|
async def fetchall(self) -> list[tuple]:
|
|
return self.rows
|
|
|
|
|
|
class _Connection:
|
|
def __init__(self, rows: list[tuple]) -> None:
|
|
self.rows, self.done = rows, []
|
|
|
|
async def begin(self) -> None:
|
|
return None
|
|
|
|
async def commit(self) -> None:
|
|
self.done.append("commit")
|
|
|
|
async def rollback(self) -> None:
|
|
self.done.append("rollback")
|
|
|
|
def cursor(self) -> _Cursor:
|
|
return _Cursor(self.rows)
|
|
|
|
|
|
class _Pool:
|
|
def __init__(self, connection: _Connection) -> None:
|
|
self.connection = connection
|
|
|
|
def acquire(self) -> "_Pool":
|
|
return self
|
|
|
|
async def __aenter__(self) -> _Connection:
|
|
return self.connection
|
|
|
|
async def __aexit__(self, *_: Any) -> None:
|
|
return None
|
|
|
|
|
|
def _save(monkeypatch, keep: bool) -> tuple[list[dict], list[int], _Connection]: # type: ignore[no-untyped-def]
|
|
"""옛 행 둘(확정 dtm_a · 수치지형도) + 새 모델 dtm_a(지문 다름) → 부른 값."""
|
|
import asyncio
|
|
|
|
import B04_PreProcess.B04_PreProcess_Repository as repo
|
|
import config.config_db as config_db
|
|
|
|
path_a = "B04_PreProcess/models/dtm_a.npz"
|
|
existing = {("dtm", "a", path_a): (11, "old"), ("dtm", "sheet_laplace", "s.npz"): (12, "s")}
|
|
calls: list[dict] = []
|
|
confirmed: list[int] = []
|
|
|
|
async def existing_models(*_: Any) -> dict:
|
|
return dict(existing)
|
|
|
|
async def upsert(_c: Any, have: dict, **kw: Any) -> int:
|
|
key = ("dtm", "a", path_a)
|
|
calls.append({**kw, "have": have.get(key)})
|
|
return 11 if have.get(key, (0, ""))[1] == "new" else 99
|
|
|
|
async def confirm(_c: Any, _p: Any, model_id: int) -> None:
|
|
confirmed.append(model_id)
|
|
|
|
connection = _Connection([(11, 1320, 7, "CONFIRMED"), (12, None, None, "COMPLETE")])
|
|
monkeypatch.setattr(repo, "_existing_models", existing_models)
|
|
monkeypatch.setattr(repo, "_upsert_surface_model", upsert)
|
|
monkeypatch.setattr(repo, "confirm_surface_model", confirm)
|
|
monkeypatch.setattr(config_db, "get_db_pool", lambda: _Pool(connection))
|
|
model = {
|
|
"model_type": "dtm",
|
|
"model_file_path": path_a,
|
|
"generation_params": {"source_filter": "a", "file_stamp": "new"},
|
|
}
|
|
asyncio.run(lidar_upload.save_models(PID, [model], EPSG, keep=keep))
|
|
return calls, confirmed, connection
|
|
|
|
|
|
def test_DB_넣기_고치기만_갈음은_id_확정_살림(monkeypatch) -> None: # type: ignore[no-untyped-def]
|
|
calls, confirmed, connection = _save(monkeypatch, keep=False) # 예정노선 = 옛 그대로
|
|
assert calls[0]["have"] == (11, "old") and calls[0]["source_file_id"] is None
|
|
assert confirmed == [] and connection.done == ["commit"]
|
|
calls, confirmed, _ = _save(monkeypatch, keep=True) # B 갈음 = 지문 맞춤 · 연결 · 확정 살림
|
|
assert calls[0]["have"] == (11, "new")
|
|
assert (calls[0]["source_file_id"], calls[0]["processed_cloud_id"]) == (1320, 7)
|
|
assert confirmed == [11]
|
|
|
|
|
|
# ── 637-2 B05 창구 ──────────────────────────────────────────────────────────
|
|
|
|
|
|
def test_라이다_확정값_고르기() -> None:
|
|
from B05_Profile import B05_Profile_Router_Lidar as b05
|
|
|
|
las = {"source_filter": "classification", "method": "dtm", "smooth": True}
|
|
assert b05.las_target(las) == las # 설계 = 라이다 → 그 필터로 갈음
|
|
sheet = {"source_filter": "sheet_laplace", "method": "dtm", "smooth": True}
|
|
assert b05.las_target(sheet) is None # 수치지형도만 = 새로 덧붙임(자동 필터)
|
|
assert b05.las_target({**sheet, "las_selection": las}) == las # 적어 둔 것
|
|
|
|
|
|
def test_노선_2m_나눔() -> None:
|
|
from B05_Profile import B05_Profile_Router_Lidar as b05
|
|
|
|
pts = b05.line_points([[0, 0], [10, 0], [10, 3]])
|
|
assert pts[0].tolist() == [0, 0] and pts[-1].tolist() == [10, 3]
|
|
assert len(pts) == 1 + 5 + 2 and np.hypot(*np.diff(pts, axis=0).T).max() <= 2.0 + 1e-9
|
|
|
|
|
|
def _route_build(tmp_path: Path, monkeypatch, line, target=None) -> dict: # type: ignore[no-untyped-def]
|
|
import B04_PreProcess.B04_PreProcess_Engine as engine
|
|
from B05_Profile import B05_Profile_Router_Lidar as b05
|
|
from B05_Profile import B05_Profile_Router_Replan as replan
|
|
|
|
monkeypatch.setattr(engine, "download_geodata", lambda *a, **k: pytest.fail("지도 받기"))
|
|
monkeypatch.setattr(replan, "_planned_line", lambda _root: line)
|
|
write_las(tmp_path / b05.UPLOAD / "t.las", 200000.0, 500000.0)
|
|
return b05.build_route_lidar(tmp_path, EPSG, target, lambda *_: None)
|
|
|
|
|
|
def test_B_뿌리_짓기_수치지형도_그대로_덮임(tmp_path: Path, monkeypatch) -> None: # type: ignore[no-untyped-def]
|
|
from B05_Profile import B05_Profile_Router_Lidar as b05
|
|
|
|
sheet = tmp_path / b05.MODELS / "dtm_sheet_laplace.npz"
|
|
sheet.parent.mkdir(parents=True)
|
|
sheet.write_bytes(b"sheet")
|
|
got = _route_build(tmp_path, monkeypatch, [[200010, 500010], [200050, 500050]])
|
|
meta = got["meta"]
|
|
assert sheet.read_bytes() == b"sheet" # 수치지형도 모델 파일 그대로
|
|
assert (tmp_path / b05.MODELS / f"dtm_{meta['selection']['source_filter']}.npz").is_file()
|
|
assert meta["covered"] is True and meta["covered_ratio"] == 1.0
|
|
assert meta["files"] == ["t.las"] and not (tmp_path / b05.UPLOAD).exists()
|
|
assert all(m["model_file_path"].startswith("B04_PreProcess/models/") for m in got["models"])
|
|
assert b05.read_meta(tmp_path)["selection"] == meta["selection"]
|
|
# 노선 반이 라이다 밖 = 덮임 거짓
|
|
again = _route_build(tmp_path, monkeypatch, [[200010, 500030], [200110, 500030]])["meta"]
|
|
assert again["covered"] is False and 0.3 < again["covered_ratio"] < 0.7
|
|
# 있던 라이다 확정값 = 그 필터로 짓기
|
|
target = {"source_filter": "classification", "method": "dtm", "smooth": True}
|
|
kept = _route_build(tmp_path, monkeypatch, [[200010, 500010], [200020, 500020]], target)
|
|
assert kept["meta"]["selection"] == target
|
|
assert {m["source_filter"] for m in kept["models"]} == {"classification"}
|
|
|
|
|
|
class _DictCursor(_Cursor):
|
|
def __init__(self, params: dict) -> None:
|
|
super().__init__([])
|
|
self.params, self.sql = params, []
|
|
|
|
async def execute(self, sql: str, args: tuple = ()) -> None:
|
|
self.sql.append((sql, args))
|
|
|
|
async def fetchone(self) -> dict:
|
|
import json
|
|
|
|
return {"params": json.dumps(self.params)}
|
|
|
|
|
|
def test_las_selection_적기_칩_열림(tmp_path: Path) -> None:
|
|
import asyncio
|
|
import json
|
|
|
|
from common_util import common_util_surface_confirmation as confirm
|
|
|
|
params = {"source_filter": "sheet_laplace", "method": "dtm", "smooth": True}
|
|
assert confirm.analysis_terrain_sources(params, tmp_path) == ["sheet"] # 칩 라이다 막힘
|
|
cursor = _DictCursor(params)
|
|
connection = type("C", (), {"cursor": lambda self, *_: cursor})()
|
|
pick = {"source_filter": "csf", "method": "dtm", "smooth": True}
|
|
asyncio.run(confirm.stash_las_selection(connection, str(PID), pick))
|
|
written = json.loads(cursor.sql[-1][1][0])
|
|
assert written["las_selection"] == pick
|
|
assert written["source_filter"] == "sheet_laplace" # 설계 지형 그대로
|
|
assert confirm.analysis_terrain_sources(written, tmp_path) == ["las", "sheet"]
|
|
assert confirm.analysis_terrain_params(written, "las")["source_filter"] == "csf"
|
|
|
|
|
|
def test_지형_바꿈_force_같은_지형도_다시_셈(monkeypatch) -> None: # type: ignore[no-untyped-def]
|
|
import test_626_1_recommend_preview as base
|
|
|
|
from B05_Profile import B05_Profile_Router_TerrainSwitch as switch
|
|
|
|
request = switch.DesignTerrainRequest
|
|
assert request(terrain="las").force is False
|
|
monkeypatch.setattr(
|
|
switch, "DesignTerrainRequest", lambda terrain: request(terrain=terrain, force=True)
|
|
)
|
|
result, calls, line = base._run(monkeypatch, "las", "las")
|
|
assert result["changed"] is True and calls["args"][4] == "las"
|
|
assert calls["args"][5] == (line, line) # 노선 그대로
|