diff --git a/B04_PreProcess/B04_PreProcess_Engine_SheetParser.py b/B04_PreProcess/B04_PreProcess_Engine_SheetParser.py index 59bf7027c..ccb83b0c7 100644 --- a/B04_PreProcess/B04_PreProcess_Engine_SheetParser.py +++ b/B04_PreProcess/B04_PreProcess_Engine_SheetParser.py @@ -34,6 +34,11 @@ SHEET_LAYERS: dict[str, tuple[str, str]] = { "기준점": ("N3P", "H0020000"), "지명": ("N3P", "H0040000"), "도로중심선": ("N3L", "A0020000"), + # D 노선분석 겹(PLAN 581-1) — 건물 · 수계(E) 면 셋(실폭하천 · 호소 · 저수지 등 물 면) + "건물": ("N3A", "B0010000"), + "실폭하천": ("N3A", "E0010001"), + "수계면E003": ("N3A", "E0032111"), + "수계면E005": ("N3A", "E0052114"), } # 산출 파일 접두사 (기존 processed/의 국가 GIS geojson과 구분) diff --git a/D03_PlanAnalysis/D03_PlanAnalysis_Layers.py b/D03_PlanAnalysis/D03_PlanAnalysis_Layers.py index 3e01d1ddc..0aee294eb 100644 --- a/D03_PlanAnalysis/D03_PlanAnalysis_Layers.py +++ b/D03_PlanAnalysis/D03_PlanAnalysis_Layers.py @@ -6,6 +6,12 @@ 저장 = `processed/연속지적도_bounds.geojson`(B04 `/geojson?layer=지적도` 자리) + 곁 열쇠 - 급경사 = 평활 안 한 사업 DEM(`models_plan/dtm_sheet_<방법>.npz`) · E07 `slope_rad` · 15° · 35° · 40° 단계 PNG · 파생 저장본(열쇠 = 격자 지문) +- 음영기복 = 같은 DEM · B07 `hillshade_png`(PLAN 581-1) +- 보호 · 규제 구역 = 브이월드 WFS 여덟 종(겹 이름 = `_분석_440-C` 표 1) · + 받은 범위로 잘라 종류 · 이름 · 넓이만 · 저장 = `processed/보호구역_bounds.geojson` + 곁 열쇠 · + 종류 하나 실패 = 그 종류만 빠짐 +- 계류 100 m · 도로 300 m 띠 = 도엽 하천중심선(세류 뺌) · 도로중심선(미분류 · 소로 · + 면리간도로 뺌) 합친 띠 · 받은 범위로 자름 · 파생 저장본(열쇠 = 두 파일 지문) 계산 자리 = ③ 서버 단독. """ @@ -49,6 +55,8 @@ MAX_SPLIT = 3 # 1 km → 125 m 까지 쪼갬 WFS_WORKERS = 4 SLOPE_KIND = "plan_slope" +HILLSHADE_KIND = "plan_hillshade" +RASTER_KINDS = ("slope", "hillshade") SLOPE_VERSION = 1 # (아래 °, 위 °, 색 RGBA) — 임도망_편성 완 · 중 · 급경사지 · 타당성평가 35° · 노선_배치 40° 연속 SLOPE_BANDS = ( @@ -115,21 +123,21 @@ def _lonlat_bounds(box: Box, epsg: int) -> dict[str, float]: return {"min_lon": min(lons), "min_lat": min(lats), "max_lon": max(lons), "max_lat": max(lats)} -def _wfs(box: Box, epsg: int) -> list[dict[str, Any]] | None: +def _wfs(box: Box, epsg: int, typename: str) -> list[dict[str, Any]] | None: """조각 하나 — B04 WFS 받기(실패 = 파일 없음 · 키 값은 남기지 않음). None = 실패.""" with tempfile.TemporaryDirectory() as folder: - request_vworld_wfs( - CADASTRAL_TYPENAME, "tile.geojson", _lonlat_bounds(box, epsg), Path(folder) - ) + request_vworld_wfs(typename, "tile.geojson", _lonlat_bounds(box, epsg), Path(folder)) data = _read_json(Path(folder) / "tile.geojson") if not isinstance(data, dict) or not isinstance(data.get("features"), list): return None return data["features"] -def fetch_tile(box: Box, epsg: int, depth: int = 0) -> tuple[list[dict[str, Any]], int]: - """(필지, 실패 조각 수) — 1,000 꽉 차면 넷으로 다시 쪼갬.""" - features = _wfs(box, epsg) +def fetch_tile( + box: Box, epsg: int, depth: int = 0, typename: str = CADASTRAL_TYPENAME +) -> tuple[list[dict[str, Any]], int]: + """(지물, 실패 조각 수) — 1,000 꽉 차면 넷으로 다시 쪼갬.""" + features = _wfs(box, epsg, typename) if features is None: return [], 1 if len(features) < WFS_LIMIT or depth >= MAX_SPLIT: @@ -137,7 +145,7 @@ def fetch_tile(box: Box, epsg: int, depth: int = 0) -> tuple[list[dict[str, Any] out: list[dict[str, Any]] = [] failed = 0 for part in quarter(box): - got, bad = fetch_tile(part, epsg, depth + 1) + got, bad = fetch_tile(part, epsg, depth + 1, typename) out.extend(got) failed += bad return out, failed @@ -230,24 +238,37 @@ def slope_bands_png(z: np.ndarray, valid: np.ndarray, cell_m: float) -> dict[str } -def slope_layer(project_root: Path, models_dir: Path, epsg: int) -> dict[str, Any]: - """급경사 겹 — 지문(격자 파일 · 단계 · 판) 같으면 저장본 그대로.""" +def hillshade_layer_png(z: np.ndarray, valid: np.ndarray, cell_m: float) -> dict[str, Any]: + """음영기복 PNG(B07 그대로 · 빈 칸 흰색 = 화면이 곱하기로 겹쳐 투명).""" + from B07_DesignDetail.B07_DesignDetail_Engine_Cad_Lidar import hillshade_png + + image, width, height = hillshade_png(z, valid, cell_m) + return {"width": width, "height": height, "image": image, "bands": []} + + +def slope_layer( + project_root: Path, models_dir: Path, epsg: int, kind: str = "slope" +) -> dict[str, Any]: + """급경사(`slope`) · 음영기복(`hillshade`) — 지문(격자 파일 · 단계 · 판) 같으면 저장본.""" from common_util.common_util_crs import tm_to_lonlat npz = models_dir / f"dtm_sheet_{SHEET_SURFACE_DEFAULT_METHOD}.npz" stamp = file_stamp(npz) if stamp is None: raise FileNotFoundError("사업 지형면이 없습니다 — 지형 받기 뒤 보입니다.") - key = derived_key( - SLOPE_KIND, SLOPE_VERSION, npz.name, stamp, int(epsg), SLOPE_BANDS, MAX_SIDE_PX + store_kind, build = ( + (SLOPE_KIND, slope_bands_png) if kind == "slope" else (HILLSHADE_KIND, hillshade_layer_png) ) - saved = read_derived(project_root, SLOPE_KIND, key) + key = derived_key( + store_kind, SLOPE_VERSION, npz.name, stamp, int(epsg), SLOPE_BANDS, MAX_SIDE_PX + ) + saved = read_derived(project_root, store_kind, key) if saved is not None: return saved["value"] with np.load(npz) as dem: x, y = np.asarray(dem["x"], float), np.asarray(dem["y"], float) cell = float(dem["resolution"][0]) if "resolution" in dem.files else float(x[1] - x[0]) - value = slope_bands_png(dem["z"], np.asarray(dem["valid_mask"], bool), cell) + value = build(dem["z"], np.asarray(dem["valid_mask"], bool), cell) half = cell / 2 box = (x[0] - half, y[0] - half, x[-1] + half, y[-1] + half) value["box_m"] = [float(v) for v in box] @@ -255,5 +276,195 @@ def slope_layer(project_root: Path, models_dir: Path, epsg: int) -> dict[str, An list(tm_to_lonlat(px, py, epsg)) for px, py in ((box[0], box[1]), (box[2], box[1]), (box[2], box[3]), (box[0], box[3])) ] - write_derived(project_root, SLOPE_KIND, key, value) + write_derived(project_root, store_kind, key, value) + return value + + +# ── 보호 · 규제 구역 ───────────────────────────────────── + +# (열쇠, 이름, 브이월드 WFS 겹 이름) — `_분석_440-C` 표 1 그대로 +ZONE_KINDS: tuple[tuple[str, str, tuple[str, ...]], ...] = ( + ("forest", "산림보호구역", ("lt_c_uf151",)), + ("baekdu", "백두대간보호지역", ("lt_c_uf901",)), + ("water", "상수원보호구역", ("lt_c_um710",)), + ("park", "자연공원", ("lt_c_wgisnpgug", "lt_c_wgisnpdo", "lt_c_wgisnpgun")), + ("heritage", "국가유산 지정 · 보호구역", ("lt_c_uo301",)), + ("wetland", "습지보호지역", ("lt_c_um901",)), + ("wildlife", "야생생물보호구역", ("lt_c_um221",)), + ("greenbelt", "개발제한구역", ("lt_c_ud801",)), +) +ZONE_FILE = "보호구역_bounds.geojson" +ZONE_KEY = "보호구역_bounds.key.json" +ZONE_VERSION = 1 +ZONE_NAME_PROPS = ("alias", "park_name", "uname") + + +def _transforms(epsg: int) -> tuple[Any, Any]: + """(경위도 → m, m → 경위도) shapely 변환 함수.""" + from pyproj import Transformer + + to_m = Transformer.from_crs("EPSG:4326", f"EPSG:{epsg}", always_xy=True).transform + to_ll = Transformer.from_crs(f"EPSG:{epsg}", "EPSG:4326", always_xy=True).transform + return to_m, to_ll + + +def clip_zones( + found: dict[str, list[dict[str, Any]]], box: Box, epsg: int +) -> tuple[list[dict[str, Any]], dict[str, dict[str, Any]]]: + """종류별 지물 → 받은 범위로 자른 지물(종류 · 이름) + 종류별 {이름, 개수, 넓이 m²}.""" + from shapely.geometry import box as rect + from shapely.geometry import mapping, shape + from shapely.ops import transform + + to_m, to_ll = _transforms(epsg) + frame = rect(*box) + features: list[dict[str, Any]] = [] + summary: dict[str, dict[str, Any]] = {} + for kind, label, _ in ZONE_KINDS: + count, area = 0, 0.0 + for feature in dedup_parcels(found.get(kind) or []): + try: + metric = transform(to_m, shape(feature["geometry"])) + except (KeyError, TypeError, ValueError, AttributeError): + continue + if not metric.is_valid: + metric = metric.buffer(0) + part = metric.intersection(frame) + if part.is_empty or part.area <= 0: + continue + props = feature.get("properties") or {} + name = next((str(props[k]) for k in ZONE_NAME_PROPS if props.get(k)), label) + count += 1 + area += part.area + features.append( + { + "type": "Feature", + "geometry": mapping(transform(to_ll, part)), + "properties": {"kind": kind, "name": name, "area_m2": round(part.area, 1)}, + } + ) + summary[kind] = {"label": label, "count": count, "area_m2": round(area, 1)} + return features, summary + + +def ensure_zones(processed: Path, box: Box, epsg: int) -> dict[str, Any]: + """보호 · 규제 구역 — 열쇠 상자가 덮고 실패 종류 없으면 그대로 · 아니면 통째로 다시 받음.""" + path, key_path = processed / ZONE_FILE, processed / ZONE_KEY + key = _read_json(key_path) or {} + if ( + path.is_file() + and key.get("version") == ZONE_VERSION + and not key.get("failed") + and _covers(key.get("box"), box) + ): + return {"path": path, "fetched": False, "failed": []} + jobs = [(kind, name) for kind, _, names in ZONE_KINDS for name in names] + # ponytail: 겹마다 범위 통째 한 번 · 1,000 꽉 차면 넷 — 구역은 필지보다 훨씬 적음 + with ThreadPoolExecutor(max_workers=WFS_WORKERS) as pool: + results = list(pool.map(lambda job: fetch_tile(box, epsg, 0, job[1]), jobs)) + failed = sorted({kind for (kind, _), (_, bad) in zip(jobs, results) if bad}) + if len(failed) == len(ZONE_KINDS): + raise RuntimeError("보호 · 규제 구역을 받지 못했습니다 — 브이월드 WFS 실패") + found: dict[str, list[dict[str, Any]]] = {} + for (kind, _), (got, _) in zip(jobs, results): + found.setdefault(kind, []).extend(got) + features, summary = clip_zones(found, box, epsg) + for kind in failed: + summary[kind]["failed"] = True + processed.mkdir(parents=True, exist_ok=True) + atomic_write_json(path, {"type": "FeatureCollection", "features": features, "summary": summary}) + atomic_write_json( + key_path, + { + "version": ZONE_VERSION, + "box": [float(v) for v in box], + "epsg": int(epsg), + "failed": failed, + }, + ) + logger.info("D 보호구역 받기: 지물 %d · 실패 종류 %s", len(features), failed) + return {"path": path, "fetched": True, "failed": failed} + + +# ── 계류 · 도로 띠 ─────────────────────────────────────── + +BAND_KIND = "plan_band" +BAND_VERSION = 1 +# 타당성평가 「산지계류 100 m 이내」 · 「도로법 도로 300 m 이내」 +STREAM_BAND_M = 100.0 +ROAD_BAND_M = 300.0 +# ponytail: 원문 = 1/25,000 주요 수계 · 도로법 도로 — 1/5,000 도엽에서 세류 · 소로급을 뺀 +# 어림(협의 값) +SMALL_STREAMS = frozenset({"세류"}) +LOCAL_ROADS = frozenset({"미분류", "소로", "면리간도로"}) +BAND_SEGMENTS = 8 # 둥근 끝 사분원 꼭짓점 + + +def band_union(lines: list[Any], dist_m: float, box: Box) -> Any: + """선(작업 좌표계 m) → 거리 `dist_m` 띠 하나(겹친 곳 합침 · 범위로 자름) · 선 없으면 None.""" + from shapely import union_all + from shapely.geometry import box as rect + + if not lines: + return None + band = union_all(lines).buffer(dist_m, quad_segs=BAND_SEGMENTS).intersection(rect(*box)) + return None if band.is_empty else band + + +def _band( + path: Path, epsg: int, box: Box, dist_m: float, prop: str, skip: frozenset[str] +) -> dict[str, Any]: + from shapely.geometry import mapping, shape + from shapely.ops import transform + + to_m, to_ll = _transforms(epsg) + features = (_read_json(path) or {}).get("features") or [] + lines = [ + transform(to_m, shape(f["geometry"])) + for f in features + if f.get("geometry") and (f.get("properties") or {}).get(prop) not in skip + ] + band = band_union(lines, dist_m, box) + frame = (box[2] - box[0]) * (box[3] - box[1]) + area = band.area if band is not None else 0.0 + shown = [] if band is None else [band.simplify(0.5)] + return { + "dist_m": dist_m, + "lines": len(lines), + "area_m2": round(area, 1), + "ratio": round(area / frame, 4) if frame > 0 else 0.0, + "collection": { + "type": "FeatureCollection", + "features": [ + {"type": "Feature", "geometry": mapping(transform(to_ll, g)), "properties": {}} + for g in shown + ], + }, + } + + +def band_layer(project_root: Path, epsg: int, box: Box) -> dict[str, Any]: + """계류 100 m · 도로 300 m 띠 — 두 도엽 겹 지문 · 범위 같으면 저장본 그대로.""" + stream = ensure_sheet_layer(project_root, "하천중심선") + road = ensure_sheet_layer(project_root, "도로중심선") + key = derived_key( + BAND_KIND, + BAND_VERSION, + file_stamp(stream), + file_stamp(road), + int(epsg), + [round(v, 1) for v in box], + STREAM_BAND_M, + ROAD_BAND_M, + sorted(SMALL_STREAMS), + sorted(LOCAL_ROADS), + ) + saved = read_derived(project_root, BAND_KIND, key) + if saved is not None: + return saved["value"] + value = { + "stream": _band(stream, epsg, box, STREAM_BAND_M, "구분", SMALL_STREAMS), + "road": _band(road, epsg, box, ROAD_BAND_M, "도로구분", LOCAL_ROADS), + } + write_derived(project_root, BAND_KIND, key, value) return value diff --git a/D03_PlanAnalysis/D03_PlanAnalysis_Router.py b/D03_PlanAnalysis/D03_PlanAnalysis_Router.py index 1de51d747..8fbd6c7fd 100644 --- a/D03_PlanAnalysis/D03_PlanAnalysis_Router.py +++ b/D03_PlanAnalysis/D03_PlanAnalysis_Router.py @@ -3,7 +3,8 @@ - `GET /api/projects/{id}/plan/terrain/extent` — 받기 전 범위(상자 · 도엽 · 칸 수 · 받은 범위) - `POST …/plan/terrain` — 사업 전체 지형 한 벌 받기(뒤에서 · 누를 때만) · 끝나면 `surface_models` 행 - `GET …/plan/progress` — 일의 진행률 -- `GET …/plan/layers/sheet?key=` · `…/cadastral` · `…/raster?kind=slope` — 지도 겹(PLAN 569-1) +- `GET …/plan/layers/sheet?key=` · `…/cadastral` · `…/raster?kind=slope|hillshade` · + `…/zones` · `…/bands` — 지도 겹(PLAN 569-1 · 581-1) 범위 = 전체 시점 · 종점(D02 정본 `load_base`) + 그은 점(`extra`) 상자 + 여유. 노선 정본(손 그은 선)은 @@ -197,12 +198,37 @@ 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°) — 격자 지문 같으면 저장본.""" - if kind != "slope": + """급경사 단계 그림(15° · 35° · 40°) · 음영기복(`hillshade`) — 격자 지문 같으면 저장본.""" + if kind not in layers.RASTER_KINDS: raise HTTPException(status_code=400, detail=f"없는 그림 겹: {kind}") root, epsg, _ = await _layer_root(project_id) try: - value = await asyncio.to_thread(layers.slope_layer, root, terrain.models_dir_of(root), epsg) + value = await asyncio.to_thread( + layers.slope_layer, root, terrain.models_dir_of(root), epsg, kind + ) except FileNotFoundError as error: raise HTTPException(status_code=404, detail=str(error)) from error return {"status": "success", "kind": kind, **value} + + +@router.get("/{project_id}/plan/layers/zones", response_model=None) +async def plan_layer_zones(project_id: UUID, request: Request) -> Response: + """보호 · 규제 구역 GeoJSON(여덟 종 · `summary` = 종류별 개수 · 넓이) — 범위 밖이면 받음.""" + root, epsg, box = await _layer_root(project_id) + processed = root / terrain.STAGE / "processed" + try: + got = await asyncio.to_thread(layers.ensure_zones, processed, box, epsg) + except RuntimeError as error: + raise HTTPException(status_code=502, detail=str(error)) from error + return cached_file_response(request, got["path"], "application/geo+json") + + +@router.get("/{project_id}/plan/layers/bands", response_model=None) +async def plan_layer_bands(project_id: UUID) -> dict[str, Any]: + """계류 100 m · 도로 300 m 띠(넓이 · 받은 범위 비율) — 두 도엽 겹 지문 같으면 저장본.""" + root, epsg, box = await _layer_root(project_id) + try: + value = await asyncio.to_thread(layers.band_layer, root, epsg, box) + except FileNotFoundError as error: + raise HTTPException(status_code=404, detail=str(error)) from error + return {"status": "success", **value} diff --git a/resources/tester/test_581_1_plan_gis_layers.py b/resources/tester/test_581_1_plan_gis_layers.py new file mode 100644 index 000000000..3bd2f37f8 --- /dev/null +++ b/resources/tester/test_581_1_plan_gis_layers.py @@ -0,0 +1,201 @@ +"""D 노선분석 GIS 겹 「지금」 묶음(PLAN 581-1) — 계류 · 도로 띠 합침 · 넓이 · 보호구역 잘라 넓이 · +종류 하나 실패 = 그 종류만 빠짐 · 모두 실패 = 오류 · 덮으면 다시 안 받음 · 음영기복 저장본 따로 · +도엽 건물 · 물 면 코드. + +./venv/Scripts/python.exe -m pytest resources/tester/test_581_1_plan_gis_layers.py -q +""" + +import json +import math +from pathlib import Path + +import numpy as np +import pytest +from shapely.geometry import LineString + +from B04_PreProcess.B04_PreProcess_Engine_SheetParser import SHEET_LAYERS +from common_util.common_util_crs import tm_to_lonlat +from D03_PlanAnalysis import D03_PlanAnalysis_Layers as layers + +EPSG = 5186 +BOX = (230000.0, 450000.0, 232000.0, 451000.0) # 2 × 1 km + + +def test_sheet_codes() -> None: + assert SHEET_LAYERS["건물"] == ("N3A", "B0010000") + assert SHEET_LAYERS["실폭하천"] == ("N3A", "E0010001") + assert {SHEET_LAYERS["수계면E003"][1], SHEET_LAYERS["수계면E005"][1]} == { + "E0032111", + "E0052114", + } + + +# ── 띠 ── + + +def test_band_one_line_area() -> None: + line = LineString([(230500.0, 450500.0), (231500.0, 450500.0)]) + band = layers.band_union([line], 100.0, BOX) + assert band.area == pytest.approx(2 * 100 * 1000 + math.pi * 100**2, rel=0.01) + + +def test_band_overlap_merged_and_clipped() -> None: + a = LineString([(230500.0, 450500.0), (231500.0, 450500.0)]) + twin = layers.band_union([a, a], 100.0, BOX) + assert twin.area == pytest.approx(layers.band_union([a], 100.0, BOX).area) # 겹친 곳 한 번 + edge = LineString([(230000.0, 450500.0), (232000.0, 450500.0)]) + assert layers.band_union([edge], 100.0, BOX).area == pytest.approx(2000 * 200, rel=1e-6) + assert layers.band_union([], 100.0, BOX) is None + far = LineString([(240000.0, 460000.0), (240100.0, 460000.0)]) + assert layers.band_union([far], 100.0, BOX) is None + + +def _line_feature(points: list[tuple[float, float]], **props) -> dict: + coords = [list(tm_to_lonlat(x, y, EPSG)) for x, y in points] + return { + "type": "Feature", + "geometry": {"type": "LineString", "coordinates": coords}, + "properties": props, + } + + +def test_band_layer_filters_and_caches(tmp_path, monkeypatch) -> None: + processed = tmp_path / "B04_PreProcess" / "processed" + processed.mkdir(parents=True) + row = [(230500.0, 450500.0), (231500.0, 450500.0)] + streams = [ + _line_feature(row, 구분="지방하천"), + _line_feature([(230200.0, 450100.0), (230200.0, 450900.0)], 구분="세류"), + ] + roads = [_line_feature(row, 도로구분="지방도"), _line_feature(row, 도로구분="소로")] + for key, feats in (("하천중심선", streams), ("도로중심선", roads)): + body = {"type": "FeatureCollection", "features": feats} + (processed / f"도엽_{key}.geojson").write_text( + json.dumps(body, ensure_ascii=False), "utf-8" + ) + real = layers.band_union + counted = [] + monkeypatch.setattr(layers, "band_union", lambda *a: counted.append(1) or real(*a)) + value = layers.band_layer(tmp_path, EPSG, BOX) + assert value["stream"]["lines"] == 1 and value["road"]["lines"] == 1 # 세류 · 소로 뺌 + assert value["stream"]["area_m2"] == pytest.approx(2e5 + math.pi * 1e4, rel=0.02) + road = 2 * 300 * 1000 + math.pi * 300**2 # 둥근 끝 · 범위 안 + assert value["road"]["area_m2"] == pytest.approx(road, rel=0.01) + assert value["road"]["ratio"] == pytest.approx(road / 2e6, abs=0.005) + assert len(value["stream"]["collection"]["features"]) == 1 + again = layers.band_layer(tmp_path, EPSG, BOX) + assert again == json.loads(json.dumps(value)) and len(counted) == 2 # 저장본 + + +# ── 보호 · 규제 구역 ── + + +def _square(x0: float, y0: float, side: float) -> dict: + ring = [(x0, y0), (x0 + side, y0), (x0 + side, y0 + side), (x0, y0 + side), (x0, y0)] + return {"type": "Polygon", "coordinates": [[list(tm_to_lonlat(x, y, EPSG)) for x, y in ring]]} + + +def test_clip_zones_area_and_names() -> None: + found = { + "forest": [ + { + "id": "a", + "geometry": _square(231800.0, 450800.0, 400.0), + "properties": {"alias": "보안림"}, + }, + { + "id": "a", + "geometry": _square(231800.0, 450800.0, 400.0), + "properties": {}, + }, # 조각 겹침 + { + "id": "b", + "geometry": _square(240000.0, 460000.0, 100.0), + "properties": {}, + }, # 범위 밖 + ], + "park": [ + { + "id": "c", + "geometry": _square(230100.0, 450100.0, 100.0), + "properties": {"park_name": "속리산"}, + } + ], + } + features, summary = layers.clip_zones(found, BOX, EPSG) + assert summary["forest"]["count"] == 1 + assert summary["forest"]["area_m2"] == pytest.approx(200 * 200, rel=0.01) # 귀퉁이만 남음 + assert summary["park"]["area_m2"] == pytest.approx(100 * 100, rel=0.01) + assert summary["water"] == {"label": "상수원보호구역", "count": 0, "area_m2": 0.0} + assert [f["properties"]["name"] for f in features] == ["보안림", "속리산"] + assert {f["properties"]["kind"] for f in features} == {"forest", "park"} + + +class FakeZoneWfs: + """겹 이름마다 정사각형 하나 · `fail` 에 든 겹 이름은 실패(파일 없음).""" + + def __init__(self, fail: frozenset[str] = frozenset()) -> None: + self.fail, self.calls = fail, [] + + def __call__(self, typename, filename, bounds, output_dir) -> None: + self.calls.append(typename) + if typename in self.fail: + return + feats = [{"id": typename, "geometry": _square(230500.0, 450200.0, 300.0), "properties": {}}] + body = {"type": "FeatureCollection", "features": feats} + (Path(output_dir) / filename).write_text(json.dumps(body), "utf-8") + + +ALL_NAMES = [name for _, _, names in layers.ZONE_KINDS for name in names] + + +def test_zones_fetch_once_partial_failure(tmp_path, monkeypatch) -> None: + assert len(layers.ZONE_KINDS) == 8 and len(ALL_NAMES) == 10 + fake = FakeZoneWfs(fail=frozenset({"lt_c_uf151"})) + monkeypatch.setattr(layers, "request_vworld_wfs", fake) + got = layers.ensure_zones(tmp_path, BOX, EPSG) + assert ( + got["fetched"] and got["failed"] == ["forest"] and sorted(fake.calls) == sorted(ALL_NAMES) + ) + saved = json.loads((tmp_path / layers.ZONE_FILE).read_text("utf-8")) + assert saved["summary"]["forest"].get("failed") is True + assert saved["summary"]["park"]["count"] == 3 # 자연공원 겹 셋 + assert len(saved["features"]) == 9 + ok = FakeZoneWfs() + monkeypatch.setattr(layers, "request_vworld_wfs", ok) + assert layers.ensure_zones(tmp_path, BOX, EPSG)["fetched"] # 실패 종류 있으면 다시 + assert not layers.ensure_zones(tmp_path, (230500.0, 450100.0, 231000.0, 450900.0), EPSG)[ + "fetched" + ] + assert len(ok.calls) == 10 # 덮으면 다시 안 받음 + + +def test_zones_all_fail(tmp_path, monkeypatch) -> None: + monkeypatch.setattr(layers, "request_vworld_wfs", FakeZoneWfs(fail=frozenset(ALL_NAMES))) + with pytest.raises(RuntimeError): + layers.ensure_zones(tmp_path, BOX, EPSG) + assert not (tmp_path / layers.ZONE_FILE).exists() + + +# ── 음영기복 ── + + +def test_hillshade_saved_apart_from_slope(tmp_path) -> None: + models = tmp_path / "B04_PreProcess" / "models_plan" + models.mkdir(parents=True) + n = 30 + z = np.tile(np.arange(n) * 1.0, (n, 1)) + x = 200000.0 + np.arange(n) * 2.0 + np.savez( + models / "dtm_sheet_laplace.npz", + x=x, + y=x + 250000.0, + z=z, + valid_mask=np.ones(z.shape, bool), + resolution=[2.0], + ) + shade = layers.slope_layer(tmp_path, models, EPSG, "hillshade") + slope = layers.slope_layer(tmp_path, models, EPSG, "slope") + assert shade["image"].startswith("data:image/png;base64,") and shade["bands"] == [] + assert slope["bands"] and shade["box_m"] == slope["box_m"] + assert layers.slope_layer(tmp_path, models, EPSG, "hillshade") == shade