eomsangdonandClaude Fable 5
f7528a4aa4
- 폴더·내부 파일 51개 접두사 개명 (git mv, 이력 보존) - 저장소 전체 참조 치환 67파일: import 경로, 라우트 슬러그(b04-preprocess), 라우트 키(B04_PREPROCESS), storage 경로 상수, locale, SQL 주석 - 로직 변경 없음 (기계적 치환). typecheck·백엔드 import 검증 통과 Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
128 lines
5.0 KiB
Python
128 lines
5.0 KiB
Python
# B04_PreProcess_Engine_SheetParser.py
|
|
# 1:5,000 수치지형도 도엽 zip 파서 + 9매 병합 → 레이어별 GeoJSON 산출.
|
|
#
|
|
# 레이어 코드 (보유 도엽 2매 실측 확정, 2026-07-26):
|
|
# N3L_E0020000 하천중심선(구분: 세류/소하천/지방하천), N3P_F0020000 표고점(수치),
|
|
# N3L_F0030000 성/절토 비탈면, N3L_F0040000 옹벽·석축, N3P_E0042326 유수방향,
|
|
# N3P_H0020000 기준점(삼각점/통합기준점, 표고), N3P_H0040000 지명(계곡·부락 등)
|
|
# 벼랑바위·너덜바위·산산맥은 표본 도엽에 미출현 — 코드 확정 시 SHEET_LAYERS에 추가.
|
|
#
|
|
# 좌표계: zip 내부 PRJ는 오라벨 사례가 있어 map_sheets_index.json의 교정 crs를 사용한다.
|
|
|
|
from __future__ import annotations
|
|
|
|
import logging
|
|
import tempfile
|
|
import zipfile
|
|
from pathlib import Path
|
|
|
|
import geopandas as gpd
|
|
import pandas as pd
|
|
|
|
from .B04_PreProcess_Engine_SheetStore import _load_index, get_sheet_path
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
# 레이어 키 → (지오메트리 접두사, 지형지물 코드)
|
|
SHEET_LAYERS: dict[str, tuple[str, str]] = {
|
|
"등고선": ("N3L", "F0010000"),
|
|
"하천중심선": ("N3L", "E0020000"),
|
|
"표고점": ("N3P", "F0020000"),
|
|
"성절토": ("N3L", "F0030000"),
|
|
"옹벽석축": ("N3L", "F0040000"),
|
|
"유수방향": ("N3P", "E0042326"),
|
|
"기준점": ("N3P", "H0020000"),
|
|
"지명": ("N3P", "H0040000"),
|
|
}
|
|
|
|
# 산출 파일 접두사 (기존 processed/의 국가 GIS geojson과 구분)
|
|
_OUTPUT_PREFIX = "도엽"
|
|
|
|
|
|
def _read_layer_from_zip(
|
|
zip_path: Path, prefix: str, code: str, crs: str
|
|
) -> gpd.GeoDataFrame | None:
|
|
"""도엽 zip에서 단일 레이어 SHP를 읽어 WGS84 GeoDataFrame으로 반환. 없으면 None."""
|
|
member_base = f"{prefix}_{code}"
|
|
with zipfile.ZipFile(zip_path) as zf:
|
|
names = zf.namelist()
|
|
wanted = {
|
|
n
|
|
for n in names
|
|
if Path(n).stem.upper() == member_base
|
|
and Path(n).suffix.lower() in (".shp", ".shx", ".dbf", ".prj", ".cpg")
|
|
}
|
|
if not any(n.lower().endswith(".shp") for n in wanted):
|
|
return None
|
|
with tempfile.TemporaryDirectory() as td:
|
|
for n in wanted:
|
|
(Path(td) / Path(n).name).write_bytes(zf.read(n))
|
|
gdf = gpd.read_file(Path(td) / f"{member_base}.shp")
|
|
|
|
if gdf.empty:
|
|
return None
|
|
# 내부 PRJ 오라벨 대비: 인덱스의 교정 crs로 강제 지정
|
|
gdf = gdf.set_crs(crs, allow_override=True)
|
|
return gdf.to_crs("EPSG:4326")
|
|
|
|
|
|
def parse_and_merge_sheets(
|
|
store_dir: str | Path,
|
|
sheet_nos: list[str],
|
|
output_dir: str | Path,
|
|
layers: list[str] | None = None,
|
|
) -> dict:
|
|
"""도엽 여러 매를 레이어별로 병합해 GeoJSON으로 저장한다.
|
|
|
|
- 도곽 경계 중복 지물은 UFID 기준 dedup (UFID 없으면 지오메트리 WKB 기준).
|
|
- 산출: {output_dir}/도엽_{레이어}.geojson (EPSG:4326)
|
|
- 반환: {레이어: {"count": n, "file": 경로}} (지물 0건 레이어는 제외)
|
|
"""
|
|
store = Path(store_dir)
|
|
out = Path(output_dir)
|
|
out.mkdir(parents=True, exist_ok=True)
|
|
index = _load_index(store)["sheets"]
|
|
target_layers = layers or list(SHEET_LAYERS)
|
|
|
|
result: dict[str, dict] = {}
|
|
for layer_key in target_layers:
|
|
if layer_key not in SHEET_LAYERS:
|
|
logger.warning("도엽 파서: 미정의 레이어 요청 무시: %s", layer_key)
|
|
continue
|
|
prefix, code = SHEET_LAYERS[layer_key]
|
|
|
|
parts: list[gpd.GeoDataFrame] = []
|
|
for sheet_no in sheet_nos:
|
|
sheet_no = str(sheet_no)
|
|
zip_path = get_sheet_path(store, sheet_no)
|
|
entry = index.get(sheet_no)
|
|
if zip_path is None or entry is None:
|
|
logger.warning("도엽 파서: 영구저장소에 없는 도엽 스킵: %s", sheet_no)
|
|
continue
|
|
try:
|
|
gdf = _read_layer_from_zip(zip_path, prefix, code, entry["crs"])
|
|
except Exception as exc:
|
|
logger.warning("도엽 파서: %s %s 읽기 실패: %s", sheet_no, layer_key, exc)
|
|
continue
|
|
if gdf is None:
|
|
continue
|
|
gdf["도엽번호"] = sheet_no
|
|
parts.append(gdf)
|
|
|
|
if not parts:
|
|
continue
|
|
|
|
merged = gpd.GeoDataFrame(pd.concat(parts, ignore_index=True), crs="EPSG:4326")
|
|
# 도곽 경계에 걸친 지물은 인접 도엽에 중복 수록됨 → UFID 우선 dedup
|
|
if "UFID" in merged.columns and merged["UFID"].notna().any():
|
|
merged = merged.drop_duplicates(subset="UFID", keep="first")
|
|
else:
|
|
merged = merged.loc[~merged.geometry.to_wkb().duplicated(keep="first")]
|
|
|
|
out_path = out / f"{_OUTPUT_PREFIX}_{layer_key}.geojson"
|
|
merged.to_file(out_path, driver="GeoJSON")
|
|
result[layer_key] = {"count": int(len(merged)), "file": str(out_path)}
|
|
logger.info("도엽 병합 완료: %s %d건 → %s", layer_key, len(merged), out_path.name)
|
|
|
|
return result
|