Merge remote-tracking branch 'origin/sub_laptop_4' into main_laptop_1
This commit is contained in:
@@ -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과 구분)
|
||||
|
||||
@@ -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
|
||||
|
||||
@@ -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}
|
||||
|
||||
@@ -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
|
||||
Reference in New Issue
Block a user