From f847ea700fd110d8d562848930d0969cdafb20d6 Mon Sep 17 00:00:00 2001 From: umsangdon Date: Fri, 17 Jul 2026 19:29:43 +0900 Subject: [PATCH] 260717_5 --- B04_wf1_Surface/B04_wf1_Surface_Router_GIS.py | 211 ++++++++++++++++++ 1 file changed, 211 insertions(+) create mode 100644 B04_wf1_Surface/B04_wf1_Surface_Router_GIS.py diff --git a/B04_wf1_Surface/B04_wf1_Surface_Router_GIS.py b/B04_wf1_Surface/B04_wf1_Surface_Router_GIS.py new file mode 100644 index 00000000..642a1594 --- /dev/null +++ b/B04_wf1_Surface/B04_wf1_Surface_Router_GIS.py @@ -0,0 +1,211 @@ +import json +import logging +from pathlib import Path +from typing import Any +from uuid import UUID + +from fastapi import APIRouter, HTTPException, Response +from fastapi.responses import FileResponse, JSONResponse + +from B03_FileInput.B03_FileInput_Repository import get_project_storage_relative_path +from common_util.common_util_storage import resolve_stored_project_path +from config.config_db import get_db_pool + +logger = logging.getLogger(__name__) +router = APIRouter(prefix="/api/projects", tags=["B04 Surface GIS"]) +tiles_router = APIRouter(tags=["B04 MVT Tiles"]) + + +# VWorld 메타 API +@router.get("/{project_id}/vworld-meta", response_model=None) +async def get_vworld_meta( + project_id: UUID, layer_name: str = "satellite" +) -> dict[str, Any] | JSONResponse: + """VWorld 위성 맵 이미지 매핑 좌표 메타데이터를 반환합니다.""" + pool = get_db_pool() + try: + async with pool.acquire() as connection: + stored_path = await get_project_storage_relative_path(connection, project_id) + project_root = Path(resolve_stored_project_path(stored_path)) + target_dir = project_root / "B04_wf1_Surface" / "processed" + + target_layer = "white" if layer_name.lower() in ["gray", "white"] else layer_name.lower() + meta_name = f"vworld_{target_layer}_meta.json" + meta_path = target_dir / meta_name + + if not meta_path.exists() and target_layer == "satellite": + meta_path = target_dir / "vworld_meta.json" + + if not meta_path.exists(): + return JSONResponse( + status_code=404, + content={ + "status": "error", + "message": f"VWorld {layer_name} 메타데이터를 찾을 수 없습니다.", + }, + ) + return json.loads(meta_path.read_text(encoding="utf-8")) + except Exception as exc: + return JSONResponse(status_code=500, content={"status": "error", "message": str(exc)}) + + +# VWorld 맵 API +@router.get("/{project_id}/vworld-map", response_model=None) +async def get_vworld_map( + project_id: UUID, layer_name: str = "satellite" +) -> FileResponse | JSONResponse: + """배경 지도 레이어 PNG 이미지를 반환합니다.""" + pool = get_db_pool() + try: + async with pool.acquire() as connection: + stored_path = await get_project_storage_relative_path(connection, project_id) + project_root = Path(resolve_stored_project_path(stored_path)) + target_dir = project_root / "B04_wf1_Surface" / "processed" + + target_layer = layer_name.lower() + if target_layer in ["gray", "white"]: + target_layer = "white" + + map_name = f"vworld_{target_layer}.png" + map_path = target_dir / map_name + + if not map_path.exists() and target_layer == "satellite": + map_path = target_dir / "vworld_map.png" + + if not map_path.exists(): + return JSONResponse( + status_code=404, + content={ + "status": "error", + "message": f"VWorld {layer_name} 지도가 존재하지 않습니다.", + }, + ) + return FileResponse(map_path, media_type="image/png") + except Exception as exc: + return JSONResponse(status_code=500, content={"status": "error", "message": str(exc)}) + + +# GeoJSON 조회 API +@router.get("/{project_id}/geojson", response_model=None) +async def get_project_geojson(project_id: UUID, layer: str) -> dict[str, Any] | JSONResponse: + """저장된 프로젝트의 특정 GeoJSON 레이어 데이터를 반환합니다.""" + pool = get_db_pool() + try: + async with pool.acquire() as connection: + stored_path = await get_project_storage_relative_path(connection, project_id) + project_root = Path(resolve_stored_project_path(stored_path)) + target_dir = project_root / "B04_wf1_Surface" / "processed" + + layer_mapping = { + "지적도": "연속지적도_bounds.geojson", + "용도지역": "용도지역도_bounds.geojson", + "행정구역_시군구": "행정구역_시군구_bounds.geojson", + "행정구역_읍면동": "행정구역_읍면동_bounds.geojson", + "수계망": "수계망_물줄기_bounds.geojson", + "등고선": "등고선_bounds.geojson", + "산사태": "산사태위험등급_bounds.geojson", + } + + filename = layer_mapping.get(layer) + if not filename: + return JSONResponse( + status_code=400, + content={"status": "error", "message": "유효하지 않은 레이어명입니다."}, + ) + + filepath = target_dir / filename + if not filepath.exists(): + return JSONResponse( + status_code=404, + content={ + "status": "error", + "message": f"요청한 레이어({layer}) 파일이 존재하지 않습니다.", + }, + ) + + if layer == "등고선": + simplified_filepath = target_dir / "등고선_bounds_simplified.geojson" + if simplified_filepath.exists(): + try: + return json.loads(simplified_filepath.read_text(encoding="utf-8")) + except Exception: + pass + + try: + import geopandas as gpd + + gdf = gpd.read_file(filepath) + gdf["geometry"] = gdf["geometry"].simplify( + tolerance=0.00003, preserve_topology=True + ) + simplified_filepath.write_text(gdf.to_json(), encoding="utf-8") + return json.loads(simplified_filepath.read_text(encoding="utf-8")) + except Exception as e: + logger.warning("등고선 단순화 처리 실패 (원본 전송): %s", e) + + return json.loads(filepath.read_text(encoding="utf-8")) + except Exception as exc: + return JSONResponse(status_code=500, content={"status": "error", "message": str(exc)}) + + +@tiles_router.get("/tiles/{project_id}/{layer}/{z}/{x}/{y}.pbf", response_model=None) +async def get_vector_tile(project_id: UUID, layer: str, z: int, x: int, y: int) -> Response: + """프로젝트의 특정 레이어에 대한 정밀 벡터 타일(MVT) 조각을 동적으로 렌더링하여 반환합니다.""" + pool = get_db_pool() + try: + async with pool.acquire() as connection: + stored_path = await get_project_storage_relative_path(connection, project_id) + project_root = Path(resolve_stored_project_path(stored_path)) + target_dir = project_root / "B04_wf1_Surface" / "processed" + + layer_mapping = { + "지적도": "연속지적도_bounds.geojson", + "용도지역": "용도지역도_bounds.geojson", + "행정구역_시군구": "행정구역_시군구_bounds.geojson", + "행정구역_읍면동": "행정구역_읍면동_bounds.geojson", + "수계망": "수계망_물줄기_bounds.geojson", + "등고선": "등고선_bounds.geojson", + "산사태": "산사태위험등급_bounds.geojson", + "임도노선": "임도노선.geojson", + } + + filename = layer_mapping.get(layer) + if not filename: + raise HTTPException(status_code=400, detail="유효하지 않은 레이어명입니다.") + + filepath = target_dir / filename + + if layer == "임도노선" and not filepath.exists(): + from config.config_system import PROJECT_ROOT + + road_shp_candidates = list(PROJECT_ROOT.glob("samples/**/*_Polyline.shp")) + if road_shp_candidates: + try: + import geopandas as gpd + + gdf = gpd.read_file(road_shp_candidates[0]) + gdf = gdf.to_crs(epsg=4326) + filepath.write_text(gdf.to_json(), encoding="utf-8") + except Exception: + pass + + if not filepath.exists(): + import mapbox_vector_tile + + empty_tile = mapbox_vector_tile.encode([]) if mapbox_vector_tile else b"" + return Response(content=empty_tile, media_type="application/x-protobuf") + + cache_key = f"{project_id}_{layer}" + from B04_wf1_Surface.B04_wf1_Surface_Engine_MvtHelper import generate_mvt_tile + + mvt_bytes = generate_mvt_tile(filepath, cache_key, z, x, y, layer_name=layer) + return Response( + content=mvt_bytes, + media_type="application/x-protobuf", + headers={"Content-Encoding": "identity"}, + ) + except Exception: + import mapbox_vector_tile + + empty_tile = mapbox_vector_tile.encode([]) if mapbox_vector_tile else b"" + return Response(content=empty_tile, media_type="application/x-protobuf")