feat(D03 겹): 노선분석 지도 산사태위험지도 겹 — 행안부 생활안전지도 WMS · 키 없으면 그 겹만 꺼짐 (PLAN 584-1 · 요구 444)

- 서버: 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>
2026-10-29 16:59:11 +09:00
parent 7d4a94a887
commit 84f243cee1
9 changed files with 262 additions and 25 deletions
@@ -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=
@@ -151,7 +151,7 @@ export const fetchPlanSheetLayer = (projectId: string, key: string): Promise<Geo

 export const fetchPlanRaster = (
  projectId: string,
-  kind: "slope" | "hillshade",
+  kind: "slope" | "hillshade" | "landslide",
 ): Promise<PlanRaster> => request(`${base(projectId)}/layers/raster?kind=${kind}`);

 /** 보호 · 규제 구역(581-1) — 지물 `kind` · `name` · `area_m2` + 종류별 개수 · 넓이(실패 = `failed`). */
@@ -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
@@ -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)
@@ -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<string, string> = {
  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<PlanLayerKey, boolean> {
+  const on = Object.fromEntries(LAYERS.map((def) => [def.key, def.on])) as Record<
+    PlanLayerKey,
+    boolean
+  >;
+  try {
+    const saved = JSON.parse(readByKey(key) ?? "{}") as Partial<Record<PlanLayerKey, unknown>>;
+    for (const def of LAYERS)
+      if (typeof saved[def.key] === "boolean") on[def.key] = saved[def.key] as boolean;
+  } catch {
+    // 깨진 값 = 처음 값
+  }
+  return on;
+}
@@ -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<LineLayerKey, PreparedLines> & {
  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<string, PreparedLayer>; summary: PlanZones["summary"] };
  /** 토양형(코드 차례) → 지물 · 색 값. */
@@ -120,21 +123,6 @@ export interface PlanLayers {
  };
 }

-function readPref(): Record<PlanLayerKey, boolean> {
-  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<Record<PlanLayerKey, unknown>>;
-    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<Prepared> = {};
  const pending = new Map<DataKey | "satellite", Promise<void>>();
  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];
@@ -67,6 +67,7 @@ def test_d03_parts():
        ("cadastral", "false"),
        ("zones", "false"),
        ("soils", "false"),
+        ("landslide", "false"),
        ("faults", "false"),
        ("lineaments", "false"),
        ("zeroOrder", "false"),
@@ -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"<xml/>"))
+    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차곡 ──


@@ -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차곡(지형 추정 골짜기 머리 · 점선) */