eomsangdonandClaude Opus 5.5
cb5c26d424
- B05 LAS 올리기: 설계 라이다 B 는 시험 자리(lidar_try)에 따로 지어 노선 덮임 검사 → 다 덮을 때만 본 자리로 옮김 - 못 덮음 = 「노선을 ○ % 만 덮음 — 설계 지형이 라이다라 바꾸지 않음」 · 옛 LAS · 모델 · 메타 · DB 행 그대로 · 시험 자리 지움 - 공용 promote_build — 입력 LAS 갈음 · processed · models 덮어 옮김(수치지형도 · 지도 그대로 · 수정시각 살아 행 지문 그대로) - 수치지형도 설계 B(덧붙임) · 예정노선 길 그대로 - 시험: 반쪽 LAS 거절 · 파일 지문 · 메타 그대로 / 다 덮는 LAS 갈음 · 행 지문 = 옮긴 파일 Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
380 lines
17 KiB
Python
380 lines
17 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, guard=False, x0=200000.0) -> 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", x0, 500000.0)
|
|
return b05.build_route_lidar(tmp_path, EPSG, target, lambda *_: None, guard)
|
|
|
|
|
|
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"
|
|
|
|
|
|
def _stamps(root: Path) -> dict[str, str]:
|
|
"""B04_PreProcess 파일 지문(경로 → 크기 · 수정시각 · 내용 앞)."""
|
|
pre = root / "B04_PreProcess"
|
|
return {
|
|
p.relative_to(pre).as_posix(): f"{p.stat().st_size}:{p.stat().st_mtime_ns}"
|
|
for p in pre.rglob("*")
|
|
if p.is_file() and "lidar_job" not in p.name
|
|
}
|
|
|
|
|
|
def test_설계_라이다_B_못_덮는_LAS_거절_다_덮으면_갈음(tmp_path: Path, monkeypatch) -> None: # type: ignore[no-untyped-def]
|
|
"""637-5 — 설계 지형이 라이다인 B: 시험 자리에 지어 덮임 검사 · 못 덮음 = 뿌리 그대로."""
|
|
from B05_Profile import B05_Profile_Router_Lidar as b05
|
|
|
|
line = [[200010, 500010], [200050, 500050]]
|
|
target = {"source_filter": "classification", "method": "dtm", "smooth": True}
|
|
first = _route_build(tmp_path, monkeypatch, line, target)
|
|
assert first["meta"]["covered"] is True
|
|
before, meta = _stamps(tmp_path), b05.read_meta(tmp_path)
|
|
# 반쪽 LAS(노선 x 200010 ~ 200050 중 200030 부터만) = 거절 · 파일 · 메타 그대로 · 시험 자리 지움
|
|
with pytest.raises(
|
|
ValueError, match=r"노선을 \d+ % 만 덮음 — 설계 지형이 라이다라 바꾸지 않음"
|
|
):
|
|
_route_build(tmp_path, monkeypatch, line, target, guard=True, x0=200030.0)
|
|
assert _stamps(tmp_path) == before and b05.read_meta(tmp_path) == meta
|
|
assert not (tmp_path / b05.TRY).exists() and not (tmp_path / b05.UPLOAD).exists()
|
|
# 다 덮는 LAS = 갈음 — 본 자리 모델 · 입력 LAS 새것 · 시험 자리 지움
|
|
got = _route_build(tmp_path, monkeypatch, line, target, guard=True, x0=199995.0)
|
|
assert got["meta"]["covered"] is True and got["meta"]["selection"] == target
|
|
assert not (tmp_path / b05.TRY).exists()
|
|
after = _stamps(tmp_path)
|
|
assert after["models/dtm_classification.npz"] != before["models/dtm_classification.npz"]
|
|
assert after["lidar_las/t.las"] != before["lidar_las/t.las"]
|
|
for model in got["models"]: # 행 지문 = 옮긴 파일 지문
|
|
from common_util.common_util_derived_store import file_stamp
|
|
|
|
path = tmp_path / model["model_file_path"]
|
|
assert path.is_file()
|
|
assert model["generation_params"]["file_stamp"] == file_stamp(path)
|
|
assert b05.route_coverage(tmp_path, target) == (True, 1.0)
|
|
|
|
|
|
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
|