feat(D03): 받은 산림 · 환경 자료 줌별 조각 — 벡터(임도 · 산림토양도 · 경제림) · 그림 대리(생태자연도 · 산사태) (PLAN 622-1 · 요구 529)

- 새 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>

D03_PlanAnalysis/D03_PlanAnalysis_Router_Tiles.py

@@ -0,0 +1,56 @@
+"""D 예정노선 받은 산림 · 환경 자료 조각 API (PLAN 622-1 · 요구 529).
+
+- `GET /api/projects/{id}/plan/vt/{name}/{z}/{x}/{y}.pbf` — 벡터 조각(forest_road · forest_soil ·
+  eco_forest · 공용 창고 · `common_util_vector_tiles`)
+- `GET …/plan/wmt/{name}/{z}/{x}/{y}.png` — 그림 조각 대리(ecology · landslide · 처음 받고 기억 ·
+  `common_util_wms_tiles`) · 빈 · 오류 조각 = 투명 PNG + `X-Tile-Empty` · 짧게 기억
+권한 = 다른 plan 길과 같음(`main.py` 등록 때 회사 보호).
+
+계산 자리 = ③ 서버 단독.
+"""
+
+from __future__ import annotations
+
+import asyncio
+from uuid import UUID
+
+from fastapi import APIRouter, HTTPException, Response
+
+from common_util import common_util_vector_tiles as vector_tiles
+from common_util import common_util_wms_tiles as wms_tiles
+
+router = APIRouter(prefix="/api/projects", tags=["D03 Plan Tiles"])
+
+KEEP = "private, max-age=86400"
+SHORT = f"private, max-age={wms_tiles.SHORT_S}"
+
+
+@router.get("/{project_id}/plan/vt/{name}/{z}/{x}/{y}.pbf", response_model=None)
+async def plan_vector_tile(project_id: UUID, name: str, z: int, x: int, y: int) -> Response:
+    """받은 산림 자료 벡터 조각 — 공용 창고 · 없으면 원자료에서 만듦 · 빈 조각 = 빈 MVT."""
+    if name not in vector_tiles.RULES:
+        raise HTTPException(status_code=404, detail=f"모르는 겹입니다 — {name}")
+    try:
+        tile = await asyncio.to_thread(vector_tiles.ensure_tile, name, z, x, y)
+    except KeyError as error:
+        raise HTTPException(status_code=400, detail="조각 번호가 범위를 벗어났습니다.") from error
+    return Response(
+        content=tile, media_type="application/x-protobuf", headers={"Cache-Control": KEEP}
+    )
+
+
+@router.get("/{project_id}/plan/wmt/{name}/{z}/{x}/{y}.png", response_model=None)
+async def plan_wms_tile(project_id: UUID, name: str, z: int, x: int, y: int) -> Response:
+    """그림 조각 대리 — 처음 받고 기억 · 키 없음 = 404 + 까닭."""
+    if name not in wms_tiles.SOURCES:
+        raise HTTPException(status_code=404, detail=f"모르는 겹입니다 — {name}")
+    try:
+        png, status = await asyncio.to_thread(wms_tiles.ensure_tile, name, z, x, y)
+    except wms_tiles.MissingKeyError as error:
+        raise HTTPException(status_code=404, detail=str(error)) from error
+    except KeyError as error:
+        raise HTTPException(status_code=400, detail="조각 번호가 범위를 벗어났습니다.") from error
+    headers = {"Cache-Control": SHORT if status else KEEP}
+    if status:
+        headers["X-Tile-Empty"] = status
+    return Response(content=png, media_type="image/png", headers=headers)

common_util/common_util_vector_tiles.py

@@ -0,0 +1,288 @@
+"""받은 산림 자료 줌별 벡터 조각 — 웹 메르카토르 z/x/y MVT 를 공용 창고에(PLAN 622-1 · 요구 529).
+
+원자료 = `resources/data_forest/` SHP(cp949 · .prj 좌표계 그대로 읽음) · 원자료마다 규칙 한 줄
+(`RULES` — 파일 · 줌 · 단순화 px · 작은 면 거름 · 남길 칸).
+색인 = 원자료를 3857 로 옮겨 남길 칸 + 면적만 담은 GeoPackage(공간 색인) — 처음 한 번
+`resources/data_forest/_index/<name>_v<규칙 판>_<원자료 지문>.gpkg`.
+조각 하나 = 색인을 조각 상자(+ 여유 8 px)로 읽음 → 작은 면 거름(SQL) → 상자로 자르고 단순화
+→ MVT(겹 = name · 지물 하나 = 도형 하나). 512 px 조각 · 범위 4096.
+창고 = `storage/_shared/vector_tiles/<name>/v<규칙 판>_<원자료 지문>/z/x/y.pbf` — 처음 볼 때
+만들어 영구 · 빈 조각도 저장 · 잠금 · `.part` → 바꿈(615 등고선 조각 꼴).
+원자료 없는 PC = 있는 판 조각을 믿음 · 없으면 빈 조각.
+길 = `D03_PlanAnalysis_Router_Tiles` `/plan/vt/{name}/{z}/{x}/{y}.pbf` · 미리 만들기 =
+`tools/precompute_vector_tiles.py`.
+"""
+
+from __future__ import annotations
+
+import functools
+import hashlib
+from dataclasses import dataclass, field
+from pathlib import Path
+from typing import Any
+
+from common_util.common_util_contour_tiles import (
+    BUFFER_PX,
+    EXTENT,
+    TILE_PX,
+    WORLD_M,
+    in_bounds,
+    tile_bounds,
+)
+from common_util.common_util_terrain_store import _Lock
+from config.config_system import PROJECT_ROOT, STORAGE_BASE_DIR
+
+DATA_ROOT = Path(PROJECT_ROOT) / "resources" / "data_forest"
+INDEX_ROOT = DATA_ROOT / "_index"
+TILES_ROOT = Path(STORAGE_BASE_DIR) / "_shared" / "vector_tiles"
+RULE_VERSION = 1
+MIN_ZOOM, MAX_ZOOM = 8, 15
+ENCODING = "cp949"
+
+# 산림입지토양도 토양형 코드(SLTP_CD) — 설명서 `DATA014_산림입지도(25000).pdf` 코드 정의서
+SOIL_TYPES = {
+    "01": "갈색건조산림토양",
+    "02": "갈색약건산림토양",
+    "03": "갈색적윤산림토양",
+    "04": "갈색약습산림토양",
+    "05": "적색계갈색건조산림토양",
+    "06": "적색계갈색약건산림토양",
+    "07": "적색건조산림토양",
+    "08": "적색약건산림토양",
+    "09": "황색건조산림토양",
+    "10": "암적색건조산림토양",
+    "11": "암적색약건산림토양",
+    "12": "암적색적윤산림토양",
+    "13": "암적갈색건조산림토양",
+    "14": "암적갈색약건산림토양",
+    "15": "회갈색건조산림토양",
+    "16": "회갈색약건산림토양",
+    "17": "화산회건조산림토양",
+    "18": "화산회약건산림토양",
+    "19": "화산회적윤산림토양",
+    "20": "화산회습윤산림토양",
+    "21": "화산회자갈많은산림토양",
+    "22": "화산회성적색건조산림토양",
+    "23": "화산회성적색약건산림토양",
+    "24": "약침식토양",
+    "25": "강침식토양",
+    "26": "사방지토양",
+    "27": "미숙토양",
+    "28": "암쇄토양",
+    "29": "전석지",
+    "82": "제지",
+    "91": "주거지",
+    "92": "초지",
+    "93": "경작지",
+    "94": "수채",
+    "95": "과수원",
+    "97": "군사지역",
+    "99": "기타",
+}
+
+
+def _text(value: Any) -> str:
+    return "" if value is None or value != value else str(value).strip()
+
+
+def _road(row: dict[str, Any], _: Path) -> dict[str, Any]:
+    kind = _text(row.get("FRRD_DV_B"))
+    year = _text(row.get("FRRD_ESTBL"))
+    width = row.get("FRRD_FCLTW")
+    return {
+        "kind": int(kind) if kind.isdigit() else 0,
+        "name": _text(row.get("FRRD_NM")),
+        "width": float(width) if width == width and width is not None else 0.0,
+        "year": int(year) if year.isdigit() else 0,
+    }
+
+
+def _soil(row: dict[str, Any], _: Path) -> dict[str, Any]:
+    code = _text(row.get("SLTP_CD"))
+    return {"code": code, "name": SOIL_TYPES.get(code, "")}
+
+
+def _eco(row: dict[str, Any], path: Path) -> dict[str, Any]:
+    own = "국유" if "ecoFrst_01" in path.as_posix() else "사유"
+    return {"own": own, "name": _text(row.get("STAGE_NMPL"))}
+
+
+@dataclass(frozen=True)
+class Rule:
+    """원자료 하나 — 파일 · 자료 줌 · 단순화 px · 작은 면 거름(px² · 줌별) · 칸 뽑기."""
+
+    globs: tuple[str, ...]
+    columns: tuple[str, ...]
+    props: Any
+    min_zoom: int = MIN_ZOOM
+    simplify_px: float = 1.0
+    min_area_px: dict[int, float] = field(default_factory=dict)
+
+
+RULES: dict[str, Rule] = {
+    "forest_road": Rule(
+        ("임도망도/*/*.shp",),
+        ("FRRD_NM", "FRRD_DV_B", "FRRD_FCLTW", "FRRD_ESTBL"),
+        _road,
+    ),
+    "forest_soil": Rule(
+        ("산림입지토양도_1-25000/*/*.shp",),
+        ("SLTP_CD",),
+        _soil,
+        min_zoom=10,
+        min_area_px={10: 32.0, 11: 16.0, 12: 8.0, 13: 2.0},
+    ),
+    "eco_forest": Rule(
+        ("경제림육성단지/ecoFrst_01_*/*/*.shp", "경제림육성단지/ecoFrst_02_*/*/*.shp"),
+        ("STAGE_NMPL",),
+        _eco,
+        min_area_px={8: 16.0, 9: 8.0, 10: 4.0, 11: 2.0, 12: 1.0},
+    ),
+}
+
+
+def sources(name: str) -> list[Path]:
+    """원자료 SHP 목록(차례 고정)."""
+    rule = RULES[name]
+    return sorted(p for pattern in rule.globs for p in DATA_ROOT.glob(pattern))
+
+
+@functools.cache
+def _stamp(name: str) -> str | None:
+    """원자료 지문(SHP · DBF 크기 · 수정 시각) — 없으면 None.
+    ponytail: 프로세스마다 한 번 셈 · 원자료를 바꾸면 서버를 다시 켬."""
+    files = sources(name)
+    if not files:
+        return None
+    parts = []
+    for shp in files:
+        for path in (shp, shp.with_suffix(".dbf")):
+            stat = path.stat()
+            parts.append(
+                f"{path.relative_to(DATA_ROOT).as_posix()}:{stat.st_size}:{stat.st_mtime_ns}"
+            )
+    return hashlib.sha1("\n".join(parts).encode()).hexdigest()[:8]
+
+
+def _folder(name: str) -> Path | None:
+    """지금 판 창고 — 원자료 지문 · 없으면 있는 판 중 마지막 · 그것도 없으면 None."""
+    tag = _stamp(name)
+    if tag:
+        return TILES_ROOT / name / f"v{RULE_VERSION}_{tag}"
+    found = sorted((TILES_ROOT / name).glob(f"v{RULE_VERSION}_*"))
+    return found[-1] if found else None
+
+
+def ensure_index(name: str) -> Path | None:
+    """3857 색인 GeoPackage — 없으면 원자료에서 만듦(잠금 · `.part` → 바꿈) · 원자료 없음 = None."""
+    import geopandas as gpd
+    import pyogrio
+
+    tag = _stamp(name)
+    if tag is None:
+        return None
+    path = INDEX_ROOT / f"{name}_v{RULE_VERSION}_{tag}.gpkg"
+    if path.is_file():
+        return path
+    rule = RULES[name]
+    with _Lock(path.with_suffix(".lock")):
+        if path.is_file():
+            return path
+        part = path.with_suffix(".part.gpkg")
+        part.unlink(missing_ok=True)
+        for shp in sources(name):
+            frame = pyogrio.read_dataframe(shp, encoding=ENCODING, columns=list(rule.columns))
+            frame = frame[frame.geometry.notna() & ~frame.geometry.is_empty].to_crs(3857)
+            rows = frame.drop(columns="geometry").to_dict("records")
+            out = gpd.GeoDataFrame(
+                [rule.props(row, shp) for row in rows] or None,
+                geometry=frame.geometry.values,
+                crs=3857,
+            )
+            out["area"] = out.geometry.area
+            pyogrio.write_dataframe(
+                out, part, layer=name, driver="GPKG", promote_to_multi=True, append=part.exists()
+            )
+        part.replace(path)
+    return path
+
+
+def _empty(name: str) -> bytes:
+    import mapbox_vector_tile
+
+    return mapbox_vector_tile.encode([{"name": name, "features": []}])
+
+
+def build_tile(name: str, z: int, x: int, y: int) -> bytes:
+    """조각 하나 만들기(창고 안 씀) — 색인에서 읽어 MVT."""
+    import mapbox_vector_tile
+    import pyogrio
+    import shapely
+
+    rule = RULES[name]
+    if not (rule.min_zoom <= z <= MAX_ZOOM) or not in_bounds(z, x, y):
+        return _empty(name)
+    index = ensure_index(name)
+    if index is None:
+        return _empty(name)
+    west, south, east, north = tile_bounds(z, x, y)
+    pad = (east - west) * BUFFER_PX / TILE_PX
+    box = (west - pad, south - pad, east + pad, north + pad)
+    px_m = WORLD_M / (TILE_PX * 2**z)
+    min_area = rule.min_area_px.get(z, 0.0) * px_m * px_m
+    frame = pyogrio.read_dataframe(
+        index, layer=name, bbox=box, where=f"area >= {min_area}" if min_area else None
+    )
+    if frame.empty:
+        return _empty(name)
+    geoms = shapely.clip_by_rect(frame.geometry.values, *box)
+    # 고리 보존 안 함 = 빠름(작게 무너진 면은 빠짐 · 그림만 쓰므로 유효성 안 따짐)
+    geoms = shapely.simplify(geoms, rule.simplify_px * px_m, preserve_topology=False)
+    # 조각 칸 좌표(위 = +y)로 묶어 옮김(인코더 한 도형씩 = 느림 · 반올림 · 겹친 점은 인코더) ·
+    # normalize = 바깥 고리 시계 방향 → 인코더가 y 를 뒤집으면 MVT 규칙(바깥 = 양의 넓이)
+    scale = EXTENT / (east - west)
+    geoms = shapely.normalize(shapely.transform(geoms, lambda xy: (xy - (west, south)) * scale))
+    keep = ~shapely.is_empty(geoms)
+    if not keep.any():
+        return _empty(name)
+    props = frame.drop(columns=["geometry", "area"]).to_dict("records")
+    features = [
+        {"geometry": geom, "properties": prop}
+        for geom, prop, ok in zip(geoms, props, keep, strict=True)
+        if ok
+    ]
+    return mapbox_vector_tile.encode(
+        [{"name": name, "features": features}],
+        default_options={"extents": EXTENT, "check_winding_order": False},
+    )
+
+
+def tile_path(folder: Path, z: int, x: int, y: int) -> Path:
+    return folder / str(z) / str(x) / f"{y}.pbf"
+
+
+def ensure_tile(name: str, z: int, x: int, y: int) -> bytes:
+    """조각 바이트 — 창고에 있으면 그대로 · 없으면 만들어 저장(잠금 · `.part` → 바꿈).
+    모르는 name · 범위 밖 번호 = `KeyError`."""
+    if name not in RULES:
+        raise KeyError(name)
+    if not (MIN_ZOOM <= z <= MAX_ZOOM) or not (0 <= x < 2**z and 0 <= y < 2**z):
+        raise KeyError(f"{z}/{x}/{y}")
+    if z < RULES[name].min_zoom or not in_bounds(z, x, y):
+        return _empty(name)
+    folder = _folder(name)
+    if folder is None:
+        return _empty(name)
+    path = tile_path(folder, z, x, y)
+    if path.is_file():
+        return path.read_bytes()
+    if _stamp(name) is None:
+        return _empty(name)  # 원자료 없는 PC = 남은 판에 없으면 빈 조각(저장 안 함)
+    with _Lock(path.with_suffix(".lock")):
+        if path.is_file():
+            return path.read_bytes()
+        data = build_tile(name, z, x, y)
+        part = path.with_suffix(".part")
+        part.write_bytes(data)
+        part.replace(path)
+        return data

common_util/common_util_wms_tiles.py

@@ -0,0 +1,156 @@
+"""그림 조각 대리 — 바깥 WMS 를 웹 메르카토르 z/x/y 256 px PNG 로(PLAN 622-1 · 요구 529).
+
+- 겹 = `SOURCES`(ecology 생태자연도 오픈API · landslide 생활안전지도 산사태위험지도) · 조각 상자
+  = EPSG:3857 bbox 그대로 요청(두 곳 모두 3857 받음 · 옮겨 그리기 없음).
+- 처음 받고 기억 = `storage/_shared/wms_tiles/<name>/z/x/y.png`(영구 · `.part` → 바꿈) · 빈 조각
+  (전부 투명) · 오류 조각 = 서버 메모리에 `SHORT_S` 초만 기억하고 투명 PNG.
+- 키 = `.env`(`SOURCES` 의 key_env) · 없으면 `MissingKeyError` · 키 값은 어떤 로그 · 응답에도 안 냄
+  (오류 글에 주소를 넣지 않음).
+길 = `D03_PlanAnalysis_Router_Tiles` `/plan/wmt/{name}/{z}/{x}/{y}.png`.
+"""
+
+from __future__ import annotations
+
+import functools
+import io
+import logging
+import os
+import threading
+import time
+import urllib.parse
+import urllib.request
+from dataclasses import dataclass, field
+from pathlib import Path
+from uuid import uuid4
+
+from common_util.common_util_contour_tiles import tile_bounds
+from config.config_system import STORAGE_BASE_DIR
+from D03_PlanAnalysis.D03_PlanAnalysis_Landslide import (
+    LANDSLIDE_KEY_ENV,
+    LANDSLIDE_LAYER,
+    LANDSLIDE_URL,
+    PNG_SIGNATURE,
+)
+
+logger = logging.getLogger(__name__)
+
+WMS_ROOT = Path(STORAGE_BASE_DIR) / "_shared" / "wms_tiles"
+MIN_ZOOM, MAX_ZOOM = 6, 19
+TILE_PX = 256
+SHORT_S = 300
+TIMEOUT_S = 30  # 생태자연도 한 조각이 8~15 s 걸릴 때 있음
+
+
+@dataclass(frozen=True)
+class Source:
+    """WMS 하나 — 주소 · 키 칸 · 고정 인자 · 상자 좌표계 칸 이름."""
+
+    url: str
+    key_env: str
+    params: dict[str, str] = field(default_factory=dict)
+    srs_param: str = "srs"
+
+
+SOURCES: dict[str, Source] = {
+    "ecology": Source(
+        "https://apis.data.go.kr/B553084/ecoapi/EcologyzmpService/wms/getEcologyzmpWMS",
+        "DATA_GO_KR_API_KEY",
+        {"format": "image/png", "transparent": "TRUE"},
+        srs_param="crs",
+    ),
+    "landslide": Source(
+        LANDSLIDE_URL,
+        LANDSLIDE_KEY_ENV,
+        {"layers": LANDSLIDE_LAYER, "styles": "", "format": "image/png", "transparent": "TRUE"},
+    ),
+}
+
+
+class MissingKeyError(RuntimeError):
+    """`.env` 에 키 없음."""
+
+
+_SHORT: dict[tuple[str, int, int, int], tuple[float, str]] = {}
+_SHORT_LOCK = threading.Lock()
+
+
+@functools.cache
+def blank() -> bytes:
+    """투명 256 px PNG."""
+    from PIL import Image
+
+    buffer = io.BytesIO()
+    Image.new("RGBA", (TILE_PX, TILE_PX), (0, 0, 0, 0)).save(buffer, format="PNG")
+    return buffer.getvalue()
+
+
+def _key(source: Source) -> str:
+    """키 — 인코딩된 키(`%` 있음)는 그대로 · 아니면 주소용으로 바꿈 · 없으면 `MissingKeyError`."""
+    key = os.environ.get(source.key_env, "").strip()
+    if not key:
+        raise MissingKeyError(f"키 없음 — .env {source.key_env}")
+    return key if "%" in key else urllib.parse.quote(key, safe="")
+
+
+def fetch(name: str, z: int, x: int, y: int) -> bytes:
+    """바깥 WMS 에서 조각 하나(PNG) — 키 없음 = `MissingKeyError` · 실패 = `RuntimeError`."""
+    source = SOURCES[name]
+    query = {
+        **source.params,
+        source.srs_param: "EPSG:3857",
+        "bbox": ",".join(f"{v:.3f}" for v in tile_bounds(z, x, y)),
+        "width": str(TILE_PX),
+        "height": str(TILE_PX),
+    }
+    url = f"{source.url}?serviceKey={_key(source)}&{urllib.parse.urlencode(query)}"
+    request = urllib.request.Request(url, headers={"User-Agent": "Mozilla/5.0"})
+    try:
+        with urllib.request.urlopen(request, timeout=TIMEOUT_S) as response:
+            body = response.read()
+    except OSError as error:
+        raise RuntimeError(f"{name} 조각을 받지 못했습니다 — {type(error).__name__}") from None
+    if not body.startswith(PNG_SIGNATURE):
+        raise RuntimeError(f"{name} 조각을 받지 못했습니다 — 그림이 아닌 응답")
+    return body
+
+
+def is_blank(png: bytes) -> bool:
+    """전부 투명한 그림인가."""
+    from PIL import Image
+
+    image = Image.open(io.BytesIO(png))
+    return image.convert("RGBA").getchannel("A").getextrema()[1] == 0
+
+
+def ensure_tile(name: str, z: int, x: int, y: int) -> tuple[bytes, str]:
+    """(PNG, 상태) — 상태 "" = 받은 그림(창고) · "empty" · "error" = 투명 PNG(짧게 기억).
+    모르는 name · 범위 밖 번호 = `KeyError` · 키 없음 = `MissingKeyError`."""
+    if name not in SOURCES:
+        raise KeyError(name)
+    if not (MIN_ZOOM <= z <= MAX_ZOOM) or not (0 <= x < 2**z and 0 <= y < 2**z):
+        raise KeyError(f"{z}/{x}/{y}")
+    path = WMS_ROOT / name / str(z) / str(x) / f"{y}.png"
+    if path.is_file():
+        return path.read_bytes(), ""
+    tile = (name, z, x, y)
+    with _SHORT_LOCK:
+        until, status = _SHORT.get(tile, (0.0, ""))
+    if until > time.monotonic():
+        return blank(), status
+    try:
+        png = fetch(name, z, x, y)
+        status = "empty" if is_blank(png) else ""
+    except MissingKeyError:
+        raise
+    except (RuntimeError, OSError) as error:  # OSError = 그림을 못 읽음
+        logger.warning("%s", error)
+        status = "error"
+    if status:
+        with _SHORT_LOCK:
+            _SHORT[tile] = (time.monotonic() + SHORT_S, status)
+        return blank(), status
+    path.parent.mkdir(parents=True, exist_ok=True)
+    part = path.with_suffix(f".{uuid4().hex}.part")
+    part.write_bytes(png)
+    part.replace(path)  # 반쯤 쓴 조각을 남기지 않음
+    return png, ""

main.py

@@ -77,6 +77,7 @@ from D01_PlanRegister.D01_PlanRegister_Router import dashboard_router as d01_das
 from D01_PlanRegister.D01_PlanRegister_Router import router as d01_plan_register_router
 from D03_PlanAnalysis.D03_PlanAnalysis_Network import router as d03_plan_network_router
 from D03_PlanAnalysis.D03_PlanAnalysis_Router import router as d03_plan_analysis_router
+from D03_PlanAnalysis.D03_PlanAnalysis_Router_Tiles import router as d03_plan_tiles_router
 from D04_PlanConfirm.D04_PlanConfirm_Link import dashboard_router as d04_link_dashboard_router
 from D04_PlanConfirm.D04_PlanConfirm_Link import router as d04_plan_link_router
 from D04_PlanConfirm.D04_PlanConfirm_Router import router as d04_plan_confirm_router
@@ -494,6 +495,7 @@ app.include_router(tiles_router, dependencies=protected_with_company)
 app.include_router(b05_route_router, dependencies=protected_with_company)
 app.include_router(b05_route_lifecycle_router, dependencies=protected_with_company)
 app.include_router(d03_plan_analysis_router, dependencies=protected_with_company)
+app.include_router(d03_plan_tiles_router, dependencies=protected_with_company)
 app.include_router(d03_plan_network_router, dependencies=protected_with_company)
 app.include_router(d04_plan_confirm_router, dependencies=protected_with_company)
 app.include_router(d04_plan_link_router, dependencies=protected_with_company)

resources/tester/test_622_vector_tiles.py

@@ -0,0 +1,271 @@
+"""받은 산림 · 환경 자료 조각(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"

tools/precompute_vector_tiles.py

@@ -0,0 +1,108 @@
+# -*- coding: utf-8 -*-
+"""받은 산림 자료 벡터 조각 미리 만들기(PLAN 622-1) — 줌아웃 조각(z8~12)을 공용 창고에 미리.
+
+실행:  ./venv/Scripts/python.exe tools/precompute_vector_tiles.py forest_road --workers 2
+       --zooms 8-12  줌 범위 · --bbox 129.0,36.7,129.3,36.9  경위도 상자만(시험 지역)
+       --status 남은 수 · 크기만
+
+- 원자료 이름 = `common_util_vector_tiles.RULES`(forest_road · forest_soil · eco_forest).
+- 색인(GeoPackage) 먼저 한 번 → 조각 목록 = 도형 상자가 닿는 z/x/y(가장 큰 줌에서 세고 위 줌은
+  부모) · 낮은 줌부터.
+- 셈 = `ensure_tile`(창고에 있으면 건너뜀 = 이어 하기 · 잠금 · `.part` → 바꿈) · 낮은 우선순위 ·
+  일꾼 여럿.
+"""
+
+from __future__ import annotations
+
+import argparse
+import sys
+import time
+from concurrent.futures import ProcessPoolExecutor, as_completed
+from pathlib import Path
+
+import numpy as np
+
+ROOT = Path(__file__).resolve().parent.parent
+if str(ROOT) not in sys.path:
+    sys.path.insert(0, str(ROOT))
+
+from common_util import common_util_vector_tiles as vt  # noqa: E402
+from tools.precompute_terrain import below_normal  # noqa: E402
+
+
+def tiles_for(bounds: np.ndarray, zooms: range) -> list[tuple[int, int, int]]:
+    """도형 상자(4 × N · 3857)가 닿는 조각(낮은 줌부터)."""
+    top = zooms.stop - 1
+    size = vt.WORLD_M / 2**top
+    half = vt.WORLD_M / 2
+    x0 = np.floor((bounds[0] + half) / size).astype(np.int64)
+    x1 = np.floor((bounds[2] + half) / size).astype(np.int64)
+    y0 = np.floor((half - bounds[3]) / size).astype(np.int64)
+    y1 = np.floor((half - bounds[1]) / size).astype(np.int64)
+    found: set[tuple[int, int]] = set()
+    for a, b, c, d in zip(x0, x1, y0, y1, strict=True):
+        found.update((x, y) for x in range(a, b + 1) for y in range(c, d + 1))
+    tiles = {(z, x >> (top - z), y >> (top - z)) for z in zooms for x, y in found}
+    return sorted(t for t in tiles if vt.in_bounds(*t))
+
+
+def build(job: tuple[str, int, int, int]) -> tuple[tuple[str, int, int, int], int, float]:
+    started = time.monotonic()
+    size = len(vt.ensure_tile(*job))
+    return job, size, round(time.monotonic() - started, 2)
+
+
+def main(argv: list[str] | None = None) -> int:
+    parser = argparse.ArgumentParser(description="받은 산림 자료 벡터 조각 미리 만들기(PLAN 622-1)")
+    parser.add_argument("name", choices=sorted(vt.RULES))
+    parser.add_argument("--zooms", default="8-12")
+    parser.add_argument("--workers", type=int, default=2)
+    parser.add_argument("--bbox", default="", help="경위도 서,남,동,북 — 이 상자에 닿는 도형만")
+    parser.add_argument("--status", action="store_true")
+    args = parser.parse_args(argv)
+    lo, _, hi = args.zooms.partition("-")
+    zooms = range(max(int(lo), vt.RULES[args.name].min_zoom), int(hi or lo) + 1)
+    if int(lo) < vt.MIN_ZOOM or zooms.stop - 1 > vt.MAX_ZOOM:
+        parser.error(f"줌 = {vt.MIN_ZOOM}~{vt.MAX_ZOOM}")
+
+    import pyogrio
+    from pyproj import Transformer
+
+    below_normal()
+    started = time.monotonic()
+    index = vt.ensure_index(args.name)
+    folder = vt._folder(args.name)
+    if index is None or folder is None:
+        raise SystemExit(f"원자료 없음: {vt.DATA_ROOT} {vt.RULES[args.name].globs}")
+    print(f"색인 {index.name} · {time.monotonic() - started:.0f} s", flush=True)
+    _, bounds = pyogrio.read_bounds(index, layer=args.name)
+    if args.bbox:
+        w, s, e, n = (float(v) for v in args.bbox.split(","))
+        to_merc = Transformer.from_crs("EPSG:4326", "EPSG:3857", always_xy=True)
+        (mw, me), (ms, mn) = to_merc.transform([w, e], [s, n])
+        hit = (bounds[2] >= mw) & (bounds[0] <= me) & (bounds[3] >= ms) & (bounds[1] <= mn)
+        bounds = bounds[:, hit]
+    tiles = tiles_for(bounds, zooms)
+    todo = [t for t in tiles if not vt.tile_path(folder, *t).is_file()]
+    print(f"조각 {len(tiles)} · 남음 {len(todo)} · 창고 {folder} · 일꾼 {args.workers}", flush=True)
+    if not args.status and todo:
+        started = time.monotonic()
+        with ProcessPoolExecutor(args.workers, initializer=below_normal) as pool:
+            futures = [pool.submit(build, (args.name, *t)) for t in todo]
+            for n, future in enumerate(as_completed(futures), 1):
+                (_, z, x, y), size, seconds = future.result()
+                if n % 50 == 0 or n == len(todo):
+                    elapsed = time.monotonic() - started
+                    print(
+                        f"{n}/{len(todo)} z{z}/{x}/{y} {size} B {seconds} s · {elapsed:.0f} s",
+                        flush=True,
+                    )
+    for z in zooms:
+        paths = [vt.tile_path(folder, *t) for t in tiles if t[0] == z]
+        have = [p.stat().st_size for p in paths if p.is_file()]
+        print(f"z{z} 조각 {len(paths)} · 있음 {len(have)} · {sum(have) / 1e6:.1f} MB", flush=True)
+    return 0
+
+
+if __name__ == "__main__":
+    raise SystemExit(main())