212 lines
8.7 KiB
Python
212 lines
8.7 KiB
Python
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")
|