"""MVT 벡터 타일 생성 유틸리티.""" from __future__ import annotations import math from pathlib import Path from typing import Any try: import geopandas as gpd from shapely.geometry import box except ImportError: gpd = None box = None try: import mapbox_vector_tile except ImportError: mapbox_vector_tile = None # 메모리 캐시: {project_id}_{layer_name}: GeoDataFrame _gdf_cache: dict[str, Any] = {} def num2deg(xtile: int, ytile: int, zoom: int) -> tuple[float, float]: """Z/X/Y 타일 인덱스를 경도/위도(WGS84) 좌표로 변환합니다.""" n = 1.0 << zoom lon_deg = xtile / n * 360.0 - 180.0 lat_rad = math.atan(math.sinh(math.pi * (1 - 2 * ytile / n))) lat_deg = math.degrees(lat_rad) return lon_deg, lat_deg def get_tile_bbox(xtile: int, ytile: int, zoom: int) -> tuple[float, float, float, float]: """Z/X/Y 타일의 경위도 Bounding Box (min_lon, min_lat, max_lon, max_lat)를 구합니다.""" left, top = num2deg(xtile, ytile, zoom) right, bottom = num2deg(xtile + 1, ytile + 1, zoom) return min(left, right), min(top, bottom), max(left, right), max(top, bottom) def load_geojson_cached(filepath: Path, cache_key: str) -> Any: """GeoJSON 파일을 한 번 읽으면 메모리에 GeoDataFrame으로 캐싱합니다.""" global _gdf_cache if cache_key not in _gdf_cache: if not filepath.exists(): raise FileNotFoundError(f"File not found: {filepath}") if gpd is None: raise ImportError("geopandas가 설치되어 있지 않습니다.") gdf = gpd.read_file(filepath) if gdf.crs is None: gdf.set_crs(epsg=4326, inplace=True) if gdf.crs.to_epsg() != 4326: gdf = gdf.to_crs(epsg=4326) _gdf_cache[cache_key] = gdf return _gdf_cache[cache_key] def generate_mvt_tile( filepath: Path, cache_key: str, z: int, x: int, y: int, layer_name: str = "contour" ) -> bytes: """주어진 GeoJSON 파일에서 Z/X/Y 영역에 해당하는 데이터를 필터링하고 MVT 타일 바이너리로 인코딩합니다. """ if mapbox_vector_tile is None: # mapbox_vector_tile 패키지가 없을 경우 빈 바이트 리턴하여 다운 방지 return b"" gdf = load_geojson_cached(filepath, cache_key) # 1. 타일 경계 구하기 min_lon, min_lat, max_lon, max_lat = get_tile_bbox(x, y, z) tile_geom = box(min_lon, min_lat, max_lon, max_lat) # 2. 해당 영역 내에 공간적으로 접하는(Intersects) 데이터 신속 조회 possible_matches_index = list(gdf.sindex.intersection((min_lon, min_lat, max_lon, max_lat))) possible_matches = gdf.iloc[possible_matches_index] # 정확하게 겹치는 피처만 정밀 필터링 및 타일 경계로 클리핑(자르기) clipped_gdf = possible_matches.clip(tile_geom) if clipped_gdf.empty: return mapbox_vector_tile.encode([]) # 3. GeoJSON 스타일 피처 목록 생성 features = [] extent = 4096 for idx, row in clipped_gdf.iterrows(): geom = row.geometry if geom.is_empty: continue properties = {} for col in clipped_gdf.columns: if col != "geometry": val = row[col] if isinstance(val, (int, float, str, bool)): properties[col] = val else: properties[col] = str(val) if val is not None else "" features.append({"geometry": geom, "properties": properties}) tile_data = [{"name": layer_name, "features": features}] mvt_bytes = mapbox_vector_tile.encode( tile_data, quantize_bounds=(min_lon, min_lat, max_lon, max_lat), y_coord_down=True, extents=extent, ) return mvt_bytes