eomsangdonandClaude Sonnet 5.5
40cecf27a2
- 서버 — 올릴 때마다 묶음 하나(`D03_PlanAnalysis/임도망도/<번호>`) · 옛 한 파일 꼴 = 묶음 legacy · 합친 값 + 묶음 줄 · 목록 GET · 묶음 DELETE · 20 묶음까지 - 화면 — 위치 칸 목록(묶음 줄 · 펼치면 선 줄 · 고름 = 지도 강조 + 맞춤 · 지움 = 물음 뒤) · 색 주황 → 밝은 분홍(`--draw-map-forest-road-file`) - 시험 584_3 — 묶음 쌓기 · 지움 · legacy · 20 묶음 · 화면 정의 Co-Authored-By: Claude Sonnet 5.5 <noreply@anthropic.com>
344 lines
14 KiB
Python
344 lines
14 KiB
Python
"""D 노선분석 지도 올린 임도망도 겹(PLAN 584-3 · 595-4 · 요구 443 · 463).
|
||
|
||
- 출처 = 사용자가 받은 산림청 임도망도 SHP(FGIS 무료신청 · 산림빅데이터거래소 · 시군구 단위 · 선)
|
||
- 올림 = zip 하나 또는 shp · shx · dbf(· prj · cpg) 묶음 · 합계 `MAX_BYTES` · 읽어 보고 자르는 범위
|
||
안 선이 있어야 받음(아니면 옛 것 그대로 · 까닭 = `ValueError` + 기준점 ~ 가장 가까운 임도 거리)
|
||
- 자르는 범위 = 기준점 ± `RADIUS_M` ∪ 받은 지형(`clip_box` · 지형 받기 전에도 기준점만으로)
|
||
- 원본 = 올릴 때마다 묶음 하나 — 프로젝트 `D03_PlanAnalysis/임도망도/<번호>.zip`(낱 파일은 zip 으로
|
||
묶음) + `<번호>.json`(파일 이름 · 올린 때) · 옛 한 파일 꼴(`임도망도.zip` · `.json`)은 묶음
|
||
`legacy` 하나로 읽음 · 지움 = 그 묶음만(PLAN 605-1)
|
||
- 좌표계 = prj · 없으면 EPSG:5179 · dbf 글자 = cpg · 없으면 cp949 · 자르는 범위로 잘라 가닥 · 길이 ·
|
||
잘린 상자(`bounds` 경위도) · 기준점 거리 → 파생 저장본(열쇠 = 원본 지문 · 범위 · 기준점)
|
||
|
||
계산 자리 = ③ 서버 단독.
|
||
"""
|
||
|
||
from __future__ import annotations
|
||
|
||
import io
|
||
import json
|
||
import math
|
||
import os
|
||
import re
|
||
import tempfile
|
||
import uuid
|
||
import zipfile
|
||
from pathlib import Path
|
||
from typing import Any
|
||
|
||
from common_util.common_util_derived_store import (
|
||
derived_key,
|
||
file_stamp,
|
||
read_derived,
|
||
write_derived,
|
||
)
|
||
from common_util.common_util_json import atomic_write_json
|
||
from common_util.common_util_time import utc_now_text
|
||
|
||
Box = tuple[float, float, float, float]
|
||
XY = tuple[float, float]
|
||
|
||
FOLDER = "D03_PlanAnalysis"
|
||
SOURCE_ZIP = "임도망도.zip" # 옛 한 파일 꼴 = 묶음 legacy
|
||
SOURCE_META = "임도망도.json"
|
||
BUNDLE_DIR = "임도망도"
|
||
LEGACY = "legacy"
|
||
MAX_BUNDLES = 20
|
||
DEFAULT_EPSG = 5179
|
||
DBF_ENCODING = "cp949"
|
||
MAX_BYTES = 100 * 1024 * 1024
|
||
SHP_PARTS = (".shp", ".shx", ".dbf", ".prj", ".cpg")
|
||
ROAD_KIND = "plan_forest_road"
|
||
ROAD_VERSION = 3 # 묶음 열쇠
|
||
KEEP = 2 * MAX_BUNDLES # 묶음마다 파생 저장본 한 벌
|
||
RADIUS_M = 5000.0 # 기준점 둘레(요구서_463 3장 18)
|
||
|
||
|
||
def clip_box(origin: XY, have: Box | None) -> Box:
|
||
"""자르는 범위 = 기준점 ± `RADIUS_M` ∪ 받은 지형."""
|
||
x, y = origin
|
||
box = (x - RADIUS_M, y - RADIUS_M, x + RADIUS_M, y + RADIUS_M)
|
||
if have is None:
|
||
return box
|
||
return min(box[0], have[0]), min(box[1], have[1]), max(box[2], have[2]), max(box[3], have[3])
|
||
|
||
|
||
def source_dir(project_root: Path) -> Path:
|
||
return Path(project_root) / FOLDER
|
||
|
||
|
||
def pack(files: list[tuple[str, bytes]]) -> bytes:
|
||
"""올린 파일 → zip 바이트 — zip 하나는 그대로 · 낱 파일은 shp · shx · dbf 가 다 있어야 묶음."""
|
||
if len(files) == 1 and files[0][0].lower().endswith(".zip"):
|
||
return files[0][1]
|
||
parts = [(name, data) for name, data in files if Path(name).suffix.lower() in SHP_PARTS]
|
||
have = {Path(name).suffix.lower() for name, _ in parts}
|
||
lack = [ext for ext in (".shp", ".shx", ".dbf") if ext not in have]
|
||
if lack:
|
||
raise ValueError(
|
||
f"SHP 묶음에 {' · '.join(lack)} 이(가) 없습니다 — "
|
||
"zip 하나 또는 shp · shx · dbf · prj 를 함께 고르세요."
|
||
)
|
||
buffer = io.BytesIO()
|
||
with zipfile.ZipFile(buffer, "w", zipfile.ZIP_DEFLATED) as zf:
|
||
for name, data in parts:
|
||
zf.writestr(Path(name).name, data)
|
||
return buffer.getvalue()
|
||
|
||
|
||
def read_shapes(zip_path: Path) -> list[Any]:
|
||
"""zip 안 SHP 마다 GeoDataFrame(좌표계 = prj · 없으면 5179) — 못 읽으면 `ValueError`."""
|
||
import geopandas as gpd
|
||
|
||
try:
|
||
zf = zipfile.ZipFile(zip_path)
|
||
except zipfile.BadZipFile as error:
|
||
raise ValueError("zip 파일을 열지 못했습니다.") from error
|
||
frames = []
|
||
with zf, tempfile.TemporaryDirectory() as folder:
|
||
members = [n for n in zf.namelist() if Path(n).suffix.lower() in SHP_PARTS]
|
||
if not any(n.lower().endswith(".shp") for n in members):
|
||
raise ValueError("zip 안에 .shp 파일이 없습니다.")
|
||
for n in members: # 폴더 없이 이름만 = 바깥으로 안 풂
|
||
(Path(folder) / Path(n).name).write_bytes(zf.read(n))
|
||
for shp in sorted(Path(folder).glob("*.shp")):
|
||
options = {} if shp.with_suffix(".cpg").is_file() else {"encoding": DBF_ENCODING}
|
||
try:
|
||
frame = gpd.read_file(shp, **options)
|
||
except Exception as error: # noqa: BLE001 — 읽기 엔진 오류 종류가 많음
|
||
raise ValueError(f"{shp.name} 을(를) 읽지 못했습니다 — {error}") from error
|
||
if frame.crs is None:
|
||
frame = frame.set_crs(DEFAULT_EPSG)
|
||
frames.append(frame)
|
||
return frames
|
||
|
||
|
||
def _plain(value: Any) -> Any:
|
||
if isinstance(value, float) and not math.isfinite(value):
|
||
return None
|
||
return value if isinstance(value, (str, int, float, bool)) or value is None else str(value)
|
||
|
||
|
||
def clip_roads(zip_path: Path, box: Box, epsg: int, origin: XY | None = None) -> dict[str, Any]:
|
||
"""SHP 선 → 범위로 잘라 경위도 선(원 속성 + `length_m`) · 가닥 · 길이 · 원본 선 수 · 좌표계 ·
|
||
잘린 상자(`bounds` [서, 남, 동, 북] · 없으면 None) · 기준점 ~ 가장 가까운 임도(`distance_m` ·
|
||
자르기 전 모든 선)."""
|
||
from pyproj import Transformer
|
||
from shapely.geometry import Point, mapping
|
||
from shapely.geometry import box as rect
|
||
from shapely.ops import transform
|
||
|
||
frame = rect(*box)
|
||
here = Point(origin) if origin else None
|
||
to_ll = Transformer.from_crs(f"EPSG:{epsg}", "EPSG:4326", always_xy=True).transform
|
||
features: list[dict[str, Any]] = []
|
||
total, length, crs = 0, 0.0, []
|
||
near: float | None = None
|
||
edges: list[tuple[float, float, float, float]] = []
|
||
for shapes in read_shapes(zip_path):
|
||
crs.append(shapes.crs.to_string())
|
||
shapes = shapes[shapes.geometry.notna()]
|
||
shapes = shapes[shapes.geometry.geom_type.isin(("LineString", "MultiLineString"))]
|
||
total += len(shapes)
|
||
metric = shapes.to_crs(epsg)
|
||
for (_, row), geometry in zip(
|
||
shapes.drop(columns=shapes.geometry.name).iterrows(), metric.geometry
|
||
):
|
||
if here is not None:
|
||
gap = geometry.distance(here)
|
||
near = gap if near is None else min(near, gap)
|
||
part = geometry.intersection(frame)
|
||
if part.is_empty or part.length <= 0:
|
||
continue
|
||
length += part.length
|
||
props = {str(k): _plain(v) for k, v in row.items()}
|
||
props["length_m"] = round(part.length, 1)
|
||
lonlat = transform(to_ll, part)
|
||
edges.append(lonlat.bounds)
|
||
features.append({"type": "Feature", "geometry": mapping(lonlat), "properties": props})
|
||
if not total:
|
||
raise ValueError("SHP 에 선 지물이 없습니다 — 임도망도(선) 파일인지 보세요.")
|
||
return {
|
||
"count": len(features),
|
||
"length_m": round(length, 1),
|
||
"total": total,
|
||
"crs": sorted(set(crs)),
|
||
"bounds": [
|
||
min(e[0] for e in edges),
|
||
min(e[1] for e in edges),
|
||
max(e[2] for e in edges),
|
||
max(e[3] for e in edges),
|
||
]
|
||
if edges
|
||
else None,
|
||
"distance_m": None if near is None else round(near, 1),
|
||
"collection": {"type": "FeatureCollection", "features": features},
|
||
}
|
||
|
||
|
||
def _paths(project_root: Path, bundle: str) -> tuple[Path, Path]:
|
||
"""(zip, 이름표 json) — `legacy` = 옛 한 파일 꼴."""
|
||
folder = source_dir(project_root)
|
||
if bundle == LEGACY:
|
||
return folder / SOURCE_ZIP, folder / SOURCE_META
|
||
return folder / BUNDLE_DIR / f"{bundle}.zip", folder / BUNDLE_DIR / f"{bundle}.json"
|
||
|
||
|
||
def _meta(path: Path) -> dict[str, Any]:
|
||
try:
|
||
return json.loads(path.read_text("utf-8"))
|
||
except (OSError, ValueError):
|
||
return {}
|
||
|
||
|
||
def _name(files: list[str]) -> str:
|
||
"""묶음 줄 이름 — 파일 하나(zip) = 그 이름 · 낱 파일 묶음 = 첫 파일 줄기."""
|
||
return "" if not files else files[0] if len(files) == 1 else Path(files[0]).stem
|
||
|
||
|
||
def bundle_ids(project_root: Path) -> list[str]:
|
||
"""올린 순서(올린 때) — 옛 한 파일 꼴 포함."""
|
||
folder = source_dir(project_root)
|
||
ids = [p.stem for p in (folder / BUNDLE_DIR).glob("*.zip")]
|
||
if (folder / SOURCE_ZIP).is_file():
|
||
ids.append(LEGACY)
|
||
|
||
def when(bundle: str) -> tuple[str, str]:
|
||
zip_path, meta_path = _paths(project_root, bundle)
|
||
return _meta(meta_path).get("uploaded_at", ""), file_stamp(zip_path) or ""
|
||
|
||
return sorted(ids, key=when) # 같은 초면 파일 시각 순
|
||
|
||
|
||
def bundle_layer(
|
||
project_root: Path, bundle: str, box: Box, epsg: int, origin: XY | None = None
|
||
) -> dict[str, Any]:
|
||
"""묶음 하나 — 원본 지문 · 범위 · 기준점 같으면 저장본 · 없으면 `FileNotFoundError`."""
|
||
path, meta = _paths(project_root, bundle)
|
||
stamp = file_stamp(path)
|
||
if stamp is None:
|
||
raise FileNotFoundError(f"올린 임도망도 묶음 {bundle} 이(가) 없습니다.")
|
||
at = [round(v, 1) for v in origin] if origin else None
|
||
key = derived_key(
|
||
ROAD_KIND, ROAD_VERSION, bundle, stamp, int(epsg), [round(v, 1) for v in box], at
|
||
)
|
||
saved = read_derived(project_root, ROAD_KIND, key)
|
||
if saved is not None:
|
||
return saved["value"]
|
||
value = clip_roads(path, box, epsg, origin)
|
||
value["source"] = _meta(meta) or None
|
||
write_derived(project_root, ROAD_KIND, key, value, KEEP)
|
||
return value
|
||
|
||
|
||
def road_layer(project_root: Path, box: Box, epsg: int, origin: XY | None = None) -> dict[str, Any]:
|
||
"""올린 임도망도 전체 — 묶음마다 자른 값을 합침(선 `__bundle` = 묶음 번호 ·
|
||
`__line` = 묶음 안 차례 · `bundles` = 묶음 줄 · 가닥 · 길이 · 원본 선 수 = 합 ·
|
||
기준점 거리 = 가장 가까운 것) · 올린 적 없으면 `FileNotFoundError`."""
|
||
ids = bundle_ids(project_root)
|
||
if not ids:
|
||
raise FileNotFoundError(
|
||
"올린 임도망도가 없습니다 — [임도망 올리기]로 시군구 SHP 를 올리세요."
|
||
)
|
||
rows, features, crs, edges, gaps = [], [], set(), [], []
|
||
for bundle in ids:
|
||
got = bundle_layer(project_root, bundle, box, epsg, origin)
|
||
source = got.get("source") or {}
|
||
files = source.get("files") or []
|
||
rows.append(
|
||
{
|
||
"id": bundle,
|
||
"name": source.get("name") or _name(files) or bundle,
|
||
"files": files,
|
||
"bytes": source.get("bytes"),
|
||
"uploaded_at": source.get("uploaded_at"),
|
||
"count": got["count"],
|
||
"length_m": got["length_m"],
|
||
"total": got["total"],
|
||
"bounds": got["bounds"],
|
||
"distance_m": got["distance_m"],
|
||
}
|
||
)
|
||
crs.update(got["crs"])
|
||
if got["bounds"]:
|
||
edges.append(got["bounds"])
|
||
if got["distance_m"] is not None:
|
||
gaps.append(got["distance_m"])
|
||
for line, feature in enumerate(got["collection"]["features"]):
|
||
props = {**feature["properties"], "__bundle": bundle, "__line": line}
|
||
features.append({**feature, "properties": props})
|
||
return {
|
||
"count": sum(r["count"] for r in rows),
|
||
"length_m": round(sum(r["length_m"] for r in rows), 1),
|
||
"total": sum(r["total"] for r in rows),
|
||
"crs": sorted(crs),
|
||
"bounds": [
|
||
min(e[0] for e in edges),
|
||
min(e[1] for e in edges),
|
||
max(e[2] for e in edges),
|
||
max(e[3] for e in edges),
|
||
]
|
||
if edges
|
||
else None,
|
||
"distance_m": min(gaps) if gaps else None,
|
||
"source": {k: rows[-1][k] for k in ("files", "bytes", "uploaded_at")},
|
||
"bundles": rows,
|
||
"collection": {"type": "FeatureCollection", "features": features},
|
||
}
|
||
|
||
|
||
def save_roads(
|
||
project_root: Path,
|
||
files: list[tuple[str, bytes]],
|
||
box: Box,
|
||
epsg: int,
|
||
origin: XY | None = None,
|
||
) -> dict[str, Any]:
|
||
"""올림 — 묶어 읽어 보고 자르는 범위 안 선이 있으면 새 묶음으로 쌓음 · 아니면 `ValueError`
|
||
(쌓인 것 그대로 · 기준점 ~ 가장 가까운 임도 거리)."""
|
||
if len(bundle_ids(project_root)) >= MAX_BUNDLES:
|
||
raise ValueError(f"올린 묶음이 {MAX_BUNDLES} 개입니다 — 하나를 지우고 올리세요.")
|
||
data = pack(files)
|
||
bundle = uuid.uuid4().hex[:12]
|
||
zip_path, meta_path = _paths(project_root, bundle)
|
||
zip_path.parent.mkdir(parents=True, exist_ok=True)
|
||
with tempfile.NamedTemporaryFile(
|
||
dir=zip_path.parent, suffix=".zip.new", delete=False
|
||
) as temporary:
|
||
temporary.write(data)
|
||
trial = Path(temporary.name)
|
||
try:
|
||
got = clip_roads(trial, box, epsg, origin)
|
||
if not got["count"]:
|
||
gap = got["distance_m"]
|
||
where = f" — 가장 가까운 임도 {gap / 1000:.1f} km" if gap is not None else ""
|
||
raise ValueError(
|
||
f"임도가 기준점 ±{RADIUS_M / 1000:.0f} km · 받은 지형 밖뿐입니다{where}"
|
||
f"(파일 선 {got['total']}가닥) — 사업지 시군구 파일인지 보세요."
|
||
)
|
||
os.replace(trial, zip_path)
|
||
finally:
|
||
trial.unlink(missing_ok=True)
|
||
names = [Path(name).name for name, _ in files]
|
||
atomic_write_json(
|
||
meta_path,
|
||
{
|
||
"files": names,
|
||
"name": _name(names),
|
||
"bytes": len(data),
|
||
"uploaded_at": utc_now_text(),
|
||
},
|
||
)
|
||
return road_layer(project_root, box, epsg, origin)
|
||
|
||
|
||
def delete_bundle(project_root: Path, bundle: str) -> None:
|
||
"""묶음 하나 지움(저장본은 열쇠가 달라져 저절로 안 쓰임) — 없으면 `FileNotFoundError`."""
|
||
if bundle != LEGACY and not re.fullmatch(r"[0-9a-f]{12}", bundle):
|
||
raise FileNotFoundError(f"올린 임도망도 묶음 {bundle} 이(가) 없습니다.")
|
||
zip_path, meta_path = _paths(project_root, bundle)
|
||
if not zip_path.is_file():
|
||
raise FileNotFoundError(f"올린 임도망도 묶음 {bundle} 이(가) 없습니다.")
|
||
zip_path.unlink()
|
||
meta_path.unlink(missing_ok=True)
|