fix(B04 도엽): 도곽선 N3L_H0010000 읽기(없으면 옛 방식) · 공용 창고 prune 막음 (PLAN 611-1 · 요구 486 · 493)

Co-Authored-By: Claude Sonnet 5.5 <noreply@anthropic.com>
This commit is contained in:
2026-10-30 22:52:00 +09:00
co-authored by Claude Sonnet 5.5
parent 0a54d52fff
commit 32b2d82790
2 changed files with 96 additions and 4 deletions
@@ -39,9 +39,11 @@ from .B04_PreProcess_Engine_MapSheet import latlon_to_sheet5k, sheet5k_to_bounds
logger = logging.getLogger(__name__)
# 도곽선 레이어 코드 (수치지형도 v2.0 도엽본)
_SHEET_FRAME_CODE = "A0010000"
# 도곽선 레이어 코드 (수치지형도 v2.0 도엽본) — 도곽선 N3L_H0010000 이 없는 zip 은 옛 A0010000(도로 면)로 대신
_SHEET_FRAME_CODE = "N3L_H0010000"
_SHEET_FRAME_FALLBACK_CODE = "A0010000"
_SHEET_NO_RE = re.compile(r"(\d{8})")
_SHARED_DIRNAME = "_shared" # storage/_shared 아래 = 여러 프로젝트가 같이 쓰는 창고
# 도엽본 좌표계 후보: 동부원점(5187)이 기본, 오라벨 zip은 중부원점(5186) 선언 사례 실측됨
_CRS_CANDIDATES = ("EPSG:5187", "EPSG:5186", "EPSG:5185", "EPSG:5188")
@@ -240,6 +242,9 @@ def prune_sheets(store_dir: str | Path, keep_sheet_nos: list[str]) -> list[str]:
store = Path(store_dir)
if not store.is_dir():
return []
if _SHARED_DIRNAME in store.resolve().parts:
logger.info("도엽 정리: 공용 창고(%s)는 지우지 않음", store)
return []
keep = {str(sheet_no) for sheet_no in keep_sheet_nos}
index = _load_index(store)
removed: list[str] = []
@@ -313,7 +318,7 @@ def ingest_zip(store_dir: str | Path, src_path: str | Path, sheet_no: str | None
def _read_frame_bounds_wgs84(zip_path: Path) -> tuple[tuple[float, float, float, float], str]:
"""zip 내 도곽선(A0010000) SHP bounds를 WGS84로 반환. (bounds, 사용된 CRS).
"""zip 내 도곽선(N3L_H0010000 · 없으면 A0010000) SHP bounds를 WGS84로 반환. (bounds, 사용된 CRS).
선언된 PRJ가 오라벨(중부/동부원점 뒤바뀜)인 실측 사례가 있어, 파일명 도엽번호
기준 예측 도곽과 대조해 정합한 CRS를 자동 판별한다.
@@ -322,7 +327,11 @@ def _read_frame_bounds_wgs84(zip_path: Path) -> tuple[tuple[float, float, float,
with zipfile.ZipFile(zip_path) as zf:
names = zf.namelist()
shp = [n for n in names if n.upper().endswith(f"{_SHEET_FRAME_CODE}.SHP")]
shp: list[str] = []
for code in (_SHEET_FRAME_CODE, _SHEET_FRAME_FALLBACK_CODE):
shp = [n for n in names if n.upper().endswith(f"{code}.SHP")]
if shp:
break
if not shp:
raise ValueError(f"{zip_path.name}: 도곽선({_SHEET_FRAME_CODE}) SHP 없음")
base = shp[0][:-4]
@@ -0,0 +1,83 @@
"""611-1 — 도곽선 N3L_H0010000 읽기(도로 없는 도엽 zip · 옛 zip 대신) · 공용 창고 prune 안 지움."""
import sys
import tempfile
import zipfile
from pathlib import Path
import geopandas as gpd
import pytest
from shapely.geometry import box
sys.path.insert(0, str(Path(__file__).resolve().parents[2]))
from B04_PreProcess.B04_PreProcess_Engine_MapSheet import sheet5k_to_bounds # noqa: E402
from B04_PreProcess.B04_PreProcess_Engine_SheetStore import ( # noqa: E402
_read_frame_bounds_wgs84,
ingest_zip,
prune_sheets,
)
SHEET = "36712079"
CRS = "EPSG:5186"
def _make_zip(path: Path, layers: dict[str, tuple[float, float, float, float]]) -> None:
"""layers: {레이어 코드: WGS84 box} — SHP 로 써서 zip 에 넣음."""
with tempfile.TemporaryDirectory() as td, zipfile.ZipFile(path, "w") as zf:
for code, bb in layers.items():
gdf = gpd.GeoDataFrame(geometry=[box(*bb)], crs="EPSG:4326").to_crs(CRS)
shp = Path(td) / f"{SHEET}_{code}.shp"
gdf.to_file(shp)
for f in Path(td).glob(f"{SHEET}_{code}.*"):
zf.write(f, f.name)
def _frame():
return sheet5k_to_bounds(SHEET)
def _road_part():
w, s, e, n = _frame()
return (w + (e - w) * 0.2, s + (n - s) * 0.2, w + (e - w) * 0.5, s + (n - s) * 0.5)
def test_도곽_겹이_있으면_도곽선_범위(tmp_path):
z = tmp_path / f"{SHEET}.zip"
_make_zip(z, {"N3L_H0010000": _frame(), "N3L_A0010000": _road_part()})
b, _ = _read_frame_bounds_wgs84(z)
for got, want in zip(b, _frame()):
assert got == pytest.approx(want, abs=1e-6)
def test_도로_없는_도엽도_받기_통과(tmp_path):
z = tmp_path / f"{SHEET}.zip"
_make_zip(z, {"N3L_H0010000": _frame()})
entry = ingest_zip(tmp_path / "store", z, sheet_no=SHEET)
assert entry["bounds_wgs84"][0] == pytest.approx(_frame()[0], abs=1e-6)
def test_도곽_겹_없으면_옛_방식(tmp_path):
z = tmp_path / f"{SHEET}.zip"
_make_zip(z, {"N3L_A0010000": _road_part()})
b, _ = _read_frame_bounds_wgs84(z)
for got, want in zip(b, _road_part()):
assert got == pytest.approx(want, abs=1e-6)
def test_공용_창고는_prune_안_지움(tmp_path):
shared = tmp_path / "_shared" / "sheets" / SHEET
shared.mkdir(parents=True)
z = shared / "36712078.zip"
_make_zip(z, {"N3L_H0010000": sheet5k_to_bounds("36712078")})
assert prune_sheets(shared, [SHEET]) == []
assert z.exists()
def test_프로젝트_창고는_prune_지움(tmp_path):
store = tmp_path / "proj" / "map_sheets"
store.mkdir(parents=True)
z = store / "36712078.zip"
_make_zip(z, {"N3L_H0010000": sheet5k_to_bounds("36712078")})
assert prune_sheets(store, [SHEET]) == ["36712078"]
assert not z.exists()