"""D 노선분석 지도 겹(PLAN 569-1 · 595-1) — 도엽 겹 풀어 냄 · 지적도 1 km 조각 · 1,000 꽉 차면 넷 · 겹친 필지 거름 · 실패 조각만 다시 · 급경사 % 값 그림 · 1 % 칸 수(603) · 격자 지문 같으면 다시 안 셈. ./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_same_odd_box_kept(tmp_path, monkeypatch) -> None: monkeypatch.setattr(layers, "request_vworld_wfs", FakeWfs()) odd = (243862.86880613962, 544271.5644142976, 245002.02265078452, 545528.8660732408) assert layers.ensure_cadastral(tmp_path, odd, EPSG)["fetched"] assert not layers.ensure_cadastral(tmp_path, odd, EPSG)["fetched"] # 소수 끝자리까지 같은 상자 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() assert ( len(json.loads((tmp_path / layers.CADASTRAL_KEY).read_text("utf-8"))["failed_tiles"]) == 1 ) # 595-1 — 실패 조각만 다시 · 필지는 옛 것 + 그 조각 = 통째로 받은 것과 같음 again = FakeWfs() monkeypatch.setattr(layers, "request_vworld_wfs", again) retry = layers.ensure_cadastral(tmp_path, BOX, EPSG) assert retry["fetched"] and again.calls == 1 and retry["failed"] == 0 assert retry["count"] == 15 * 2 + 1 assert not layers.ensure_cadastral(tmp_path, BOX, EPSG)["fetched"] and again.calls == 1 # 옛 열쇠(실패 조각 기록 없음) = 통째로 key = json.loads((tmp_path / layers.CADASTRAL_KEY).read_text("utf-8")) key.update(failed=1) key.pop("failed_tiles") (tmp_path / layers.CADASTRAL_KEY).write_text(json.dumps(key), "utf-8") layers.ensure_cadastral(tmp_path, BOX, EPSG) assert again.calls == 1 + 15 # 다시 받기도 모두 실패 = 옛 필지 그대로 · 실패 조각 남김(오류 아님) key.update(failed=1, failed_tiles=[list(layers.tiles(BOX)[0])]) (tmp_path / layers.CADASTRAL_KEY).write_text(json.dumps(key), "utf-8") monkeypatch.setattr(layers, "request_vworld_wfs", FakeWfs(fail=99)) kept = layers.ensure_cadastral(tmp_path, BOX, EPSG) assert kept["failed"] == 1 and kept["count"] == 15 * 2 + 1 # ── 급경사 ── 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_values_and_north_up() -> None: """값 그림(603) — 칸 = 기울기 % 내림 · 위 = 북 · 빈 칸 · 끝 줄 = 255 · 칸 수 합 = 유효 칸.""" z = _plane(10.0, 37.0) value = layers.slope_values_png(z, np.ones(z.shape, bool), 2.0) index = _indices(value) assert value["width"] == value["height"] == 40 and "bands" not in value assert (index[5, 5:35] == 75).all() and (index[-5, 5:35] == 17).all() # 위 = 북 = 37° = 75 % assert (index[0] == layers.SLOPE_EMPTY).all() # 격자 끝 = 셈 못 함 assert sum(value["counts"]) == int((index != layers.SLOPE_EMPTY).sum()) == 38 * 38 assert len(value["counts"]) == layers.SLOPE_MAX_PCT + 1 cliff = layers.slope_values_png(_plane(80.0, 80.0), np.ones((40, 40), bool), 2.0) assert cliff["counts"][layers.SLOPE_MAX_PCT] == 38 * 38 # 567 % → 250 % 이상 칸 hole = np.ones(z.shape, bool) hole[25:30, 10:20] = False holed = _indices(layers.slope_values_png(z, hole, 2.0)) assert (holed[-29:-24, 10:20] == layers.SLOPE_EMPTY).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_values_png monkeypatch.setattr(layers, "slope_values_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)