Merge remote-tracking branches 'origin/sub_laptop_3' and 'origin/sub_laptop_4' into main_laptop_1
This commit is contained in:
@@ -1,6 +1,6 @@
|
||||
"""D 예정노선 지형 받기(PLAN 530-1 · 561-2 · 595) — 사업 전체 한 벌 · 범위 보기 = 내려받기 없음 ·
|
||||
옛 도엽 유지 · 새 도엽만 · 범위는 줄지 않음 · 빠짐 없음 · 지적도 뺌 · 새로(reset) = 옛 범위 버림 ·
|
||||
받은 범위 좌표계 · 칸 상한 · 거리.
|
||||
받은 범위 좌표계 · 칸 상한 · 거리 · DEM = 공용 지형 창고 이어 붙임(608-1 · 값 같으면 안 다시 씀).
|
||||
|
||||
./venv/Scripts/python.exe -m pytest resources/tester/test_530_1_plan_terrain.py -q
|
||||
"""
|
||||
@@ -8,6 +8,7 @@
|
||||
import json
|
||||
from pathlib import Path
|
||||
|
||||
import numpy as np
|
||||
import pytest
|
||||
from pyproj import Transformer
|
||||
|
||||
@@ -46,15 +47,22 @@ def fake(tmp_path, monkeypatch):
|
||||
(output_dir / "vworld_satellite_meta.json").write_text(json.dumps(meta), "utf-8")
|
||||
return output_dir / "vworld_satellite.png"
|
||||
|
||||
def dem(root, processed, models_dir, route_xy, crs, methods=None, grid_m=None):
|
||||
seen["dem"].append((models_dir.name, route_xy.tolist(), crs))
|
||||
seen["grid_m"] = grid_m
|
||||
return [{"model_file_path": f"B04_PreProcess/{models_dir.name}/dtm.npz"}]
|
||||
def load_terrain(epsg, *, sheets=None, box=None, cell_m=2.0, on_progress=None):
|
||||
seen["dem"].append(list(box))
|
||||
seen["grid_m"] = cell_m
|
||||
on_progress(50, "지형 계산")
|
||||
x = (np.arange(10) + np.floor(box[0] / cell_m) + 0.5) * cell_m # 작은 격자 · 원점 칸 배수
|
||||
y = (np.arange(8) + np.floor(box[1] / cell_m) + 0.5) * cell_m
|
||||
z = seen["z"] + 0.1 * np.arange(80, dtype=np.float32).reshape(8, 10)
|
||||
grid = {"x": x, "y": y, "z": z, "valid_mask": np.isfinite(z), "cell_m": cell_m}
|
||||
return grid | {"epsg": epsg, "sheets": ["1"], "computed": [], "failed": {}}
|
||||
|
||||
seen["z"] = np.float32(100.0)
|
||||
|
||||
monkeypatch.setattr(store_mod, "download_sheet", download_sheet)
|
||||
monkeypatch.setattr(terrain, "parse_and_merge_sheets", merge)
|
||||
monkeypatch.setattr(terrain, "download_vworld_satellite_map", satellite)
|
||||
monkeypatch.setattr(terrain, "build_sheet_surface_model", dem)
|
||||
monkeypatch.setattr(terrain, "load_terrain", load_terrain)
|
||||
return tmp_path, seen
|
||||
|
||||
|
||||
@@ -63,10 +71,8 @@ def _zips(root):
|
||||
|
||||
|
||||
def _dem_box(seen) -> list[float]:
|
||||
"""가짜 DEM 이 받은 꼭짓점 상자 + B 절취 여유 = 실제 DEM 상자."""
|
||||
(x0, y0), (x1, y1) = seen["dem"][-1][1]
|
||||
m = SHEET_SURFACE_MARGIN_M
|
||||
return [x0 - m, y0 - m, x1 + m, y1 + m]
|
||||
"""창고에 물은 상자 = 받은 범위 그대로."""
|
||||
return seen["dem"][-1]
|
||||
|
||||
|
||||
def test_one_set_and_extend_keeps_old(fake) -> None:
|
||||
@@ -75,7 +81,13 @@ def test_one_set_and_extend_keeps_old(fake) -> None:
|
||||
old = _zips(root)
|
||||
assert sorted(seen["download"]) == old == sorted(first["sheets"]["downloaded"])
|
||||
assert first["satellite"] == "done" and len(first["models"]) == 1
|
||||
assert seen["dem"][-1][0] == "models_plan" # 연도 없는 한 벌
|
||||
model = first["models"][0]["model_file_path"]
|
||||
assert model == "B04_PreProcess/models_plan/dtm_sheet_laplace.npz" # 연도 없는 한 벌
|
||||
with np.load(root / model) as dem: # 뒤 소비자 꼴 — y 오름차순 · 칸 2 m
|
||||
assert dem["z"].shape == (8, 10) and dem["y"][1] > dem["y"][0]
|
||||
assert float(dem["resolution"][0]) == 2.0 and dem["valid_mask"].all()
|
||||
assert dem["z"][1, 0] == pytest.approx(101.0) # 창고 값 그대로(뒤집힘 없음)
|
||||
assert (root / model.replace(".npz", "_smooth.npz")).is_file()
|
||||
m = terrain.DEFAULT_MARGIN_M
|
||||
box = [min(A[0], B[0]) - m, min(A[1], B[1]) - m, max(A[0], B[0]) + m, max(A[1], B[1]) + m]
|
||||
assert _dem_box(seen) == pytest.approx(box) == first["box_m"]
|
||||
@@ -103,6 +115,16 @@ def test_one_set_and_extend_keeps_old(fake) -> None:
|
||||
assert seen["download"] == [] and third["satellite"] == "skip"
|
||||
assert len(seen["merge"]) == merges and third["box_m"] == grown
|
||||
|
||||
# 창고 값 같음 = DEM 안 다시 씀 · 값 바뀜(창고 다시 셈) = 다시 씀
|
||||
npz = root / third["models"][0]["model_file_path"]
|
||||
stamp = npz.stat().st_mtime_ns
|
||||
terrain.ensure_plan_terrain(root, [A, B], EPSG)
|
||||
assert npz.stat().st_mtime_ns == stamp
|
||||
seen["z"] = np.float32(200.0)
|
||||
terrain.ensure_plan_terrain(root, [A, B], EPSG)
|
||||
with np.load(npz) as dem:
|
||||
assert float(dem["z"][0, 0]) == 200.0
|
||||
|
||||
|
||||
def test_reset_drops_old_box_sheets_satellite(fake) -> None:
|
||||
"""595-2 — 새로 = 옛 받은 범위 · 병합 도엽 · 위성 범위 버림 · 좌표계 · 가운데 기록 ·
|
||||
@@ -174,7 +196,8 @@ def test_box_sheets_leave_no_gap(fake) -> None:
|
||||
|
||||
def test_empty_surface_raises_and_keeps_no_box(fake, monkeypatch) -> None:
|
||||
root, _ = fake
|
||||
monkeypatch.setattr(terrain, "build_sheet_surface_model", lambda *a, **k: [])
|
||||
seen = fake[1]
|
||||
seen["z"] = np.float32(np.nan) # 창고에 등고선 없음(빈 도엽)
|
||||
with pytest.raises(ValueError, match="사업 지형면을 만들지 못했습니다"):
|
||||
terrain.ensure_plan_terrain(root, [A, B], EPSG)
|
||||
assert terrain.received_box(root) is None
|
||||
|
||||
@@ -0,0 +1,104 @@
|
||||
"""D 0차곡 흐름누적 = 도엽별 공용 창고(PLAN 608-2) — 도엽 + 여유 200 m 조각 = 통째 셈 ·
|
||||
처음만 셈 · 이웃 도엽 지형이 새로 생기면 다시 · 임계 바꿈은 누적 그대로 · 창고 격자 아닌 DEM = 통째.
|
||||
|
||||
./venv/Scripts/python.exe -m pytest resources/tester/test_608_plan_zero_order_store.py -q
|
||||
"""
|
||||
|
||||
import numpy as np
|
||||
import pytest
|
||||
|
||||
from common_util import common_util_terrain_store as store
|
||||
from common_util.common_util_atomic import atomic_write_npz
|
||||
from common_util.common_util_json import atomic_write_json
|
||||
from D03_PlanAnalysis import D03_PlanAnalysis_ZeroOrder as zo
|
||||
|
||||
EPSG = 5187
|
||||
WEST, EAST = "36901074", "36901075"
|
||||
CELL = 20.0 # 시험은 굵은 칸(여유 200 m = 10 칸)
|
||||
|
||||
|
||||
def _height(x, y):
|
||||
"""기운 면 + 골 · 마루(파장 240 m) — 웅덩이 없음 · 작은 집수."""
|
||||
return 0.08 * x + 6.0 * np.sin(y / 38.0) + 2.0 * np.sin(x / 55.0)
|
||||
|
||||
|
||||
def _put_sheet(no: str) -> None:
|
||||
ix0, iy0, own = store._owned(no, EPSG, CELL)
|
||||
rows, cols = own.shape
|
||||
gx, gy = np.meshgrid((np.arange(cols) + ix0 + 0.5) * CELL, (np.arange(rows) + iy0 + 0.5) * CELL)
|
||||
z = _height(gx - 300000.0, gy).astype(np.float32)
|
||||
z[~own] = np.nan
|
||||
npz, key_path, _ = store._names(no, EPSG, CELL)
|
||||
atomic_write_npz(npz, z=z, ix0=np.int64(ix0), iy0=np.int64(iy0), cell_m=np.float64(CELL))
|
||||
atomic_write_json(key_path, {**store._key(EPSG, CELL, {}), "source": "gpkg", "seconds": 1})
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def shared(tmp_path, monkeypatch):
|
||||
monkeypatch.setattr(store, "TERRAIN_ROOT", tmp_path / "terrain")
|
||||
calls = []
|
||||
real = zo.d8_accumulation
|
||||
monkeypatch.setattr(zo, "d8_accumulation", lambda *a: calls.append(1) or real(*a))
|
||||
return calls
|
||||
|
||||
|
||||
def _acc(no: str):
|
||||
with np.load(zo.sheet_accumulation(no, EPSG, CELL)) as data:
|
||||
return int(data["ix0"]), int(data["iy0"]), np.asarray(data["acc"], float)
|
||||
|
||||
|
||||
def test_tiles_equal_whole_and_saved(shared) -> None:
|
||||
calls = shared
|
||||
_put_sheet(WEST)
|
||||
assert zo.sheet_accumulation(EAST, EPSG, CELL) is None # 지형 안 센 도엽
|
||||
lone = _acc(WEST)[2]
|
||||
assert len(calls) == 1
|
||||
_acc(WEST)
|
||||
assert len(calls) == 1 # 두 번째 = 저장본
|
||||
_put_sheet(EAST) # 이웃 지형이 생김 = 다시 셈
|
||||
pieces = [_acc(WEST), _acc(EAST)]
|
||||
assert len(calls) == 3
|
||||
|
||||
whole = store.load_terrain(EPSG, sheets=[WEST, EAST], cell_m=CELL)
|
||||
ix0, iy0 = round(whole["x"][0] / CELL - 0.5), round(whole["y"][0] / CELL - 0.5)
|
||||
acc = zo.d8_accumulation(whole["z"], whole["valid_mask"])
|
||||
tiled = np.zeros_like(acc)
|
||||
for px, py, part in pieces:
|
||||
dst = tiled[py - iy0 : py - iy0 + part.shape[0], px - ix0 : px - ix0 + part.shape[1]]
|
||||
np.copyto(dst, part, where=part > 0)
|
||||
valid = whole["valid_mask"]
|
||||
cells = CELL * CELL
|
||||
for low, high in ((2, 8), (1, 10)): # 0차곡 칸 = 집수 칸 수 [low, high)
|
||||
a = zo.zero_order_mask(acc, valid, CELL, low * cells, high * cells)
|
||||
b = zo.zero_order_mask(tiled, valid, CELL, low * cells, high * cells)
|
||||
assert a.any() and np.array_equal(a, b)
|
||||
# 이웃 없이 셌던 서쪽 도엽은 동쪽 경계에서 달랐음(이웃이 들어와 고쳐짐)
|
||||
assert not np.array_equal(lone, pieces[0][2])
|
||||
|
||||
|
||||
def test_layer_uses_store_and_threshold_reuses(shared, tmp_path, monkeypatch) -> None:
|
||||
"""사업 DEM 이 창고 격자 위면 도엽 누적을 이어 붙임 · 임계 바꿈 = 누적 다시 안 셈."""
|
||||
calls = shared
|
||||
monkeypatch.setattr(zo, "ZERO_ORDER_CELL_M", CELL)
|
||||
_put_sheet(WEST)
|
||||
_put_sheet(EAST)
|
||||
grid = store.load_terrain(EPSG, sheets=[WEST, EAST], cell_m=CELL)
|
||||
x, y, z = grid["x"][5:-5], grid["y"][5:-5], grid["z"][5:-5, 5:-5]
|
||||
models = tmp_path / "models"
|
||||
models.mkdir()
|
||||
np.savez(
|
||||
models / f"dtm_sheet_{zo.SHEET_SURFACE_DEFAULT_METHOD}.npz",
|
||||
x=x,
|
||||
y=y,
|
||||
z=np.nan_to_num(z, nan=0.0),
|
||||
valid_mask=np.isfinite(z),
|
||||
resolution=np.array([CELL], np.float32),
|
||||
)
|
||||
cells = CELL * CELL
|
||||
first = zo.zero_order_layer(tmp_path, models, EPSG, 2 * cells, 8 * cells)
|
||||
assert first["count"] >= 1 and first["cell_m"] == CELL
|
||||
assert len(calls) == 2 # 도엽 둘 한 번씩
|
||||
zo.zero_order_layer(tmp_path, models, EPSG, 1 * cells, 10 * cells)
|
||||
assert len(calls) == 2 # 임계 바꿈 = 저장 누적에서 칸 고르기만
|
||||
# 창고 격자에서 반 칸 어긋난 DEM = 통째 셈
|
||||
assert zo.store_accumulation_paths(x + CELL / 2, y, EPSG, CELL) is None
|
||||
Reference in New Issue
Block a user