eomsangdonandClaude Opus 5.5
84f243cee1
- 서버: IF_0046_WMS 겹 A2SM_SANSATAI 를 받은 범위 경위도 상자로 한 장(2 m/px · 한 변 2,048) → 작업 좌표계 상자로 옮겨 그림 · 덮으면 저장본 · 키 없음 = 503 「키 없음」 · 받기 실패 · 그림 아님 = 502 - 키 = .env SAFEMAP_API_KEY 새 칸(값 비움) - 화면: 단추 하나(alert) · 묶음 셋째 · 처음 끔 · 바탕 캔버스 · 범례 1~5 등급(--draw-map-landslide-1~5) - 켬끔 취향 읽기를 LayerDefs 로 옮김(700줄 한도) - test_553_4: 겹 목록에 landslide 더함(새 겹 · 기대 목록 갱신) Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
132 lines
5.1 KiB
Python
132 lines
5.1 KiB
Python
"""D 노선분석 지도 산사태위험지도 겹(PLAN 584-1 · 요구 444).
|
||
|
||
- 출처 = 행안부 생활안전지도 산사태위험지도 WMS(`IF_0046_WMS` · 겹 `A2SM_SANSATAI` · 1~5 등급 ·
|
||
safemap objtId=125) · 키 = `.env` `SAFEMAP_API_KEY`(없으면 오류 · 값은 어디에도 안 냄)
|
||
- 받은 범위의 경위도 상자로 그림 한 장 받음(2 m/px · 한 변 2,048 한도) → 작업 좌표계 상자로
|
||
옮겨 그림(가까운 칸) → PNG data URL · 저장 = `processed/산사태위험지도_bounds.json`
|
||
(상자가 덮으면 다시 안 받음)
|
||
- 실패(키 없음 · 받기 실패 · 그림 아님) = 오류(파일 안 씀)
|
||
|
||
계산 자리 = ③ 서버 단독.
|
||
"""
|
||
|
||
from __future__ import annotations
|
||
|
||
import base64
|
||
import io
|
||
import json
|
||
import logging
|
||
import math
|
||
import os
|
||
import urllib.parse
|
||
import urllib.request
|
||
from pathlib import Path
|
||
from typing import Any
|
||
|
||
import numpy as np
|
||
|
||
from common_util.common_util_json import atomic_write_json
|
||
|
||
logger = logging.getLogger(__name__)
|
||
|
||
Box = tuple[float, float, float, float]
|
||
|
||
LANDSLIDE_URL = "http://www.safemap.go.kr/openapi2/IF_0046_WMS"
|
||
LANDSLIDE_LAYER = "A2SM_SANSATAI"
|
||
LANDSLIDE_KEY_ENV = "SAFEMAP_API_KEY"
|
||
LANDSLIDE_FILE = "산사태위험지도_bounds.json"
|
||
LANDSLIDE_VERSION = 1
|
||
PX_M = 2.0
|
||
MAX_PX = 2048
|
||
PNG_SIGNATURE = b"\x89PNG\r\n\x1a\n"
|
||
|
||
|
||
class MissingKeyError(RuntimeError):
|
||
"""`.env` 에 키 없음."""
|
||
|
||
|
||
def request_wms(bounds: dict[str, float], width: int, height: int) -> bytes:
|
||
"""WMS 그림 한 장(PNG 바이트) — 키 없음 = `MissingKeyError` · 실패 = `RuntimeError`."""
|
||
key = os.environ.get(LANDSLIDE_KEY_ENV, "").strip()
|
||
if not key:
|
||
raise MissingKeyError(f"산사태위험지도 키 없음 — .env {LANDSLIDE_KEY_ENV}")
|
||
params = {
|
||
"serviceKey": key,
|
||
"layers": LANDSLIDE_LAYER,
|
||
"styles": "",
|
||
"srs": "EPSG:4326",
|
||
"bbox": f"{bounds['min_lon']},{bounds['min_lat']},{bounds['max_lon']},{bounds['max_lat']}",
|
||
"format": "image/png",
|
||
"width": width,
|
||
"height": height,
|
||
"transparent": "TRUE",
|
||
}
|
||
url = LANDSLIDE_URL + "?" + urllib.parse.urlencode(params)
|
||
try:
|
||
with urllib.request.urlopen(
|
||
urllib.request.Request(url, headers={"User-Agent": "Mozilla/5.0"}), timeout=20
|
||
) as response:
|
||
body = response.read()
|
||
except OSError as error:
|
||
raise RuntimeError(f"산사태위험지도를 받지 못했습니다 — {type(error).__name__}") from None
|
||
if not body.startswith(PNG_SIGNATURE):
|
||
raise RuntimeError("산사태위험지도를 받지 못했습니다 — 그림이 아닌 응답")
|
||
return body
|
||
|
||
|
||
def warp_to_box(png: bytes, bounds: dict[str, float], box: Box, epsg: int) -> Any:
|
||
"""경위도 상자 그림 → 작업 좌표계 상자 그림(RGBA · 위 = 북 · 밖 = 투명 · 가까운 칸)."""
|
||
from PIL import Image
|
||
from pyproj import Transformer
|
||
|
||
source = np.asarray(Image.open(io.BytesIO(png)).convert("RGBA"))
|
||
rows, cols = source.shape[:2]
|
||
x0, y0, x1, y1 = box
|
||
width, height = cols, rows
|
||
xs = x0 + (np.arange(width) + 0.5) * (x1 - x0) / width
|
||
ys = y1 - (np.arange(height) + 0.5) * (y1 - y0) / height
|
||
gx, gy = np.meshgrid(xs, ys)
|
||
to_ll = Transformer.from_crs(f"EPSG:{epsg}", "EPSG:4326", always_xy=True)
|
||
lon, lat = to_ll.transform(gx, gy)
|
||
col = np.floor((lon - bounds["min_lon"]) / (bounds["max_lon"] - bounds["min_lon"]) * cols)
|
||
row = np.floor((bounds["max_lat"] - lat) / (bounds["max_lat"] - bounds["min_lat"]) * rows)
|
||
inside = (col >= 0) & (col < cols) & (row >= 0) & (row < rows)
|
||
out = np.zeros((height, width, 4), np.uint8)
|
||
out[inside] = source[row[inside].astype(int), col[inside].astype(int)]
|
||
return Image.fromarray(out, mode="RGBA")
|
||
|
||
|
||
def ensure_landslide(processed: Path, box: Box, epsg: int) -> dict[str, Any]:
|
||
"""산사태위험지도 그림 겹 — 저장 상자가 덮으면 그대로 · 아니면 받아 옮겨 그림."""
|
||
from D03_PlanAnalysis.D03_PlanAnalysis_Layers import _covers, _lonlat_bounds
|
||
|
||
path = processed / LANDSLIDE_FILE
|
||
try:
|
||
saved = json.loads(path.read_text(encoding="utf-8"))
|
||
except (OSError, ValueError):
|
||
saved = None
|
||
if (
|
||
isinstance(saved, dict)
|
||
and saved.get("version") == LANDSLIDE_VERSION
|
||
and _covers(saved.get("box_m"), box)
|
||
):
|
||
return saved
|
||
bounds = _lonlat_bounds(box, epsg)
|
||
scale = max((box[2] - box[0]) / PX_M, (box[3] - box[1]) / PX_M) / MAX_PX
|
||
step = PX_M * max(1.0, scale)
|
||
width = max(1, math.ceil((box[2] - box[0]) / step))
|
||
height = max(1, math.ceil((box[3] - box[1]) / step))
|
||
image = warp_to_box(request_wms(bounds, width, height), bounds, box, epsg)
|
||
buffer = io.BytesIO()
|
||
image.save(buffer, format="PNG", optimize=True)
|
||
value = {
|
||
"version": LANDSLIDE_VERSION,
|
||
"box_m": [float(v) for v in box],
|
||
"image": "data:image/png;base64," + base64.b64encode(buffer.getvalue()).decode("ascii"),
|
||
"bands": [],
|
||
}
|
||
processed.mkdir(parents=True, exist_ok=True)
|
||
atomic_write_json(path, value)
|
||
logger.info("D 산사태위험지도 받기: %d × %d px", width, height)
|
||
return value
|