eomsangdonandClaude Opus 5.5
4220eb6e0a
- tools/precompute_terrain.py — 도엽 목록(전국 등고선 꼭짓점이 닿는 도엽 · 길이 · 높이 차 · 지문 같으면 저장본) · 산지 먼저 · 단계 dem → zero → e07 · 이어 하기(진행 기록) · 일꾼 수 · 낮은 우선순위 · 범위 고름(--sheets 앞자리 · --box · --limit · --min-relief) · 죽은 잠금 치움 - 셈은 공용 창고 · D 0차곡 · E07 함수만 부름(원자 쓰기 · 열쇠 건너뜀) - 시험 test_612_precompute 6 Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
125 lines
5.1 KiB
Python
125 lines
5.1 KiB
Python
"""PLAN 612-1 — 전국 미리 계산 일괄 도구(`tools/precompute_terrain.py`) 지킴 시험.
|
|
|
|
도엽 목록 칸 모음 · 고름 · 산지 차례 · 좌표계 띠 · 이어 하기(성공 건너뜀 · 실패 다시) ·
|
|
죽은 잠금 치움 — 셈 함수는 가짜로 바꿔 서버 · 전국 파일 없이 돎.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import json
|
|
import os
|
|
import sys
|
|
import time
|
|
from pathlib import Path
|
|
|
|
import numpy as np
|
|
import pytest
|
|
import shapely
|
|
from pyproj import Transformer
|
|
|
|
ROOT = Path(__file__).resolve().parents[2]
|
|
sys.path.insert(0, str(ROOT))
|
|
|
|
from B04_PreProcess.B04_PreProcess_Engine_MapSheet import latlon_to_sheet5k # noqa: E402
|
|
from tools import precompute_terrain as tool # noqa: E402
|
|
|
|
|
|
def _line_5179(lon_lat: list[tuple[float, float]]) -> bytes:
|
|
to = Transformer.from_crs("EPSG:4326", "EPSG:5179", always_xy=True)
|
|
xs, ys = to.transform(*zip(*lon_lat))
|
|
return shapely.to_wkb(shapely.LineString(list(zip(xs, ys))))
|
|
|
|
|
|
def test_aggregate_cells_length_and_relief() -> None:
|
|
# 도엽 하나 안 두 줄(높이 100 · 300) + 다른 도엽 한 줄
|
|
a = [(129.0010 + 0.0001 * i, 36.9901) for i in range(200)] # 약 1.8 km
|
|
b = [(129.0010 + 0.0001 * i, 36.9951) for i in range(100)]
|
|
c = [(129.1001 + 0.0001 * i, 36.9901) for i in range(50)]
|
|
cells = tool.aggregate([_line_5179(a), _line_5179(b), _line_5179(c)], np.array([100, 300, 50]))
|
|
assert len(cells) == 2
|
|
big = max(cells.values(), key=lambda v: v[0])
|
|
assert big[1:] == [100.0, 300.0]
|
|
assert 2_300 < big[0] < 2_800 # 1.78 + 0.89 km — 솎음 현이라 조금 짧음
|
|
|
|
|
|
def test_pick_sheets_prefix_box_relief_order() -> None:
|
|
index = {
|
|
"36901001": [10.0, 100.0, 400.0],
|
|
"36901002": [30.0, 100.0, 200.0],
|
|
"36901003": [30.0, 100.0, 500.0],
|
|
"36712089": [50.0, 0.0, 20.0],
|
|
}
|
|
assert tool.pick_sheets(index) == ["36901003", "36901001"] # 높이 차 150 m 이상 · 길이 순
|
|
# 앞자리 고름은 높이 차 거름 안 씀 · 길이 같으면 높이 차 큰 것 먼저
|
|
assert tool.pick_sheets(index, sheets=["36901"]) == ["36901003", "36901002", "36901001"]
|
|
assert tool.pick_sheets(index, sheets=["36901"], limit=1) == ["36901003"]
|
|
assert tool.pick_sheets(index, sheets=["99999999"]) == [] # 전국 파일 밖 = 안 고름
|
|
assert tool.pick_sheets(index, min_relief=0, box=(127.5, 36.0, 128.0, 36.5)) == ["36712089"]
|
|
|
|
|
|
def test_epsg_band_follows_longitude() -> None:
|
|
assert tool.epsg_of(latlon_to_sheet5k(37.5, 125.5)) == 5185
|
|
assert tool.epsg_of(latlon_to_sheet5k(37.5, 127.5)) == 5186
|
|
assert tool.epsg_of("36901074") == 5187
|
|
|
|
|
|
@pytest.fixture()
|
|
def fake_store(tmp_path, monkeypatch):
|
|
monkeypatch.setattr(tool.store, "TERRAIN_ROOT", tmp_path)
|
|
monkeypatch.setattr(tool.store, "_gpkg_stamp", lambda: "stamp-1")
|
|
calls: list[tuple[str, str]] = []
|
|
fail = {"36901002"}
|
|
|
|
def fake_run(step: str, no: str):
|
|
calls.append((step, no))
|
|
if no in fail:
|
|
return step, no, {"ok": False, "error": "RuntimeError: 받기 실패"}
|
|
return step, no, {"ok": True, "s": 0.01, "fresh": True, "empty": False}
|
|
|
|
monkeypatch.setattr(tool, "run_step", fake_run)
|
|
return tmp_path, calls, fail
|
|
|
|
|
|
def test_resume_skips_done_and_retries_failed(fake_store) -> None:
|
|
root, calls, fail = fake_store
|
|
sheets = ["36901001", "36901002", "36901003"]
|
|
progress = tool.load_progress()
|
|
tool.run(sheets, ["e07", "dem"], 1, progress, say=lambda _: None)
|
|
assert [s for s, _ in calls] == ["dem"] * 3 + ["e07"] * 3 # 차례 고정
|
|
saved = json.loads((root / tool.PROGRESS_NAME).read_text(encoding="utf-8"))
|
|
assert saved["steps"]["dem"]["36901001"]["ok"] is True
|
|
assert saved["steps"]["dem"]["36901002"]["ok"] is False
|
|
|
|
calls.clear()
|
|
fail.clear()
|
|
tool.run(sheets, ["dem", "e07"], 1, tool.load_progress(), say=lambda _: None)
|
|
assert calls == [("dem", "36901002"), ("e07", "36901002")] # 성공은 건너뜀 · 실패만 다시
|
|
assert tool.summary(sheets, tool.load_progress())["dem"]["ok"] == 3
|
|
|
|
|
|
def test_progress_restarts_when_source_changes(fake_store, monkeypatch) -> None:
|
|
root, calls, _ = fake_store
|
|
tool.run(["36901001"], ["dem"], 1, tool.load_progress(), say=lambda _: None)
|
|
monkeypatch.setattr(tool.store, "_gpkg_stamp", lambda: "stamp-2")
|
|
assert tool.load_progress()["steps"]["dem"] == {}
|
|
|
|
|
|
def test_clean_dead_locks_and_old_tmp(tmp_path) -> None:
|
|
folder = tmp_path / "36901001"
|
|
folder.mkdir()
|
|
dead = folder / ".dem_5187_2m.lock"
|
|
dead.write_text("999999999 0", encoding="utf-8")
|
|
live = folder / ".zero_acc_5187_2m.lock"
|
|
live.write_text(f"{os.getpid()} 0", encoding="utf-8")
|
|
old_tmp = folder / ".dem_5187_2m.npz.abc.tmp"
|
|
old_tmp.write_bytes(b"x")
|
|
stale = time.time() - tool.store.LOCK_STALE_S - 10
|
|
os.utime(old_tmp, (stale, stale))
|
|
new_tmp = folder / ".e07_5187_2m.npz.def.tmp"
|
|
new_tmp.write_bytes(b"x")
|
|
kept = folder / "dem_5187_2m.npz"
|
|
kept.write_bytes(b"x")
|
|
assert tool.clean_dead(tmp_path) == 2
|
|
assert not dead.exists() and not old_tmp.exists()
|
|
assert live.exists() and new_tmp.exists() and kept.exists()
|