eomsangdonandClaude Opus 5.5
5db9cb49f1
- 임도망도: 지도 겹 줄에 올리기 단추(zip 하나 · shp · shx · dbf · prj 묶음 · 100 MB · 넘으면 413) → 서버가 읽어 받은 범위로 자름(prj 없으면 EPSG:5179 · cpg 없으면 cp949) · 원본 = 프로젝트 D03_PlanAnalysis/임도망도.zip · 자른 선 = 파생 저장본 · 못 읽거나 범위 안 선 0 = 400 · 옛 것 그대로 - 범례: 「임도 짐작」(주황 점선) · 「올린 임도망도」(진한 주황 실선) 구분 · 짐작 겹 그대로 - 산사태취약지역: data.go.kr 시도 표 넷(경북 · 충남 · 광주 · 경남)을 resources/data_landslide_zones 에 한 벌(머리 = 이름 · 받은 날 · 기준일 · 출처) · 시도마다 다른 열 이름은 별칭으로 · 위경도 = 받은 범위 안 점 · 지번(시군구 · 읍면동 · 리 · 산 · 본번 · 부번) = 지적 필지 테두리만 · 사업지 시도에 표 없으면 범례 「이 시도 자료 없음」 · 지적 못 받으면 점만 - 두 겹 처음 끔 · 바탕 캔버스 · 색 --draw-map-forest-road-file · landslide-zone · 아이콘 alertOctagon · 고르기 창 pickFiles(부품 상자 ui_template_drop) - 범례 · 겹별 수 · 받은 자료 꼴을 D03_PlanAnalysis_UI_LayerLegend.ts 로 옮김(700줄 한도) - test_553_4: 겹 목록에 forestRoadFile · landslideZones 더함(새 겹 · 기대 목록 갱신) Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
265 lines
9.6 KiB
Python
265 lines
9.6 KiB
Python
"""D 노선분석 지도 산사태취약지역 겹(PLAN 584-3 · 요구 443).
|
|
|
|
- 출처 = 시도 표(data.go.kr · 이용허락범위 제한 없음) 한 벌 = `resources/data_landslide_zones/
|
|
<시도코드>_<시도이름>.csv`(UTF-8 · 머리 `#` 줄 = 이름 · 받은 날 · 기준일 · 출처) ·
|
|
해마다 파일만 바꿈
|
|
- 시도마다 열 이름이 다름 → `COLUMNS` 별칭으로 읽음 · 위경도(도 · 분 · 초) 없는 시도(경남) = 지번만
|
|
- 위경도 있는 줄 = 받은 범위 안 점 · 지번(시군구 · 읍면동 · 리 · 산 · 본번 · 부번)이 지적 필지와
|
|
맞으면 필지 테두리(지정 넓이 ≪ 필지 넓이 — 칠하지 않음)
|
|
- 사업지 시도 = 지적 필지 `sido_cd` · 표 없는 시도 = `missing`
|
|
- 저장 = 파생 저장본(열쇠 = 표 지문 · 지적 지문 · 범위)
|
|
|
|
계산 자리 = ③ 서버 단독.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import csv
|
|
import json
|
|
import re
|
|
from functools import lru_cache
|
|
from pathlib import Path
|
|
from typing import Any
|
|
|
|
from common_util.common_util_derived_store import (
|
|
derived_key,
|
|
file_stamp,
|
|
read_derived,
|
|
write_derived,
|
|
)
|
|
|
|
Box = tuple[float, float, float, float]
|
|
|
|
DATA_DIR = Path(__file__).resolve().parents[1] / "resources" / "data_landslide_zones"
|
|
ZONE_KIND = "plan_landslide_zone"
|
|
ZONE_VERSION = 1
|
|
# 열쇠 → 시도 표 열 이름 별칭(경북 · 충남 · 광주 · 경남)
|
|
COLUMNS: dict[str, tuple[str, ...]] = {
|
|
"sgg": ("시군구", "조사지(시군구)"),
|
|
"emd": ("읍면동", "조사지(동)"),
|
|
"ri": ("리",),
|
|
"jibun": ("지번",),
|
|
"lat_d": ("(위도)도", "위도(도)"),
|
|
"lat_m": ("(위도)분", "위도(분)"),
|
|
"lat_s": ("(위도)초", "위도(초)"),
|
|
"lon_d": ("(경도)도", "경도(도)"),
|
|
"lon_m": ("(경도)분", "경도(분)"),
|
|
"lon_s": ("(경도)초", "경도(초)"),
|
|
"kind": ("취약지역유형",),
|
|
"area": ("지정면적(제곱미터)",),
|
|
}
|
|
JIBUN = re.compile(r"^\s*(산)?\s*(\d+)(?:\s*-\s*(\d+))?")
|
|
|
|
ParcelKey = tuple[str, str, str, bool, int, int]
|
|
|
|
|
|
def jibun_key(text: str) -> tuple[bool, int, int] | None:
|
|
"""「산42임」 · 「산77-1번지」 · 「130-61임, 파도리 산65」 → (산, 본번, 부번) · 첫 지번만 ·
|
|
못 읽으면 None."""
|
|
found = JIBUN.match(str(text or ""))
|
|
if not found:
|
|
return None
|
|
return bool(found.group(1)), int(found.group(2)), int(found.group(3) or 0)
|
|
|
|
|
|
def _name(text: Any) -> str:
|
|
return re.sub(r"\s+", "", str(text or ""))
|
|
|
|
|
|
def _number(text: Any) -> float | None:
|
|
try:
|
|
return float(str(text).strip())
|
|
except (TypeError, ValueError):
|
|
return None
|
|
|
|
|
|
def _dms(row: dict[str, str], prefix: str) -> float | None:
|
|
parts = [_number(row.get(f"{prefix}_{p}")) for p in "dms"]
|
|
if parts[0] is None:
|
|
return None
|
|
d, m, s = (v or 0.0 for v in parts)
|
|
return d + m / 60 + s / 3600
|
|
|
|
|
|
def read_table(path: Path) -> tuple[dict[str, str], list[dict[str, Any]]]:
|
|
"""시도 표 → (머리 {이름 · 받은 날 · 기준일 · 출처}, 줄 [{sgg, emd, ri, jibun, key, lon, lat,
|
|
kind, area_m2}]) — 위경도 없거나 못 읽으면 lon · lat = None."""
|
|
return _read_table(str(path), file_stamp(path))
|
|
|
|
|
|
@lru_cache(maxsize=8)
|
|
def _read_table(path: str, _stamp: str | None) -> tuple[dict[str, str], list[dict[str, Any]]]:
|
|
head: dict[str, str] = {}
|
|
body: list[str] = []
|
|
for line in Path(path).read_text(encoding="utf-8-sig").splitlines():
|
|
if line.startswith("#"):
|
|
name, _, value = line[1:].partition(":")
|
|
head[name.strip()] = value.strip()
|
|
elif line.strip():
|
|
body.append(line)
|
|
reader = csv.DictReader(body)
|
|
fields = reader.fieldnames or []
|
|
alias = {
|
|
key: next((f for f in fields if f.strip() in names), None) for key, names in COLUMNS.items()
|
|
}
|
|
rows: list[dict[str, Any]] = []
|
|
for raw in reader:
|
|
row = {key: (raw.get(col) or "").strip() if col else "" for key, col in alias.items()}
|
|
lat, lon = _dms(row, "lat"), _dms(row, "lon")
|
|
rows.append(
|
|
{
|
|
"sgg": row["sgg"],
|
|
"emd": row["emd"],
|
|
"ri": row["ri"],
|
|
"jibun": row["jibun"],
|
|
"key": jibun_key(row["jibun"]),
|
|
"lon": lon if lat is not None else None,
|
|
"lat": lat if lon is not None else None,
|
|
"kind": row["kind"] or None,
|
|
"area_m2": _number(row["area"]),
|
|
}
|
|
)
|
|
return head, rows
|
|
|
|
|
|
def table_of(code: str) -> Path | None:
|
|
"""시도 코드(두 자리) → 표 파일 · 없으면 None."""
|
|
return next(iter(sorted(DATA_DIR.glob(f"{code}_*.csv"))), None)
|
|
|
|
|
|
def parcel_key(props: dict[str, Any]) -> ParcelKey | None:
|
|
"""지적 필지 속성 → 맞춤 열쇠(임야대장 `gbn_cd` 2 = 산)."""
|
|
try:
|
|
main, sub = int(props.get("mnnm")), int(props.get("slno") or 0)
|
|
except (TypeError, ValueError):
|
|
return None
|
|
san = str(props.get("gbn_cd")) == "2"
|
|
return (
|
|
_name(props.get("sgg_nm")),
|
|
_name(props.get("emd_nm")),
|
|
_name(props.get("ri_nm")),
|
|
san,
|
|
main,
|
|
sub,
|
|
)
|
|
|
|
|
|
def row_key(row: dict[str, Any]) -> ParcelKey | None:
|
|
if row["key"] is None:
|
|
return None
|
|
return (_name(row["sgg"]), _name(row["emd"]), _name(row["ri"]), *row["key"])
|
|
|
|
|
|
def match_zones(
|
|
parcels: list[dict[str, Any]], tables: dict[str, Path | None], box: Box, epsg: int
|
|
) -> dict[str, Any]:
|
|
"""지적 필지 · 시도 표 → 받은 범위 안 점 · 지번 맞는 필지 테두리 · 시도별 표 · 없는 시도."""
|
|
from pyproj import Transformer
|
|
from shapely.geometry import Point
|
|
from shapely.geometry import box as rect
|
|
|
|
to_m = Transformer.from_crs("EPSG:4326", f"EPSG:{epsg}", always_xy=True).transform
|
|
frame = rect(*box)
|
|
index: dict[ParcelKey, list[dict[str, Any]]] = {}
|
|
for parcel in parcels:
|
|
key = parcel_key(parcel.get("properties") or {})
|
|
if key is not None:
|
|
index.setdefault(key, []).append(parcel)
|
|
names = {
|
|
str((p.get("properties") or {}).get("sido_cd")): str(
|
|
(p.get("properties") or {}).get("sido_nm")
|
|
)
|
|
for p in parcels
|
|
}
|
|
points: list[dict[str, Any]] = []
|
|
hits: dict[int, dict[str, Any]] = {}
|
|
sidos: list[dict[str, Any]] = []
|
|
missing: list[str] = []
|
|
for code, path in sorted(tables.items()):
|
|
if path is None:
|
|
missing.append(names.get(code) or code)
|
|
continue
|
|
head, rows = read_table(path)
|
|
located = 0
|
|
for row in rows:
|
|
place = f"{row['sgg']} {row['emd']} {row['ri']} {row['jibun']}".replace(
|
|
" ", " "
|
|
).strip()
|
|
if row["lon"] is not None:
|
|
located += 1
|
|
if frame.contains(Point(to_m(row["lon"], row["lat"]))):
|
|
points.append(
|
|
{
|
|
"lon": round(row["lon"], 7),
|
|
"lat": round(row["lat"], 7),
|
|
"kind": row["kind"],
|
|
"area_m2": row["area_m2"],
|
|
"place": place,
|
|
}
|
|
)
|
|
for parcel in index.get(row_key(row) or ("",), []):
|
|
hit = hits.setdefault(
|
|
id(parcel), {"parcel": parcel, "count": 0, "kinds": set(), "area_m2": 0.0}
|
|
)
|
|
hit["count"] += 1
|
|
hit["area_m2"] += row["area_m2"] or 0.0
|
|
if row["kind"]:
|
|
hit["kinds"].add(row["kind"])
|
|
sidos.append(
|
|
{
|
|
"code": code,
|
|
"name": names.get(code) or path.stem.split("_", 1)[-1],
|
|
"rows": len(rows),
|
|
"located": located,
|
|
"title": head.get("이름"),
|
|
"basis": head.get("기준일"),
|
|
"source": head.get("출처"),
|
|
}
|
|
)
|
|
features = [
|
|
{
|
|
"type": "Feature",
|
|
"geometry": hit["parcel"]["geometry"],
|
|
"properties": {
|
|
"jibun": (hit["parcel"].get("properties") or {}).get("jibun"),
|
|
"count": hit["count"],
|
|
"kinds": sorted(hit["kinds"]),
|
|
"area_m2": round(hit["area_m2"], 1),
|
|
},
|
|
}
|
|
for hit in hits.values()
|
|
]
|
|
return {
|
|
"points": points,
|
|
"parcels": {"type": "FeatureCollection", "features": features},
|
|
"sidos": sidos,
|
|
"missing": missing,
|
|
}
|
|
|
|
|
|
def zone_layer(project_root: Path, cadastral: Path | None, box: Box, epsg: int) -> dict[str, Any]:
|
|
"""산사태취약지역 — 지적 필지 시도의 표로 맞춤 · 지적이 없으면(못 받음) 위경도 있는 표
|
|
모두 점만."""
|
|
parcels = []
|
|
if cadastral is not None:
|
|
parcels = (json.loads(cadastral.read_text(encoding="utf-8")) or {}).get("features") or []
|
|
codes = sorted({str((p.get("properties") or {}).get("sido_cd")) for p in parcels} - {"None"})
|
|
if not codes:
|
|
codes = sorted(path.name.split("_", 1)[0] for path in DATA_DIR.glob("*.csv"))
|
|
tables = {code: table_of(code) for code in codes}
|
|
key = derived_key(
|
|
ZONE_KIND,
|
|
ZONE_VERSION,
|
|
{code: file_stamp(path) for code, path in tables.items()},
|
|
file_stamp(cadastral),
|
|
int(epsg),
|
|
[round(v, 1) for v in box],
|
|
)
|
|
saved = read_derived(project_root, ZONE_KIND, key)
|
|
if saved is not None:
|
|
return saved["value"]
|
|
value = match_zones(parcels, tables, box, epsg)
|
|
value["cadastral"] = cadastral is not None
|
|
write_derived(project_root, ZONE_KIND, key, value)
|
|
return value
|