eomsangdonandClaude Opus 5.5
251dfd9a5b
- 새 common_util_vector_tiles: SHP(cp949 · .prj) → 3857 색인 GeoPackage(data_forest/_index) → z8~15 MVT · 원자료마다 규칙 한 줄(줌 · 단순화 · 작은 면 거름 · 남길 칸) · 창고 storage/_shared/vector_tiles/<name>/v1_<지문> · 615 꼴 잠금 · .part
- 새 common_util_wms_tiles: WMS → 256 px PNG(3857 bbox) · 처음 받고 storage/_shared/wms_tiles 기억 · 빈 · 오류 = 투명 PNG 300 s 기억 · 키 없음 = 404 · 키 안 샘
- 새 D03_PlanAnalysis_Router_Tiles: /plan/vt/{name}/{z}/{x}/{y}.pbf · /plan/wmt/{name}/{z}/{x}/{y}.png · main.py 등록
- 새 tools/precompute_vector_tiles.py: 원자료 이름 · 줌 · 경위도 상자 · 일꾼
- 시험 test_622_vector_tiles.py
Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
272 lines
11 KiB
Python
272 lines
11 KiB
Python
"""받은 산림 · 환경 자료 조각(PLAN 622-1) — 벡터 조각 MVT 풀어 읽기(겹 = name · 남길 칸) · 줌별 규칙
|
|
(자료 줌 밖 = 빈 조각 · 작은 면은 낮은 줌에서 빠짐) · 빈 조각 · 처음만 만들고 둘째 = 파일 ·
|
|
원자료 없음 · 그림 조각 대리 기억(받은 것 = 파일 · 빈 · 오류 = 짧게) · 키 없음 = 404 · 키 안 샘.
|
|
원자료 = 시험 폴더의 작은 SHP(cp949 · EPSG:5186) · 바깥 WMS 는 부르지 않음.
|
|
|
|
./venv/Scripts/python.exe -m pytest resources/tester/test_622_vector_tiles.py -q
|
|
"""
|
|
|
|
import io
|
|
import logging
|
|
import math
|
|
|
|
import geopandas as gpd
|
|
import mapbox_vector_tile
|
|
import pytest
|
|
from fastapi import FastAPI
|
|
from fastapi.testclient import TestClient
|
|
from PIL import Image
|
|
from pyproj import Transformer
|
|
from shapely.geometry import LineString, Polygon
|
|
|
|
from common_util import common_util_vector_tiles as vt
|
|
from common_util import common_util_wms_tiles as wt
|
|
from D03_PlanAnalysis.D03_PlanAnalysis_Router_Tiles import router
|
|
|
|
LON, LAT = 128.0, 36.5 # 시험 도형 자리(국내 산지)
|
|
_TO_TM = Transformer.from_crs("EPSG:4326", "EPSG:5186", always_xy=True)
|
|
|
|
|
|
def _tile(lon: float, lat: float, z: int) -> tuple[int, int, int]:
|
|
n = 2**z
|
|
r = math.radians(lat)
|
|
y = (1 - math.log(math.tan(r) + 1 / math.cos(r)) / math.pi) / 2 * n
|
|
return z, int((lon + 180) / 360 * n), int(y)
|
|
|
|
|
|
def _square(cx: float, cy: float, half: float, hole: float = 0.0) -> Polygon:
|
|
ring = [
|
|
(cx - half, cy - half),
|
|
(cx + half, cy - half),
|
|
(cx + half, cy + half),
|
|
(cx - half, cy + half),
|
|
]
|
|
holes = [
|
|
[
|
|
(cx - hole, cy - hole),
|
|
(cx - hole, cy + hole),
|
|
(cx + hole, cy + hole),
|
|
(cx + hole, cy - hole),
|
|
]
|
|
]
|
|
return Polygon(ring, holes if hole else None)
|
|
|
|
|
|
def _write(path, rows: list[dict], geoms: list) -> None:
|
|
path.parent.mkdir(parents=True, exist_ok=True)
|
|
gpd.GeoDataFrame(rows, geometry=geoms, crs="EPSG:5186").to_file(path, encoding="cp949")
|
|
|
|
|
|
@pytest.fixture
|
|
def data(tmp_path, monkeypatch):
|
|
"""시험 원자료 셋 + 창고 · 색인을 시험 폴더로."""
|
|
root = tmp_path / "data_forest"
|
|
cx, cy = _TO_TM.transform(LON, LAT)
|
|
_write(
|
|
root / "임도망도" / "시험도" / "1.shp",
|
|
[{"FRRD_NM": "용화간선", "FRRD_DV_B": "1", "FRRD_FCLTW": 4.0, "FRRD_ESTBL": "2019"}],
|
|
[LineString([(cx - 3000, cy - 2000), (cx + 3000, cy + 2000)])],
|
|
)
|
|
_write(
|
|
root / "산림입지토양도_1-25000" / "FS_IJ100_47" / "47.shp",
|
|
[{"SLTP_CD": "02"}, {"SLTP_CD": "28"}],
|
|
[_square(cx, cy, 2000, hole=500), _square(cx + 2600, cy, 40)], # 큰 면(구멍) · 작은 면
|
|
)
|
|
_write(
|
|
root / "경제림육성단지" / "ecoFrst_01_20141202" / "47" / "47.shp",
|
|
[{"STAGE_NMPL": "형제봉단지"}],
|
|
[_square(cx - 1000, cy, 800)],
|
|
)
|
|
_write(
|
|
root / "경제림육성단지" / "ecoFrst_02_20141202" / "47" / "47.shp",
|
|
[{"STAGE_NMPL": "영주북부단지"}],
|
|
[_square(cx + 1000, cy, 800)],
|
|
)
|
|
monkeypatch.setattr(vt, "DATA_ROOT", root)
|
|
monkeypatch.setattr(vt, "INDEX_ROOT", tmp_path / "index")
|
|
monkeypatch.setattr(vt, "TILES_ROOT", tmp_path / "tiles")
|
|
vt._stamp.cache_clear()
|
|
yield tmp_path
|
|
vt._stamp.cache_clear()
|
|
|
|
|
|
def _features(name: str, data: bytes) -> list[dict]:
|
|
return mapbox_vector_tile.decode(data).get(name, {"features": []})["features"]
|
|
|
|
|
|
def test_road_tile_keeps_columns(data):
|
|
feats = _features("forest_road", vt.ensure_tile("forest_road", *_tile(LON, LAT, 13)))
|
|
assert len(feats) == 1
|
|
assert feats[0]["properties"] == {"kind": 1, "name": "용화간선", "width": 4.0, "year": 2019}
|
|
assert feats[0]["geometry"]["type"] == "LineString"
|
|
|
|
|
|
def test_soil_zoom_rules(data):
|
|
# 자료 줌 밖(z8 · 9) = 빈 조각 · 저장 안 함
|
|
assert _features("forest_soil", vt.ensure_tile("forest_soil", *_tile(LON, LAT, 9))) == []
|
|
assert not list((data / "tiles").rglob("*.pbf"))
|
|
# z10 = 큰 면만(작은 면 80 m 는 32 px² 밑) · z15 = 작은 면도
|
|
low = _features("forest_soil", vt.ensure_tile("forest_soil", *_tile(LON, LAT, 10)))
|
|
assert [f["properties"] for f in low] == [{"code": "02", "name": "갈색약건산림토양"}]
|
|
lon, lat = Transformer.from_crs("EPSG:5186", "EPSG:4326", always_xy=True).transform(
|
|
*(v + d for v, d in zip(_TO_TM.transform(LON, LAT), (2600, 0), strict=True))
|
|
)
|
|
high = _features("forest_soil", vt.ensure_tile("forest_soil", *_tile(lon, lat, 15)))
|
|
assert {"code": "28", "name": "암쇄토양"} in [f["properties"] for f in high]
|
|
|
|
|
|
def test_polygon_winding_keeps_hole(data):
|
|
feats = _features("forest_soil", vt.ensure_tile("forest_soil", *_tile(LON, LAT, 12)))
|
|
big = [f for f in feats if f["properties"]["code"] == "02"]
|
|
assert big and big[0]["geometry"]["type"] == "Polygon"
|
|
assert len(big[0]["geometry"]["coordinates"]) == 2 # 바깥 + 구멍(감김 방향이 맞아야 둘)
|
|
|
|
|
|
def test_eco_owner(data):
|
|
feats = _features("eco_forest", vt.ensure_tile("eco_forest", *_tile(LON, LAT, 12)))
|
|
assert sorted((f["properties"]["own"], f["properties"]["name"]) for f in feats) == [
|
|
("국유", "형제봉단지"),
|
|
("사유", "영주북부단지"),
|
|
]
|
|
|
|
|
|
def test_empty_tile_saved_and_second_is_file(data, monkeypatch):
|
|
tile = _tile(LON + 1.0, LAT + 1.0, 12) # 국내 · 도형 없음
|
|
first = vt.ensure_tile("forest_road", *tile)
|
|
assert _features("forest_road", first) == []
|
|
assert vt.tile_path(vt._folder("forest_road"), *tile).is_file()
|
|
|
|
def boom(*_):
|
|
raise AssertionError("다시 셈")
|
|
|
|
monkeypatch.setattr(vt, "build_tile", boom)
|
|
assert vt.ensure_tile("forest_road", *tile) == first
|
|
|
|
|
|
def test_range_and_unknown(data):
|
|
with pytest.raises(KeyError):
|
|
vt.ensure_tile("forest_road", 7, 0, 0)
|
|
with pytest.raises(KeyError):
|
|
vt.ensure_tile("forest_road", 16, 0, 0)
|
|
with pytest.raises(KeyError):
|
|
vt.ensure_tile("nope", 10, 0, 0)
|
|
assert _features("forest_road", vt.ensure_tile("forest_road", 10, 0, 0)) == [] # 국내 밖
|
|
|
|
|
|
def test_no_source_is_empty(tmp_path, monkeypatch):
|
|
monkeypatch.setattr(vt, "DATA_ROOT", tmp_path / "none")
|
|
monkeypatch.setattr(vt, "TILES_ROOT", tmp_path / "tiles")
|
|
vt._stamp.cache_clear()
|
|
try:
|
|
assert _features("forest_road", vt.ensure_tile("forest_road", *_tile(LON, LAT, 12))) == []
|
|
assert not (tmp_path / "tiles").exists()
|
|
finally:
|
|
vt._stamp.cache_clear()
|
|
|
|
|
|
# ── 그림 조각 대리 ─────────────────────────────────────────────────────────────
|
|
|
|
|
|
def _png(alpha: int) -> bytes:
|
|
buffer = io.BytesIO()
|
|
Image.new("RGBA", (256, 256), (20, 160, 0, alpha)).save(buffer, format="PNG")
|
|
return buffer.getvalue()
|
|
|
|
|
|
@pytest.fixture
|
|
def wms(tmp_path, monkeypatch):
|
|
monkeypatch.setattr(wt, "WMS_ROOT", tmp_path / "wms")
|
|
monkeypatch.setattr(wt, "_SHORT", {})
|
|
calls: list[tuple] = []
|
|
answers: dict[str, object] = {}
|
|
|
|
def fetch(name, z, x, y):
|
|
calls.append((name, z, x, y))
|
|
answer = answers.get(name, _png(255))
|
|
if isinstance(answer, Exception):
|
|
raise answer
|
|
return answer
|
|
|
|
monkeypatch.setattr(wt, "fetch", fetch)
|
|
return calls, answers
|
|
|
|
|
|
def test_wms_first_fetch_then_file(wms):
|
|
calls, _ = wms
|
|
png, status = wt.ensure_tile("ecology", 13, 7000, 3200)
|
|
assert status == "" and png.startswith(b"\x89PNG")
|
|
assert wt.ensure_tile("ecology", 13, 7000, 3200) == (png, "")
|
|
assert len(calls) == 1
|
|
assert (wt.WMS_ROOT / "ecology" / "13" / "7000" / "3200.png").is_file()
|
|
|
|
|
|
def test_wms_empty_and_error_short_memory(wms, monkeypatch):
|
|
calls, answers = wms
|
|
answers["landslide"] = _png(0)
|
|
assert wt.ensure_tile("landslide", 12, 3500, 1600) == (wt.blank(), "empty")
|
|
assert wt.ensure_tile("landslide", 12, 3500, 1600)[1] == "empty"
|
|
answers["ecology"] = RuntimeError("ecology 조각을 받지 못했습니다 — URLError")
|
|
assert wt.ensure_tile("ecology", 12, 3500, 1600) == (wt.blank(), "error")
|
|
assert len(calls) == 2 # 둘째는 기억에서
|
|
assert not list(wt.WMS_ROOT.rglob("*.png"))
|
|
# 기억 시간 지나면 다시 받음
|
|
now = wt.time.monotonic()
|
|
monkeypatch.setattr(wt.time, "monotonic", lambda: now + wt.SHORT_S + 1)
|
|
answers["ecology"] = _png(255)
|
|
assert wt.ensure_tile("ecology", 12, 3500, 1600)[1] == ""
|
|
assert len(calls) == 3
|
|
|
|
|
|
def test_wms_range_and_unknown(wms):
|
|
with pytest.raises(KeyError):
|
|
wt.ensure_tile("ecology", 5, 0, 0)
|
|
with pytest.raises(KeyError):
|
|
wt.ensure_tile("nope", 10, 0, 0)
|
|
|
|
|
|
def test_wms_key_missing_404_and_key_not_leaked(tmp_path, monkeypatch, caplog):
|
|
monkeypatch.setattr(wt, "WMS_ROOT", tmp_path / "wms")
|
|
monkeypatch.setattr(wt, "_SHORT", {})
|
|
monkeypatch.delenv("SAFEMAP_API_KEY", raising=False)
|
|
app = FastAPI()
|
|
app.include_router(router)
|
|
client = TestClient(app)
|
|
pid = "ce266c60-ab1f-416b-92f2-f37d1bf539d8"
|
|
res = client.get(f"/api/projects/{pid}/plan/wmt/landslide/12/3500/1600.png")
|
|
assert res.status_code == 404 and "SAFEMAP_API_KEY" in res.json()["detail"]
|
|
|
|
# 받기 실패 = 투명 PNG + 머리 · 로그 · 응답에 키 없음
|
|
secret = "SECRET-KEY-622"
|
|
monkeypatch.setenv("SAFEMAP_API_KEY", secret)
|
|
|
|
def fail(request, timeout):
|
|
raise OSError(f"fail {request.full_url}")
|
|
|
|
monkeypatch.setattr(wt.urllib.request, "urlopen", fail)
|
|
with caplog.at_level(logging.WARNING):
|
|
res = client.get(f"/api/projects/{pid}/plan/wmt/landslide/12/3500/1600.png")
|
|
assert res.status_code == 200 and res.headers["X-Tile-Empty"] == "error"
|
|
assert res.headers["Cache-Control"] == f"private, max-age={wt.SHORT_S}"
|
|
assert secret not in caplog.text and secret.encode() not in res.content
|
|
assert client.get(f"/api/projects/{pid}/plan/wmt/nope/12/0/0.png").status_code == 404
|
|
|
|
|
|
def test_router_vector(data):
|
|
app = FastAPI()
|
|
app.include_router(router)
|
|
client = TestClient(app)
|
|
pid = "ce266c60-ab1f-416b-92f2-f37d1bf539d8"
|
|
z, x, y = _tile(LON, LAT, 13)
|
|
res = client.get(f"/api/projects/{pid}/plan/vt/forest_road/{z}/{x}/{y}.pbf")
|
|
assert res.status_code == 200 and res.headers["content-type"] == "application/x-protobuf"
|
|
assert _features("forest_road", res.content)[0]["properties"]["kind"] == 1
|
|
assert client.get(f"/api/projects/{pid}/plan/vt/nope/{z}/{x}/{y}.pbf").status_code == 404
|
|
assert client.get(f"/api/projects/{pid}/plan/vt/forest_road/7/0/0.pbf").status_code == 400
|
|
|
|
|
|
def test_key_quoting(monkeypatch):
|
|
monkeypatch.setenv("DATA_GO_KR_API_KEY", "a+b/c=")
|
|
assert wt._key(wt.SOURCES["ecology"]) == "a%2Bb%2Fc%3D"
|
|
monkeypatch.setenv("DATA_GO_KR_API_KEY", "a%2Bb")
|
|
assert wt._key(wt.SOURCES["ecology"]) == "a%2Bb"
|