From 84f243cee13cfd0586f253ea6a97dafd4a640534 Mon Sep 17 00:00:00 2001 From: umsangdon Date: Thu, 29 Oct 2026 16:59:11 +0900 Subject: [PATCH] =?UTF-8?q?feat(D03=20=EA=B2=B9):=20=EB=85=B8=EC=84=A0?= =?UTF-8?q?=EB=B6=84=EC=84=9D=20=EC=A7=80=EB=8F=84=20=EC=82=B0=EC=82=AC?= =?UTF-8?q?=ED=83=9C=EC=9C=84=ED=97=98=EC=A7=80=EB=8F=84=20=EA=B2=B9=20?= =?UTF-8?q?=E2=80=94=20=ED=96=89=EC=95=88=EB=B6=80=20=EC=83=9D=ED=99=9C?= =?UTF-8?q?=EC=95=88=EC=A0=84=EC=A7=80=EB=8F=84=20WMS=20=C2=B7=20=ED=82=A4?= =?UTF-8?q?=20=EC=97=86=EC=9C=BC=EB=A9=B4=20=EA=B7=B8=20=EA=B2=B9=EB=A7=8C?= =?UTF-8?q?=20=EA=BA=BC=EC=A7=90=20(PLAN=20584-1=20=C2=B7=20=EC=9A=94?= =?UTF-8?q?=EA=B5=AC=20444)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - 서버: 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 --- .env | 2 + D03_PlanAnalysis/D03_PlanAnalysis_Api.ts | 2 +- .../D03_PlanAnalysis_Landslide.py | 131 ++++++++++++++++++ D03_PlanAnalysis/D03_PlanAnalysis_Router.py | 16 ++- .../D03_PlanAnalysis_UI_LayerDefs.ts | 32 ++++- .../D03_PlanAnalysis_UI_Layers.ts | 38 +++-- resources/tester/test_553_4_de_parts.py | 1 + resources/tester/test_584_1_plan_geo_lines.py | 60 +++++++- ui_template/ui_template_theme.css | 5 + 9 files changed, 262 insertions(+), 25 deletions(-) create mode 100644 D03_PlanAnalysis/D03_PlanAnalysis_Landslide.py diff --git a/.env b/.env index e8edab1cb..509a5e6e5 100644 --- a/.env +++ b/.env @@ -88,3 +88,5 @@ PROJECT_DELETE_HARD_ENABLED=True AUTH_OTP_DISABLED=True # 대시보드 지식 챗봇(B01_Dashboard_Knowledge_Chat) NVIDIA 키 NVIDIA_API_KEY=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 +# 노선분석 산사태위험지도(행안부 생활안전지도 WMS IF_0046 · D03_PlanAnalysis_Landslide) 키 — 비우면 겹 꺼짐 +SAFEMAP_API_KEY= diff --git a/D03_PlanAnalysis/D03_PlanAnalysis_Api.ts b/D03_PlanAnalysis/D03_PlanAnalysis_Api.ts index baff2fb76..a64b0b8e0 100644 --- a/D03_PlanAnalysis/D03_PlanAnalysis_Api.ts +++ b/D03_PlanAnalysis/D03_PlanAnalysis_Api.ts @@ -151,7 +151,7 @@ export const fetchPlanSheetLayer = (projectId: string, key: string): Promise => request(`${base(projectId)}/layers/raster?kind=${kind}`); /** 보호 · 규제 구역(581-1) — 지물 `kind` · `name` · `area_m2` + 종류별 개수 · 넓이(실패 = `failed`). */ diff --git a/D03_PlanAnalysis/D03_PlanAnalysis_Landslide.py b/D03_PlanAnalysis/D03_PlanAnalysis_Landslide.py new file mode 100644 index 000000000..b9c44cd6a --- /dev/null +++ b/D03_PlanAnalysis/D03_PlanAnalysis_Landslide.py @@ -0,0 +1,131 @@ +"""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 diff --git a/D03_PlanAnalysis/D03_PlanAnalysis_Router.py b/D03_PlanAnalysis/D03_PlanAnalysis_Router.py index 00e016170..1b0cb883f 100644 --- a/D03_PlanAnalysis/D03_PlanAnalysis_Router.py +++ b/D03_PlanAnalysis/D03_PlanAnalysis_Router.py @@ -3,7 +3,7 @@ - `GET /api/projects/{id}/plan/terrain/extent` — 받기 전 범위(상자 · 도엽 · 칸 수 · 받은 범위) - `POST …/plan/terrain` — 사업 전체 지형 한 벌 받기(뒤에서 · 누를 때만) · 끝나면 `surface_models` 행 - `GET …/plan/progress` — 일의 진행률 -- `GET …/plan/layers/sheet?key=` · `…/cadastral` · `…/raster?kind=slope|hillshade` · +- `GET …/plan/layers/sheet?key=` · `…/cadastral` · `…/raster?kind=slope|hillshade|landslide` · `…/zones` · `…/soils` · `…/bands` · `…/lines?kind=` · `…/zero-order` — 지도 겹 (PLAN 569-1 · 581-1 · 581-2 · 584-1) @@ -29,6 +29,7 @@ from common_util.common_util_http_cache import cached_file_response from common_util.common_util_storage import resolve_stored_project_path from config.config_db import get_db_pool from D02_PlanBase import D02_PlanBase_Router as base +from D03_PlanAnalysis import D03_PlanAnalysis_Landslide as landslide from D03_PlanAnalysis import D03_PlanAnalysis_Layers as layers from D03_PlanAnalysis import D03_PlanAnalysis_Terrain as terrain from D03_PlanAnalysis import D03_PlanAnalysis_ZeroOrder as zero_order @@ -200,7 +201,18 @@ async def plan_layer_cadastral(project_id: UUID, request: Request) -> Response: @router.get("/{project_id}/plan/layers/raster", response_model=None) async def plan_layer_raster(project_id: UUID, kind: str = "slope") -> dict[str, Any]: - """급경사 단계 그림(15° · 35° · 40°) · 음영기복(`hillshade`) — 격자 지문 같으면 저장본.""" + """급경사 단계 그림(15° · 35° · 40°) · 음영기복(`hillshade`) — 격자 지문 같으면 저장본 · + 산사태위험지도(`landslide` · 행안부 WMS) — 받은 범위를 덮으면 저장본.""" + if kind == "landslide": + root, epsg, box = await _layer_root(project_id) + processed = root / terrain.STAGE / "processed" + try: + value = await asyncio.to_thread(landslide.ensure_landslide, processed, box, epsg) + except landslide.MissingKeyError as error: + raise HTTPException(status_code=503, detail=str(error)) from error + except RuntimeError as error: + raise HTTPException(status_code=502, detail=str(error)) from error + return {"status": "success", "kind": kind, **value} if kind not in layers.RASTER_KINDS: raise HTTPException(status_code=400, detail=f"없는 그림 겹: {kind}") root, epsg, _ = await _layer_root(project_id) diff --git a/D03_PlanAnalysis/D03_PlanAnalysis_UI_LayerDefs.ts b/D03_PlanAnalysis/D03_PlanAnalysis_UI_LayerDefs.ts index b6709226e..30204e25d 100644 --- a/D03_PlanAnalysis/D03_PlanAnalysis_UI_LayerDefs.ts +++ b/D03_PlanAnalysis/D03_PlanAnalysis_UI_LayerDefs.ts @@ -1,11 +1,13 @@ /* ============================================================================= * D03_PlanAnalysis_UI_LayerDefs.ts * D03 바탕 겹 정의(PLAN 569-2 · 581-1 · 581-2 · 584-1) — 겹 단추 묶음 · 처음 값 · 보호구역 · 토양형 색 · - * 선 겹(단층 · 지질구조선 · 0차곡) 꼴 · 면 겹 채우기 · 선 긋기 · 그림 겹 놓기. + * 산사태위험지도 범례 · 선 겹(단층 · 지질구조선 · 0차곡) 꼴 · 켬끔 취향 읽기 · 면 겹 채우기 · 선 긋기 · 그림 겹 놓기. * 부르는 곳 = `D03_PlanAnalysis_UI_Layers.ts`(받기 · 그리기 차례 · 범례). * ========================================================================== */ import type { IconName } from "@ui/ui_template_icons"; +import type { LegendItem } from "@ui/ui_template_flow_legend"; +import { readByKey } from "../A00_Common/b_page_state"; import { drawPreparedLayer, normalizedToScreen, @@ -27,6 +29,7 @@ export type PlanLayerKey = | "cadastral" | "zones" | "soils" + | "landslide" | "faults" | "lineaments" | "zeroOrder" @@ -74,6 +77,12 @@ export const GROUPS: readonly (readonly LayerDef[])[] = [ title: "산림입지도(토양형 · 산림청 산림입지토양도 · 참고용)", on: false, }, + { + key: "landslide", + icon: "alert", + title: "산사태위험지도(1~5 등급 · 행안부 생활안전지도 · 참고용)", + on: false, + }, { key: "faults", icon: "bolt", title: "단층(브이월드 지질 · 참고용)", on: false }, { key: "lineaments", @@ -109,6 +118,11 @@ export const ZONE_COLORS: Record = { wildlife: "--draw-map-zone-wildlife", greenbelt: "--draw-map-zone-greenbelt", }; +/** 산사태위험지도 범례(1 등급 = 가장 위험). */ +export const LANDSLIDE_LEGEND: LegendItem[] = [1, 2, 3, 4, 5].map((grade) => ({ + label: `${grade}등급`, + color: `var(--draw-map-landslide-${grade})`, +})); /** 선 겹 — 이름 · 그림 꼴(서버 `LINE_KINDS` 열쇠 = `kind` · 없으면 0차곡). 새 선 겹 = 여기 한 줄. */ export type LineLayerKey = "faults" | "lineaments" | "zeroOrder"; export const LINE_LAYERS: Record< @@ -213,3 +227,19 @@ export function drawRaster( const [right, bottom] = map.toScreen([x1, y0]); context.drawImage(raster.image, left, top, right - left, bottom - top); } + +/** 켬끔 취향(`key`) — 없거나 깨지면 처음 값. */ +export function readPref(key: string): Record { + const on = Object.fromEntries(LAYERS.map((def) => [def.key, def.on])) as Record< + PlanLayerKey, + boolean + >; + try { + const saved = JSON.parse(readByKey(key) ?? "{}") as Partial>; + for (const def of LAYERS) + if (typeof saved[def.key] === "boolean") on[def.key] = saved[def.key] as boolean; + } catch { + // 깨진 값 = 처음 값 + } + return on; +} diff --git a/D03_PlanAnalysis/D03_PlanAnalysis_UI_Layers.ts b/D03_PlanAnalysis/D03_PlanAnalysis_UI_Layers.ts index cb65ef2ea..5ecb8639c 100644 --- a/D03_PlanAnalysis/D03_PlanAnalysis_UI_Layers.ts +++ b/D03_PlanAnalysis/D03_PlanAnalysis_UI_Layers.ts @@ -5,11 +5,11 @@ * * 겹 = 바탕(위성 · 음영기복 · 도엽 등고선(계곡선 25 m 굵게 · 수치) · 급경사 색) · * 물 · 길(세류선 · 하천 · 물 면(실폭하천 · 저수지) · 기존 도로 · 기존 임도 짐작(소로 · 미분류)) · - * 구역(지적도 · 보호 · 규제 구역 여덟 종 · 산림입지도(토양형) · 선 겹(단층 · 지질구조선 · 0차곡) · + * 구역(지적도 · 보호 · 규제 구역 여덟 종 · 산림입지도(토양형) · 산사태위험지도 · 선 겹(단층 · 지질구조선 · 0차곡) · * 계류 100 m · 도로 300 m 띠 · 건물 · 지명). * 처음 켬 = 위성 · 등고선 · 세류 · 고른 값 = 화면 취향(`aislo:pref:d03.layers`) 기억. * 받기 = 켤 때만(한 번 받으면 다시 안 받음) · 지형을 안 받았으면 단추를 막음 · 받기 실패 = 그 겹만 꺼짐 + 알림. - * 그리기 차례(아래 → 위) = 위성(`scene.image`) → 음영기복 → 급경사 → 산림입지도 → 보호구역 → 띠 → 지적 → 물 면 → + * 그리기 차례(아래 → 위) = 위성(`scene.image`) → 음영기복 → 급경사 → 산사태위험지도 → 산림입지도 → 보호구역 → 띠 → 지적 → 물 면 → * 건물 → 등고선 → 도로 → 임도 짐작 → 세류 → 선 겹 → 지명 → 그은 선. * 겹 모두 = 바탕 캔버스(`base` · 582) · 등고선 = 줌 단계 색인 · 글 폭 기억. * 범례 = 켠 겹의 받은 범위 안 수 · 넓이(왼쪽 아래). @@ -21,7 +21,7 @@ import { createIconButton } from "@ui/ui_template_button"; import { createMapTools } from "@ui/ui_template_map_tools"; import { themeColor } from "@ui/ui_template_palette"; import { fetchCachedSheetLayer } from "../A00_Common/b_asset_cache"; -import { readByKey, writeByKey } from "../A00_Common/b_page_state"; +import { writeByKey } from "../A00_Common/b_page_state"; import { fetchVWorldMeta, getVWorldMapUrl, @@ -56,6 +56,7 @@ import { import { FOREST_ROAD_KINDS, GROUPS, + LANDSLIDE_LEGEND, LAYERS, LINE_KEYS, LINE_LAYERS, @@ -66,6 +67,7 @@ import { fillLayer, hectare, loadImage, + readPref, stroke, type LineLayerKey, type PlanLayerKey, @@ -91,6 +93,7 @@ type Prepared = Record & { road: { all: PreparedLayer; small: PreparedLayer }; slope: { raster: PlanRaster; image: HTMLImageElement }; hillshade: { raster: PlanRaster; image: HTMLImageElement }; + landslide: { raster: PlanRaster; image: HTMLImageElement }; water: PreparedLayer; zones: { kinds: Map; summary: PlanZones["summary"] }; /** 토양형(코드 차례) → 지물 · 색 값. */ @@ -120,21 +123,6 @@ export interface PlanLayers { }; } -function readPref(): Record { - const on = Object.fromEntries(LAYERS.map((def) => [def.key, def.on])) as Record< - PlanLayerKey, - boolean - >; - try { - const saved = JSON.parse(readByKey(PREF_KEY) ?? "{}") as Partial>; - for (const def of LAYERS) - if (typeof saved[def.key] === "boolean") on[def.key] = saved[def.key] as boolean; - } catch { - // 깨진 값 = 처음 값 - } - return on; -} - export function createPlanLayers(opts: { projectId: string; map: PlanMap; @@ -142,7 +130,7 @@ export function createPlanLayers(opts: { points: () => Vertex[]; }): PlanLayers { const { projectId, map } = opts; - const on = readPref(); + const on = readPref(PREF_KEY); const data: Partial = {}; const pending = new Map>(); let meta: VWorldMeta | null = null; @@ -186,6 +174,7 @@ export function createPlanLayers(opts: { color: band.color, })), ); + if (on.landslide && data.landslide) rows.push(LANDSLIDE_LEGEND); if (on.zones && data.zones) { const items = Object.entries(data.zones.summary) .filter(([, kind]) => kind.count || kind.failed) @@ -366,7 +355,7 @@ export function createPlanLayers(opts: { pick(collection, (props) => FOREST_ROAD_KINDS.has(String(props["도로구분"]))), ), }; - } else if (key === "slope" || key === "hillshade") { + } else if (key === "slope" || key === "hillshade" || key === "landslide") { const raster = await fetchPlanRaster(projectId, key); data[key] = { raster, image: await loadImage(raster.image) }; } else if (key === "water") { @@ -461,6 +450,14 @@ export function createPlanLayers(opts: { drawRaster(context, map, data.slope); }, }, + { + base: true, + draw: (context, map) => { + if (!on.landslide || !data.landslide) return; + context.globalAlpha = 0.6; + drawRaster(context, map, data.landslide); + }, + }, { base: true, draw: (context, map) => { @@ -659,6 +656,7 @@ export function createPlanLayers(opts: { band_road_m2: data.bands.info.road.area_m2, }), ...(data.hillshade && { hillshade_px: data.hillshade.image.naturalWidth }), + ...(data.landslide && { landslide_px: data.landslide.image.naturalWidth }), ...Object.fromEntries( LINE_KEYS.flatMap((key) => { const got = data[key]; diff --git a/resources/tester/test_553_4_de_parts.py b/resources/tester/test_553_4_de_parts.py index 5cb67c81a..c5bc0a1f7 100644 --- a/resources/tester/test_553_4_de_parts.py +++ b/resources/tester/test_553_4_de_parts.py @@ -67,6 +67,7 @@ def test_d03_parts(): ("cadastral", "false"), ("zones", "false"), ("soils", "false"), + ("landslide", "false"), ("faults", "false"), ("lineaments", "false"), ("zeroOrder", "false"), diff --git a/resources/tester/test_584_1_plan_geo_lines.py b/resources/tester/test_584_1_plan_geo_lines.py index 7a742d9e6..4f855c427 100644 --- a/resources/tester/test_584_1_plan_geo_lines.py +++ b/resources/tester/test_584_1_plan_geo_lines.py @@ -1,9 +1,12 @@ """D 노선분석 GIS 겹 더함 2(PLAN 584-1) — 단층 · 지질구조선 잘라 가닥 · 길이 · 실패 = 오류 · -덮으면 다시 안 받음 · 0차곡 흐름누적(웅덩이 · 평지 넘김) · 임계 · 선 뽑기 · 저장본 · 화면 겹 정의. +덮으면 다시 안 받음 · 산사태위험지도(키 없음 · 그림 아님 = 오류 · 옮겨 그림 · 덮으면 저장본) · +0차곡 흐름누적(웅덩이 · 평지 넘김) · 임계 · 선 뽑기 · 저장본 · 화면 겹 정의. ./venv/Scripts/python.exe -m pytest resources/tester/test_584_1_plan_geo_lines.py -q """ +import base64 +import io import json import math import re @@ -13,6 +16,7 @@ import numpy as np import pytest from common_util.common_util_crs import tm_to_lonlat +from D03_PlanAnalysis import D03_PlanAnalysis_Landslide as landslide from D03_PlanAnalysis import D03_PlanAnalysis_Layers as layers from D03_PlanAnalysis import D03_PlanAnalysis_ZeroOrder as zo @@ -81,6 +85,60 @@ def test_line_kinds_vworld_names() -> None: assert layers.LINE_KINDS["lineaments"][1] == "lt_l_gimslinea" +# ── 산사태위험지도 ── + + +def _png(width: int, height: int, rgba: tuple[int, int, int, int]) -> bytes: + from PIL import Image + + buffer = io.BytesIO() + Image.new("RGBA", (width, height), rgba).save(buffer, format="PNG") + return buffer.getvalue() + + +def test_landslide_no_key(tmp_path, monkeypatch) -> None: + monkeypatch.setenv(landslide.LANDSLIDE_KEY_ENV, " ") + with pytest.raises(landslide.MissingKeyError, match="키 없음"): + landslide.ensure_landslide(tmp_path, BOX, EPSG) + assert not list(tmp_path.iterdir()) + + +def test_landslide_not_image(tmp_path, monkeypatch) -> None: + class Reply(io.BytesIO): + def __enter__(self): + return self + + def __exit__(self, *exc) -> None: + pass + + monkeypatch.setenv(landslide.LANDSLIDE_KEY_ENV, "k") + monkeypatch.setattr(landslide.urllib.request, "urlopen", lambda *a, **k: Reply(b"")) + with pytest.raises(RuntimeError, match="그림이 아닌"): + landslide.ensure_landslide(tmp_path, BOX, EPSG) + assert not list(tmp_path.iterdir()) + + +def test_landslide_warp_and_saved(tmp_path, monkeypatch) -> None: + from PIL import Image + + calls = [] + + def fake(bounds, width, height): + calls.append((width, height)) + return _png(width, height, (220, 38, 38, 200)) + + monkeypatch.setattr(landslide, "request_wms", fake) + value = landslide.ensure_landslide(tmp_path, BOX, EPSG) + assert calls == [(1000, 500)] # 2 m/px + assert value["box_m"] == list(BOX) + raw = base64.b64decode(value["image"].split(",", 1)[1]) + image = np.asarray(Image.open(io.BytesIO(raw))) + assert image.shape == (500, 1000, 4) + assert tuple(image[250, 500]) == (220, 38, 38, 200) # 가운데 = 받은 색 + inner = (230100.0, 450100.0, 231000.0, 450900.0) + assert landslide.ensure_landslide(tmp_path, inner, EPSG) == value and len(calls) == 1 + + # ── 0차곡 ── diff --git a/ui_template/ui_template_theme.css b/ui_template/ui_template_theme.css index 21b83afa4..342e1bc6f 100644 --- a/ui_template/ui_template_theme.css +++ b/ui_template/ui_template_theme.css @@ -575,6 +575,11 @@ html.is-bm-route .m01lab-src__body * { --draw-map-soil-4: #bef264; --draw-map-soil-5: #86efac; --draw-map-soil-6: #e879f9; + --draw-map-landslide-1: #dc2626; /* 산사태위험지도 1등급(가장 위험) */ + --draw-map-landslide-2: #f97316; + --draw-map-landslide-3: #facc15; + --draw-map-landslide-4: #a3e635; + --draw-map-landslide-5: #4ade80; --draw-map-fault: #f43f5e; /* 단층 */ --draw-map-lineament: #c4b5fd; /* 지질구조선(점선) */ --draw-map-zero-order: #7dd3fc; /* 0차곡(지형 추정 골짜기 머리 · 점선) */