"""도엽 표시본 잘라 두기 — 평면도 = 노선 띠 · 배수유역도 = 유역 + 여유(509-1 · 509-6 · 서버 단독 ③). 도엽 사본은 약 6.7×8.3 km 라 노선 편집 · 배수유역도가 통째로 받으면 그릴 꼭짓점의 9 할이 화면에 안 드는 헛일이다(요구 316). 화면마다 **보는 범위**만 잘라 보낸다 — 형상은 그대로 · 범위 밖만 사라짐. 분석(배수유역 · 관 자리)은 원본 도엽을 읽으므로 이 사본과 상관없다. - 범위(사업지 m · 바깥으로 `SNAP_M` 눈금에 맞춤 → 노선을 조금 옮겨도 같은 열쇠) - `route` = 저장 노선 bbox + 띠 `ROUTE_BAND_M` + 여유 `ROUTE_SPARE_M` — 편집이 이 밖으로 나가면 화면이 통째 사본을 받는다(`clip_lonlat` 으로 판정). - `drainage` = 세부 유역(노선 교차 관으로 모이는 유역) ∪ 줌아웃 한계(노선 bbox ×2) + `DRAIN_SPARE_M`. - 파생 저장본 = `_derived/sheet_<범위>/<열쇠>.json`(GeoJSON 그대로 · 열쇠 = 사본 지문 + 범위) · 지워도 다시 셈 · 범위를 못 정하면 사본 그대로. - 경위도 ↔ m 는 화면과 같은 가정(위성 메타의 m 사각형 = 경위도 사각형 · `metricToScreen`). """ from __future__ import annotations import json import logging import math import os import tempfile from pathlib import Path from typing import Any from shapely import clip_by_rect from shapely.geometry import mapping, shape from common_util.common_util_derived_store import ( derived_dir, derived_key, file_stamp, prune_derived, ) from common_util.common_util_json import LONLAT_DIGITS, round_floats from common_util.common_util_route_geometry import ( expected_route_csv_path, planned_route_initial_path, planned_route_working_path, read_planned_route_csv, ) from config.config_system import DRAINAGE_CACHE_DIRNAME, DRAINAGE_DETAIL_FILENAME logger = logging.getLogger(__name__) SCOPES = ("route", "drainage") CLIP_VERSION = 1 ROUTE_BAND_M = 300.0 # = B05 `CONTOUR_BAND_M`(등고선을 보일 노선 둘레 띠) ROUTE_SPARE_M = 300.0 # 띠 밖 여유 — 노선을 이만큼 옮겨도 다시 안 받음 DRAIN_SPARE_M = 300.0 # 유역 · 줌아웃 한계 밖 여유 — 화면 가로세로 비 · 테두리 ZOOM_OUT_ROUTE_FACTOR = 2.0 # = B05 `ZOOM_OUT_ROUTE_FACTOR`(노선 범위 몇 배까지 멀리 보나) SNAP_M = 50.0 Rect = tuple[float, float, float, float] def _read_meta(processed: Path) -> dict[str, Any] | None: for name in ("vworld_satellite_meta.json", "vworld_meta.json"): try: meta = json.loads((processed / name).read_text(encoding="utf-8")) except (OSError, ValueError): continue keys = ("x_min", "y_min", "width_meters", "height_meters") if all( meta.get(k) is not None for k in keys + ("lon_min", "lon_max", "lat_min", "lat_max") ): return meta return None def _to_lonlat(meta: dict[str, Any], x: float, y: float) -> tuple[float, float]: lon = meta["lon_min"] + (x - meta["x_min"]) / (meta["width_meters"] or 1) * ( meta["lon_max"] - meta["lon_min"] ) lat = meta["lat_min"] + (y - meta["y_min"]) / (meta["height_meters"] or 1) * ( meta["lat_max"] - meta["lat_min"] ) return lon, lat def _to_metric(meta: dict[str, Any], lon: float, lat: float) -> tuple[float, float]: x = ( meta["x_min"] + (lon - meta["lon_min"]) / ((meta["lon_max"] - meta["lon_min"]) or 1) * (meta["width_meters"]) ) y = ( meta["y_min"] + (lat - meta["lat_min"]) / ((meta["lat_max"] - meta["lat_min"]) or 1) * (meta["height_meters"]) ) return x, y def _bounds(points: list[tuple[float, float]]) -> Rect | None: if not points: return None xs = [p[0] for p in points] ys = [p[1] for p in points] return min(xs), min(ys), max(xs), max(ys) def _grow(rect: Rect, dx: float, dy: float | None = None) -> Rect: dy = dx if dy is None else dy return rect[0] - dx, rect[1] - dy, rect[2] + dx, rect[3] + dy def _union(a: Rect, b: Rect | None) -> Rect: if b is None: return a return min(a[0], b[0]), min(a[1], b[1]), max(a[2], b[2]), max(a[3], b[3]) def _route_points(project_root: Path) -> list[tuple[float, float]]: """저장 노선(`read_route_plan` planned 와 같은 차례 — 수정본 · 초기 · 예상노선).""" for path in ( planned_route_working_path(project_root), planned_route_initial_path(project_root), expected_route_csv_path(project_root), ): route = read_planned_route_csv(path) if path.is_file() else None if route is not None and len(route.vertices) >= 2: return [(float(v.x), float(v.y)) for v in route.vertices] return [] def _basin_points(project_root: Path, meta: dict[str, Any]) -> list[tuple[float, float]]: """세부 유역 꼭짓점(m) — 관마다 모이는 유역(B05 배수유역도가 그리는 것).""" path = project_root / "B04_PreProcess" / DRAINAGE_CACHE_DIRNAME / DRAINAGE_DETAIL_FILENAME try: features = json.loads(path.read_text(encoding="utf-8")).get("features") or [] except (OSError, ValueError): return [] points: list[tuple[float, float]] = [] for feature in features: try: lon0, lat0, lon1, lat1 = shape(feature["geometry"]).bounds except Exception: # 깨진 피처 하나가 범위를 막지 않음 continue points += [_to_metric(meta, lon0, lat0), _to_metric(meta, lon1, lat1)] return points def clip_rect_m(project_root: Path, scope: str, meta: dict[str, Any]) -> Rect | None: """범위 사각형(사업지 m · 눈금 맞춤). 노선이 없으면(또는 배수유역도에 유역이 없으면) None.""" route = _bounds(_route_points(project_root)) if route is None: return None if scope == "route": rect = _grow(route, ROUTE_BAND_M + ROUTE_SPARE_M) else: basins = _bounds(_basin_points(project_root, meta)) if basins is None: return None half = (ZOOM_OUT_ROUTE_FACTOR - 1) / 2 zoom_out = _grow(route, (route[2] - route[0]) * half, (route[3] - route[1]) * half) rect = _grow(_union(zoom_out, basins), DRAIN_SPARE_M) steps = [round(value / SNAP_M, 6) for value in rect] # 경위도 되셈의 끝자리 흔들림 걷음 return ( math.floor(steps[0]) * SNAP_M, math.floor(steps[1]) * SNAP_M, math.ceil(steps[2]) * SNAP_M, math.ceil(steps[3]) * SNAP_M, ) def _coord_bounds(coords: Any) -> Rect: """GeoJSON 좌표(중첩 목록)의 bbox — shapely 를 만들기 전 거르기.""" stack = [coords] x0 = y0 = math.inf x1 = y1 = -math.inf while stack: item = stack.pop() if item and isinstance(item[0], (int, float)): x, y = item[0], item[1] x0, y0, x1, y1 = min(x0, x), min(y0, y), max(x1, x), max(y1, y) else: stack.extend(item) return x0, y0, x1, y1 def clip_collection(collection: dict[str, Any], rect: Rect) -> dict[str, Any]: """사각형(경위도) 밖을 걷은 컬렉션. 안에 다 든 피처는 그대로 · 걸친 것만 잘라 냄.""" west, south, east, north = rect features = [] for feature in collection.get("features") or []: geometry = feature.get("geometry") if not geometry or not geometry.get("coordinates"): continue x0, y0, x1, y1 = _coord_bounds(geometry["coordinates"]) if x1 < west or x0 > east or y1 < south or y0 > north: continue if not (x0 >= west and x1 <= east and y0 >= south and y1 <= north): cut = clip_by_rect(shape(geometry), *rect) # 점은 bbox 로 이미 갈림 if cut.is_empty: continue geometry = round_floats(mapping(cut), LONLAT_DIGITS) features.append({**feature, "geometry": geometry}) return {**collection, "features": features, "clip_lonlat": [round(v, 7) for v in rect]} def _write_compact(path: Path, value: Any) -> None: path.parent.mkdir(parents=True, exist_ok=True) handle, temporary = tempfile.mkstemp(dir=path.parent, prefix=f".{path.name}.", suffix=".tmp") try: with os.fdopen(handle, "w", encoding="utf-8") as file: json.dump(value, file, ensure_ascii=False, separators=(",", ":")) os.replace(temporary, path) except BaseException: Path(temporary).unlink(missing_ok=True) raise def clipped_sheet(project_root: Path, source: Path, scope: str) -> Path: """`scope` 범위로 자른 표시본 경로. 범위를 못 정하거나 실패하면 `source` 그대로.""" try: meta = _read_meta(project_root / "B04_PreProcess" / "processed") rect_m = clip_rect_m(project_root, scope, meta) if meta and scope in SCOPES else None if rect_m is None or meta is None: return source west, south = _to_lonlat(meta, rect_m[0], rect_m[1]) east, north = _to_lonlat(meta, rect_m[2], rect_m[3]) kind = f"sheet_{scope}" key = derived_key(CLIP_VERSION, source.name, file_stamp(source), rect_m) target = derived_dir(project_root, kind) / f"{key}.json" if target.is_file(): return target collection = json.loads(source.read_text(encoding="utf-8")) _write_compact(target, clip_collection(collection, (west, south, east, north))) prune_derived(project_root, kind, keep=4) # 레이어 둘 × 두 벌 logger.info( "도엽 %s 범위 사본: %s (%.1fMB → %.1fMB)", scope, source.name, source.stat().st_size / 1e6, target.stat().st_size / 1e6, ) return target except Exception as exc: # 자르기 실패가 화면을 막으면 안 됨 — 통째 사본 logger.warning("도엽 %s 범위 사본 실패 (%s): %s", scope, source.name, exc) return source