Merge remote-tracking branch 'origin/sub_laptop_2' into main_laptop_1
This commit is contained in:
@@ -54,6 +54,7 @@ def fake(tmp_path, monkeypatch):
|
||||
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.layers, "ensure_cadastral", lambda *a: {"count": 0}) # 569-1
|
||||
return tmp_path, seen
|
||||
|
||||
|
||||
|
||||
@@ -0,0 +1,181 @@
|
||||
"""D 노선분석 지도 겹(PLAN 569-1) — 도엽 겹 풀어 냄 · 지적도 1 km 조각 · 1,000 꽉 차면 넷 ·
|
||||
겹친 필지 거름 · 급경사 15° · 35° · 40° 단계 · 격자 지문 같으면 다시 안 셈.
|
||||
|
||||
./venv/Scripts/python.exe -m pytest resources/tester/test_569_1_plan_layers.py -q
|
||||
"""
|
||||
|
||||
import base64
|
||||
import io
|
||||
import itertools
|
||||
import json
|
||||
import os
|
||||
from pathlib import Path
|
||||
|
||||
import numpy as np
|
||||
import pytest
|
||||
from PIL import Image
|
||||
|
||||
from B04_PreProcess.B04_PreProcess_Engine_SheetParser import SHEET_LAYERS
|
||||
from D03_PlanAnalysis import D03_PlanAnalysis_Layers as layers
|
||||
|
||||
EPSG = 5186
|
||||
BOX = (230000.0, 450000.0, 234700.0, 452500.0) # 4.7 × 2.5 km
|
||||
|
||||
|
||||
# ── 도엽 겹 ──
|
||||
|
||||
|
||||
def test_road_layer_listed() -> None:
|
||||
assert SHEET_LAYERS["도로중심선"] == ("N3L", "A0020000")
|
||||
assert SHEET_LAYERS["하천중심선"] == ("N3L", "E0020000")
|
||||
|
||||
|
||||
def test_sheet_layer_unpacks_once(tmp_path, monkeypatch) -> None:
|
||||
processed = tmp_path / "B04_PreProcess" / "processed"
|
||||
processed.mkdir(parents=True)
|
||||
calls = []
|
||||
|
||||
def merge(store, sheets, out, keys):
|
||||
calls.append((list(sheets), keys)) # 지물 0 건 = 파일 안 씀
|
||||
|
||||
monkeypatch.setattr(layers, "parse_and_merge_sheets", merge)
|
||||
with pytest.raises(FileNotFoundError):
|
||||
layers.ensure_sheet_layer(tmp_path, "도로중심선")
|
||||
with pytest.raises(KeyError):
|
||||
layers.ensure_sheet_layer(tmp_path, "없는겹")
|
||||
(processed / "sheet_merge.key.json").write_text(
|
||||
json.dumps({"sheets": ["36712088", "36712089"], "parser": 2}), "utf-8"
|
||||
)
|
||||
path = layers.ensure_sheet_layer(tmp_path, "도로중심선")
|
||||
assert calls == [(["36712088", "36712089"], ["도로중심선"])]
|
||||
assert json.loads(path.read_text("utf-8"))["features"] == []
|
||||
layers.ensure_sheet_layer(tmp_path, "도로중심선")
|
||||
assert len(calls) == 1 # 있으면 그대로
|
||||
|
||||
|
||||
# ── 지적도 ──
|
||||
|
||||
|
||||
def test_tiles_cover_box() -> None:
|
||||
parts = layers.tiles(BOX)
|
||||
assert len(parts) == 15
|
||||
assert sum((b[2] - b[0]) * (b[3] - b[1]) for b in parts) == pytest.approx(4700 * 2500)
|
||||
assert max(b[2] for b in parts) == BOX[2] and max(b[3] for b in parts) == BOX[3]
|
||||
assert len(layers.quarter(parts[0])) == 4
|
||||
|
||||
|
||||
class FakeWfs:
|
||||
"""조각 너비가 `full_deg` 보다 넓으면 1,000 필지(꽉 참) · 아니면 셋(하나는 이웃과 같은 pnu) ·
|
||||
처음 `fail` 번은 실패(B04 꼴 — 파일 없음)."""
|
||||
|
||||
def __init__(self, full_deg: float = 1.0, fail: int = 0) -> None:
|
||||
self.full_deg, self.fail, self._count = full_deg, fail, itertools.count(1)
|
||||
self.calls = 0
|
||||
|
||||
def __call__(self, typename, filename, bounds, output_dir) -> None:
|
||||
assert typename == "lt_c_landinfobasemap"
|
||||
self.calls = next(self._count) # 조각을 여러 줄기로 받음 — count 는 한 번에 하나
|
||||
if self.calls <= self.fail:
|
||||
return
|
||||
width = bounds["max_lon"] - bounds["min_lon"]
|
||||
tag = f"{bounds['min_lon']:.5f},{bounds['min_lat']:.5f}"
|
||||
n = 1000 if width > self.full_deg else 3
|
||||
feats = [{"id": f"{tag}/{i}", "properties": {"pnu": f"{tag}/{i}"}} for i in range(n)]
|
||||
feats[-1]["properties"]["pnu"] = "SHARED"
|
||||
body = {"type": "FeatureCollection", "features": feats}
|
||||
(Path(output_dir) / filename).write_text(json.dumps(body), "utf-8")
|
||||
|
||||
|
||||
def test_full_tile_splits_into_four(monkeypatch) -> None:
|
||||
fake = FakeWfs(full_deg=0.008) # 1 km 칸 ≈ 0.011° 꽉 참 · 500 m 칸 = 셋
|
||||
monkeypatch.setattr(layers, "request_vworld_wfs", fake)
|
||||
got, failed = layers.fetch_tile((230000.0, 450000.0, 231000.0, 451000.0), EPSG)
|
||||
assert fake.calls == 5 and failed == 0 and len(got) == 12
|
||||
|
||||
|
||||
def test_dedup_by_pnu() -> None:
|
||||
feats = [{"properties": {"pnu": "1"}}, {"properties": {"pnu": "1"}}, {"id": "x"}]
|
||||
assert len(layers.dedup_parcels(feats)) == 2
|
||||
|
||||
|
||||
def test_cadastral_fetch_once_then_grow(tmp_path, monkeypatch) -> None:
|
||||
fake = FakeWfs()
|
||||
monkeypatch.setattr(layers, "request_vworld_wfs", fake)
|
||||
got = layers.ensure_cadastral(tmp_path, BOX, EPSG)
|
||||
assert got["fetched"] and fake.calls == 15 and got["failed"] == 0
|
||||
assert got["count"] == 15 * 2 + 1 # 조각마다 둘 + 같은 pnu 하나
|
||||
saved = json.loads((tmp_path / layers.CADASTRAL_FILE).read_text("utf-8"))
|
||||
assert len(saved["features"]) == got["count"]
|
||||
again = layers.ensure_cadastral(tmp_path, (231000.0, 450500.0, 233000.0, 452000.0), EPSG)
|
||||
assert not again["fetched"] and fake.calls == 15 # 덮으면 다시 안 받음
|
||||
layers.ensure_cadastral(tmp_path, (*BOX[:2], BOX[2] + 500, BOX[3]), EPSG)
|
||||
assert fake.calls == 15 + 18 # 넓어지면 상자 통째로 다시(5.2 km = 6 × 3 조각)
|
||||
|
||||
|
||||
def test_cadastral_failures(tmp_path, monkeypatch) -> None:
|
||||
monkeypatch.setattr(layers, "request_vworld_wfs", FakeWfs(fail=99))
|
||||
with pytest.raises(RuntimeError):
|
||||
layers.ensure_cadastral(tmp_path, BOX, EPSG)
|
||||
assert not (tmp_path / layers.CADASTRAL_FILE).exists()
|
||||
monkeypatch.setattr(layers, "request_vworld_wfs", FakeWfs(fail=1))
|
||||
first = layers.ensure_cadastral(tmp_path, BOX, EPSG)
|
||||
assert first["failed"] == 1 and (tmp_path / layers.CADASTRAL_FILE).exists()
|
||||
again = FakeWfs()
|
||||
monkeypatch.setattr(layers, "request_vworld_wfs", again)
|
||||
assert layers.ensure_cadastral(tmp_path, BOX, EPSG)["fetched"] and again.calls == 15
|
||||
|
||||
|
||||
# ── 급경사 ──
|
||||
|
||||
|
||||
def _plane(deg_south: float, deg_north: float, n: int = 40, cell: float = 2.0) -> np.ndarray:
|
||||
"""남쪽 반 = deg_south · 북쪽 반 = deg_north 경사(동쪽으로 오름)."""
|
||||
x = np.arange(n) * cell
|
||||
z = np.empty((n, n))
|
||||
z[: n // 2] = x * np.tan(np.radians(deg_south))
|
||||
z[n // 2 :] = x * np.tan(np.radians(deg_north))
|
||||
return z
|
||||
|
||||
|
||||
def _indices(value) -> np.ndarray:
|
||||
raw = base64.b64decode(value["image"].split(",", 1)[1])
|
||||
return np.asarray(Image.open(io.BytesIO(raw)))
|
||||
|
||||
|
||||
def test_slope_bands_and_north_up() -> None:
|
||||
z = _plane(10.0, 37.0)
|
||||
value = layers.slope_bands_png(z, np.ones(z.shape, bool), 2.0)
|
||||
ratios = [b["ratio"] for b in value["bands"]]
|
||||
assert [b["min_deg"] for b in value["bands"]] == [15.0, 35.0, 40.0]
|
||||
assert ratios[0] < 0.1 and 0.4 < ratios[1] < 0.6 and ratios[2] < 0.1 # 경계 줄만
|
||||
index = _indices(value)
|
||||
assert value["width"] == value["height"] == 40
|
||||
assert (index[5, 5:35] == 2).all() and (index[-5, 5:35] == 0).all() # 위 = 북 = 37°
|
||||
steep = layers.slope_bands_png(_plane(45.0, 20.0), np.ones((40, 40), bool), 2.0)
|
||||
assert steep["bands"][2]["ratio"] > 0.4 and steep["bands"][0]["ratio"] > 0.4
|
||||
hole = np.ones(z.shape, bool)
|
||||
hole[25:30, 10:20] = False
|
||||
assert (_indices(layers.slope_bands_png(z, hole, 2.0))[-29:-24, 10:20] == 0).all()
|
||||
|
||||
|
||||
def test_slope_layer_fingerprint(tmp_path, monkeypatch) -> None:
|
||||
models = tmp_path / "B04_PreProcess" / "models_plan"
|
||||
models.mkdir(parents=True)
|
||||
npz = models / "dtm_sheet_laplace.npz"
|
||||
z = _plane(37.0, 37.0)
|
||||
x = 200000.0 + np.arange(40) * 2.0
|
||||
y = 450000.0 + np.arange(40) * 2.0
|
||||
np.savez(npz, x=x, y=y, z=z, valid_mask=np.ones(z.shape, bool), resolution=[2.0])
|
||||
counted = []
|
||||
real = layers.slope_bands_png
|
||||
monkeypatch.setattr(layers, "slope_bands_png", lambda *a: counted.append(1) or real(*a))
|
||||
first = layers.slope_layer(tmp_path, models, EPSG)
|
||||
assert first["box_m"] == [199999.0, 449999.0, 200079.0, 450079.0] and len(first["box"]) == 4
|
||||
assert layers.slope_layer(tmp_path, models, EPSG) == first and len(counted) == 1
|
||||
np.savez(npz, x=x, y=y, z=z * 0.5, valid_mask=np.ones(z.shape, bool), resolution=[2.0])
|
||||
os.utime(npz, ns=(1, 1))
|
||||
layers.slope_layer(tmp_path, models, EPSG)
|
||||
assert len(counted) == 2 # 격자가 바뀌면 다시 셈
|
||||
(npz).unlink()
|
||||
with pytest.raises(FileNotFoundError):
|
||||
layers.slope_layer(tmp_path, models, EPSG)
|
||||
Reference in New Issue
Block a user