Aislo/common_util/common_util_vector_tiles.py

"""받은 산림 자료 줌별 벡터 조각 — 웹 메르카토르 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