eomsangdonandClaude Opus 5.5
b3dcb73784
- 공용 A00_Common/map/map_lidar_upload — 파일 고름 · POST multipart · 1 초 묻기 · 서클 · 끝난 뒤 일(주소 · 진행 묻기만 인자) - 예정노선 UI_LidarUpload = 부품 부르기(화면 · 동작 그대로) · 안 쓰는 uploadPlanLidar 걷음 - B05 UI_RouteEdit_Terrain — 칩 · 단추 · 서클 한 칸 · 끝나면 확인창 「설계 지형을 라이다로 바꿈?」(이미 라이다 = force 다시 셈) · covered 거짓 = 바꿈 막고 알림 · 들어올 때 도는 일 이어 봄 - common_util_lidar_upload.drop_stale_filters — 갈음 뒤 안 지은 라이다 필터(옛 LAS) 모델 · 등고선 · 마스크 · 지면점 · manifest 줄 지움 · save_models drop = 그 행만(확정 행 남김) - 시험 test_637_lidar(필터 지움 · 행 지움 · B05 화면) · test_624_4b(올리기 = 공용 부품으로 옮김) Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
338 lines
15 KiB
Python
338 lines
15 KiB
Python
"""PLAN 637 — LAS 나중 올리기 공용(`common_util_lidar_upload`) · B05 창구(요구 546).
|
|
|
|
637-1 = 예정노선 길을 공용으로(동작 그대로 · `test_624_4_plan_lidar` 함께) —
|
|
갈음 · 짓기 · DB 넣기/고치기만.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import json
|
|
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], log: list | None = None) -> None:
|
|
self.rows, self.log = rows, log if log is not None else []
|
|
|
|
async def __aenter__(self) -> "_Cursor":
|
|
return self
|
|
|
|
async def __aexit__(self, *_: Any) -> None:
|
|
return None
|
|
|
|
async def execute(self, sql: str, args: tuple = ()) -> None:
|
|
self.log.append((sql, args))
|
|
|
|
async def fetchall(self) -> list[tuple]:
|
|
return self.rows
|
|
|
|
|
|
class _Connection:
|
|
def __init__(self, rows: list[tuple]) -> None:
|
|
self.rows, self.done, self.sql = 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, self.sql)
|
|
|
|
|
|
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, drop: list[str] | None = None
|
|
) -> 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, drop=drop))
|
|
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]
|
|
assert not any("DELETE" in sql for sql, _ in connection.sql) # drop 없음 = 안 지움
|
|
|
|
|
|
def test_DB_지운_필터_행만_지움_확정_남김(monkeypatch) -> None: # type: ignore[no-untyped-def]
|
|
gone = ["B04_PreProcess/models/dtm_csf.npz", "B04_PreProcess/models/dtm_csf_smooth.npz"]
|
|
_, _, connection = _save(monkeypatch, keep=True, drop=gone)
|
|
sql, args = [row for row in connection.sql if "DELETE" in row[0]][0]
|
|
assert "status <> 'CONFIRMED'" in sql and "project_id = %s" in sql
|
|
assert args == (str(PID), *gone) and connection.done == ["commit"]
|
|
|
|
|
|
def test_갈음_안_지은_필터_파일_지움(tmp_path: Path) -> None:
|
|
models = tmp_path / "B04_PreProcess" / "models"
|
|
processed = tmp_path / "B04_PreProcess" / "processed"
|
|
models.mkdir(parents=True)
|
|
processed.mkdir(parents=True)
|
|
keep = [
|
|
"dtm_classification.npz",
|
|
"dtm_classification_smooth.npz",
|
|
"contour_classification_dtm_1.0m.json",
|
|
"dtm_sheet_laplace.npz",
|
|
"dtm_sheet_laplace_smooth_preview.glb",
|
|
"dtm_sheet_laplace.key.json",
|
|
]
|
|
stale = [
|
|
"dtm_csf.npz",
|
|
"dtm_csf_preview.glb",
|
|
"dtm_csf_smooth.npz",
|
|
"dtm_csf_smooth_grid.bin",
|
|
"tin_csf.npz",
|
|
"contour_csf_dtm_smooth_1.0m.json",
|
|
]
|
|
for name in keep + stale:
|
|
(models / name).write_bytes(b"x")
|
|
for name in ["mask_csf.npy", "ground_points_csf.npz", "preview_points_csf.bin"]:
|
|
(processed / name).write_bytes(b"x")
|
|
(processed / "mask_classification.npy").write_bytes(b"x")
|
|
manifest = {"source_filters": {"classification": {}, "csf": {}}, "x": 1}
|
|
(models / "manifest.json").write_text(json.dumps(manifest), encoding="utf-8")
|
|
gone = lidar_upload.drop_stale_filters(tmp_path, ["classification"])
|
|
assert gone == sorted(
|
|
f"B04_PreProcess/models/{n}" for n in ["dtm_csf.npz", "dtm_csf_smooth.npz", "tin_csf.npz"]
|
|
)
|
|
assert sorted(p.name for p in models.iterdir()) == sorted([*keep, "manifest.json"])
|
|
assert [p.name for p in processed.iterdir()] == ["mask_classification.npy"]
|
|
saved = json.loads((models / "manifest.json").read_text(encoding="utf-8"))
|
|
assert saved == {"source_filters": {"classification": {}}, "x": 1}
|
|
assert lidar_upload.drop_stale_filters(tmp_path, ["classification"]) == [] # 두 번 = 할 것 없음
|
|
|
|
|
|
# ── 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"}
|
|
# 갈음 = 안 지은 필터(앞서 자동으로 지은 csf)는 옛 LAS 것 → 지움 · 수치지형도 그대로(637-3)
|
|
assert "B04_PreProcess/models/dtm_csf.npz" in kept["dropped"]
|
|
assert not list((tmp_path / b05.MODELS).glob("*_csf*")) and sheet.read_bytes() == b"sheet"
|
|
|
|
|
|
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) # 노선 그대로
|
|
|
|
|
|
# ── 637-3 B05 화면 ──────────────────────────────────────────────────────────
|
|
|
|
|
|
def test_B05_칩_옆_올리기_확인창_force_덮임_막음() -> None:
|
|
root = Path(__file__).resolve().parents[2]
|
|
ui = (root / "B05_Profile" / "B05_Profile_UI_RouteEdit_Terrain.ts").read_text(encoding="utf-8")
|
|
assert "createLidarUpload<RouteLidar>({" in ui # 예정노선과 같은 공용 부품
|
|
assert "const lidarPath = `/projects/${projectId}/route/lidar`;" in ui
|
|
assert "root.append(chip.root, upload.button, circle.root);" in ui # 칩 옆 · 서클
|
|
assert "createProgressCircle({ size: 16, inline: true })" in ui
|
|
# 끝 = 칩 다시 · 못 덮음 = 막고 알림 · 이미 라이다 = force 다시 셈 · 확인창 거쳐서만
|
|
assert 'await change("las", design === "las");' in ui
|
|
assert "if (uncovered != null) return warnUncovered();" in ui
|
|
assert 'if (next === "las" && uncovered != null) {' in ui
|
|
assert "JSON.stringify({ terrain: next, force })" in ui
|
|
assert ui.index("await askConfirm({") < ui.index(
|
|
"`/projects/${projectId}/route/design-terrain`"
|
|
)
|
|
assert "if (info.running && !upload.busy()) void upload.follow();" in ui # 들어올 때 이어 봄
|
|
assert len(ui.splitlines()) < 700
|