feat(D03 겹): 노선분석 지도 산림입지도 겹 — 브이월드 WFS 토양형 · 묶음 셋째 · 처음 끔 (PLAN 581-2 · 요구 440-8)
- 서버: lt_c_fsdifrsts 받은 범위로 잘라 토양형 코드 · 이름 · 넓이 · 토양형별 합 · 덮으면 다시 안 받음 · 실패 = 502 - 화면: 단추 하나(잎 아이콘 · 부품 상자에 더함) · 토양형 색 --draw-map-soil-1~6 · 범례 = 토양형 이름 · ha · 바탕 캔버스 - 산사태위험지도(생활안전지도 = 공공누리 4유형 · 데이터 준비 중) · 단층 · 지질구조선(이용 조건 문서 없음)은 넣지 않음(브레인 결정 A) - test_553_4: 겹 목록에 soils 더함(새 겹 · 기대 목록 갱신) Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
This commit is contained in:
@@ -10,6 +10,9 @@
|
||||
- 보호 · 규제 구역 = 브이월드 WFS 여덟 종(겹 이름 = `_분석_440-C` 표 1) ·
|
||||
받은 범위로 잘라 종류 · 이름 · 넓이만 · 저장 = `processed/보호구역_bounds.geojson` + 곁 열쇠 ·
|
||||
종류 하나 실패 = 그 종류만 빠짐
|
||||
- 산림입지도 = 브이월드 WFS `lt_c_fsdifrsts`(원자료 산림청 산림입지토양도 · PLAN 581-2) ·
|
||||
받은 범위로 잘라 토양형 코드 · 이름 · 넓이만 ·
|
||||
저장 = `processed/산림입지도_bounds.geojson` + 곁 열쇠
|
||||
- 계류 100 m · 도로 300 m 띠 = 도엽 하천중심선(세류 뺌) · 도로중심선(미분류 · 소로 ·
|
||||
면리간도로 뺌) 합친 띠 · 받은 범위로 자름 · 파생 저장본(열쇠 = 두 파일 지문)
|
||||
|
||||
@@ -316,12 +319,27 @@ def _transforms(epsg: int) -> tuple[Any, Any]:
|
||||
return to_m, to_ll
|
||||
|
||||
|
||||
def _clip_metric(feature: dict[str, Any], frame: Any, to_m: Any) -> Any:
|
||||
"""지물 면(경위도) → 작업 좌표계 m 로 바꿔 범위로 자름 · 못 읽거나 빈 것 = None."""
|
||||
from shapely.geometry import shape
|
||||
from shapely.ops import transform
|
||||
|
||||
try:
|
||||
metric = transform(to_m, shape(feature["geometry"]))
|
||||
except (KeyError, TypeError, ValueError, AttributeError):
|
||||
return None
|
||||
if not metric.is_valid:
|
||||
metric = metric.buffer(0)
|
||||
part = metric.intersection(frame)
|
||||
return None if part.is_empty or part.area <= 0 else part
|
||||
|
||||
|
||||
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.geometry import mapping
|
||||
from shapely.ops import transform
|
||||
|
||||
to_m, to_ll = _transforms(epsg)
|
||||
@@ -331,14 +349,8 @@ def clip_zones(
|
||||
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:
|
||||
part = _clip_metric(feature, frame, to_m)
|
||||
if part is None:
|
||||
continue
|
||||
props = feature.get("properties") or {}
|
||||
name = next((str(props[k]) for k in ZONE_NAME_PROPS if props.get(k)), label)
|
||||
@@ -394,6 +406,73 @@ def ensure_zones(processed: Path, box: Box, epsg: int) -> dict[str, Any]:
|
||||
return {"path": path, "fetched": True, "failed": failed}
|
||||
|
||||
|
||||
# ── 산림입지도 ──────────────────────────────────────────
|
||||
|
||||
SOIL_TYPENAME = "lt_c_fsdifrsts"
|
||||
SOIL_FILE = "산림입지도_bounds.geojson"
|
||||
SOIL_KEY = "산림입지도_bounds.key.json"
|
||||
SOIL_VERSION = 1
|
||||
|
||||
|
||||
def clip_soils(
|
||||
found: list[dict[str, Any]], box: Box, epsg: int
|
||||
) -> tuple[list[dict[str, Any]], list[dict[str, Any]]]:
|
||||
"""산림입지도 지물 → 받은 범위로 자른 지물(토양형 코드 · 이름 · 넓이) + 토양형별 {코드, 이름,
|
||||
개수, 넓이 m²}(코드 차례). 이름 없으면 토양형 글자 그대로."""
|
||||
from shapely.geometry import box as rect
|
||||
from shapely.geometry import mapping
|
||||
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 feature in dedup_parcels(found):
|
||||
part = _clip_metric(feature, frame, to_m)
|
||||
if part is None:
|
||||
continue
|
||||
props = feature.get("properties") or {}
|
||||
code = str(props.get("toyanghyun") or props.get("sltp_cd") or "?")
|
||||
name = str(props.get("name") or code)
|
||||
row = summary.setdefault(
|
||||
code,
|
||||
{"code": code, "sort": str(props.get("sltp_cd") or code), "name": name, "count": 0},
|
||||
)
|
||||
row["count"] += 1
|
||||
row["area_m2"] = row.get("area_m2", 0.0) + part.area
|
||||
features.append(
|
||||
{
|
||||
"type": "Feature",
|
||||
"geometry": mapping(transform(to_ll, part)),
|
||||
"properties": {"code": code, "name": name, "area_m2": round(part.area, 1)},
|
||||
}
|
||||
)
|
||||
rows = sorted(summary.values(), key=lambda row: (row["sort"], row["code"]))
|
||||
for row in rows:
|
||||
del row["sort"]
|
||||
row["area_m2"] = round(row["area_m2"], 1)
|
||||
return features, rows
|
||||
|
||||
|
||||
def ensure_soils(processed: Path, box: Box, epsg: int) -> dict[str, Any]:
|
||||
"""산림입지도 — 열쇠 상자가 덮으면 그대로 · 아니면 범위 통째로 다시 받음 · 실패 = 오류."""
|
||||
path, key_path = processed / SOIL_FILE, processed / SOIL_KEY
|
||||
key = _read_json(key_path) or {}
|
||||
if path.is_file() and key.get("version") == SOIL_VERSION and _covers(key.get("box"), box):
|
||||
return {"path": path, "fetched": False}
|
||||
got, failed = fetch_tile(box, epsg, 0, SOIL_TYPENAME)
|
||||
if failed:
|
||||
raise RuntimeError("산림입지도를 받지 못했습니다 — 브이월드 WFS 실패")
|
||||
features, summary = clip_soils(got, box, epsg)
|
||||
processed.mkdir(parents=True, exist_ok=True)
|
||||
atomic_write_json(path, {"type": "FeatureCollection", "features": features, "summary": summary})
|
||||
atomic_write_json(
|
||||
key_path, {"version": SOIL_VERSION, "box": [float(v) for v in box], "epsg": int(epsg)}
|
||||
)
|
||||
logger.info("D 산림입지도 받기: 지물 %d · 토양형 %d", len(features), len(summary))
|
||||
return {"path": path, "fetched": True}
|
||||
|
||||
|
||||
# ── 계류 · 도로 띠 ───────────────────────────────────────
|
||||
|
||||
BAND_KIND = "plan_band"
|
||||
|
||||
Reference in New Issue
Block a user