"""국가 GIS 벡터 데이터 다운로드 엔진.""" from __future__ import annotations import json import logging import urllib.parse import urllib.request from pathlib import Path from pyproj import Transformer # 설정 불러오기 try: from config import config_system VWORLD_API_KEY = getattr( config_system, "VWORLD_API_KEY", "3DBD7306-7DBD-38BB-B292-267C5ED7AC6B" ) PROJECT_ROOT = getattr(config_system, "PROJECT_ROOT", Path(__file__).resolve().parent.parent) except ImportError: VWORLD_API_KEY = "3DBD7306-7DBD-38BB-B292-267C5ED7AC6B" PROJECT_ROOT = Path(__file__).resolve().parent.parent from B04_PreProcess.B04_PreProcess_Engine_VWorld import get_epsg_from_prj logger = logging.getLogger(__name__) def request_vworld_wfs(typename: str, filename: str, bounds_wgs84: dict, output_dir: Path) -> None: """브이월드 WFS API를 사용하여 지정된 Bounds 영역 내의 공간 벡터 데이터를 GeoJSON으로 다운로드합니다. """ # 4326(위경도) bbox 파라미터 구성 (BBOX=minx,miny,maxx,maxy) bbox_str = ( f"{bounds_wgs84['min_lon']},{bounds_wgs84['min_lat']}," f"{bounds_wgs84['max_lon']},{bounds_wgs84['max_lat']}" ) # WFS 파라미터 조립 params = { "key": VWORLD_API_KEY, "domain": "localhost", "service": "WFS", "version": "1.1.0", "request": "GetFeature", "typename": typename, "bbox": bbox_str, "output": "application/json", # GeoJSON 포맷 요청 "srsName": "EPSG:4326", } url = "http://api.vworld.kr/req/wfs?" + urllib.parse.urlencode(params) headers = {"User-Agent": "Mozilla/5.0"} req = urllib.request.Request(url, headers=headers) try: with urllib.request.urlopen(req, timeout=15) as res: content = res.read() data = json.loads(content.decode("utf-8")) output_path = output_dir / filename output_path.write_text(json.dumps(data, indent=2, ensure_ascii=False), encoding="utf-8") except Exception: pass def read_bounds_wgs84_from_meta(meta_dir: Path) -> dict | None: """VWorld 메타 JSON에서 여유폭 0.010도를 더한 위경도 경계를 읽어옵니다.""" meta_path = meta_dir / "vworld_white_meta.json" if not meta_path.exists(): meta_path = meta_dir / "vworld_satellite_meta.json" if not meta_path.exists(): return None try: with open(meta_path, "r", encoding="utf-8") as f: meta_data = json.load(f) return { "min_lon": meta_data["lon_min"] - 0.010, "max_lon": meta_data["lon_max"] + 0.010, "min_lat": meta_data["lat_min"] - 0.010, "max_lat": meta_data["lat_max"] + 0.010, } except Exception as exc: logger.warning("VWorld 메타 경계 읽기 실패: %s (%s)", meta_path, exc) return None def crop_national_contours(bounds_wgs84: dict, output_dir: Path) -> bool: """국가 등고선 gpkg에서 지정 위경도 영역을 잘라 등고선_bounds.geojson을 생성합니다.""" gpkg_path = PROJECT_ROOT / "resources" / "data_global_contours" / "national_contours.gpkg" if not gpkg_path.exists(): logger.warning("국가 등고선 gpkg가 없습니다: %s", gpkg_path) return False try: import geopandas as gpd from shapely.geometry import box # gpkg는 EPSG:5179(KGD2002 / Unified CS) 좌표계이므로 bbox도 같은 계로 변환한다. t_to_5179 = Transformer.from_crs("EPSG:4326", "EPSG:5179", always_xy=True) min_x_5179, min_y_5179 = t_to_5179.transform( bounds_wgs84["min_lon"], bounds_wgs84["min_lat"] ) max_x_5179, max_y_5179 = t_to_5179.transform( bounds_wgs84["max_lon"], bounds_wgs84["max_lat"] ) bbox = box(min_x_5179, min_y_5179, max_x_5179, max_y_5179) gdf = gpd.read_file(gpkg_path, bbox=bbox) if gdf.empty: logger.warning("국가 등고선 크롭 결과가 비어 있습니다: bbox=%s", bounds_wgs84) return False # bbox 필터는 교차한 도엽 전체 라인을 반환하므로 실제 경계로 잘라 전송량을 줄인다. gdf = gdf.clip(bbox) gdf = gdf[~gdf.geometry.is_empty & gdf.geometry.notna()] if gdf.empty: logger.warning("국가 등고선 클리핑 결과가 비어 있습니다: bbox=%s", bounds_wgs84) return False # 라벨·구분에 필요한 속성만 유지 (Timestamp 등 나머지는 JSON 직렬화 불가·전송량 낭비) keep = [c for c in ("CTRLN_SE", "TPGRPH_SE", "CTRLN_HG") if c in gdf.columns] gdf = gdf[keep + ["geometry"]] output_dir.mkdir(parents=True, exist_ok=True) gdf_wgs84 = gdf.to_crs("EPSG:4326") geojson_out = output_dir / "등고선_bounds.geojson" gdf_wgs84.to_file(geojson_out, driver="GeoJSON") logger.info("국가 등고선 크롭 완료: %d개 라인 -> %s", len(gdf_wgs84), geojson_out) return True except Exception as exc: logger.warning("국가 등고선 크롭 실패: %s", exc) return False def download_all_gis_vectors(prj_path: Path, bounds_meter: dict, output_dir: Path) -> None: """지형의 로컬 미터단위 bounds 정보를 위경도로 변환 후 5대 국가 GIS 데이터를 다운로드합니다.""" output_dir.mkdir(parents=True, exist_ok=True) # 1. vworld_white_meta.json이 생성되어 있다면 직접 위경도 경계 획득하여 pyproj 축왜곡 방지 bounds_wgs84 = read_bounds_wgs84_from_meta(prj_path.parent) if bounds_wgs84 is None: # 폴백용 pyproj 변환 src_epsg = "EPSG:5186" if prj_path.exists(): src_epsg = get_epsg_from_prj(prj_path.read_text(encoding="utf-8", errors="ignore")) transformer = Transformer.from_crs(src_epsg, "EPSG:4326", always_xy=True) x_min, x_max = bounds_meter["x"][0], bounds_meter["x"][1] y_min, y_max = bounds_meter["y"][0], bounds_meter["y"][1] lon_min, lat_min = transformer.transform(x_min, y_min) lon_max, lat_max = transformer.transform(x_max, y_max) bounds_wgs84 = { "min_lon": min(lon_min, lon_max) - 0.010, "max_lon": max(lon_min, lon_max) + 0.010, "min_lat": min(lat_min, lat_max) - 0.010, "max_lat": max(lat_min, lat_max) + 0.010, } # 5대 레이어 다운로드 request_vworld_wfs( "lt_c_landinfobasemap", "연속지적도_bounds.geojson", bounds_wgs84, output_dir ) request_vworld_wfs("lt_c_uq111", "용도지역도_bounds.geojson", bounds_wgs84, output_dir) request_vworld_wfs("lt_c_adsigg", "행정구역_시군구_bounds.geojson", bounds_wgs84, output_dir) request_vworld_wfs("lt_c_adri", "행정구역_읍면동_bounds.geojson", bounds_wgs84, output_dir) # 수계망(도엽 하천중심선으로 대체)·산사태(시군구 예보뿐, 소구역 미제공)는 # 다운로드하지 않음 (2026-07-26 사용자 지시) # 6. gpkg 등고선(national_contours.gpkg) 데이터에서 # 기준 영역 크롭하여 등고선_bounds.geojson 생성 crop_national_contours(bounds_wgs84, output_dir)