269 lines
11 KiB
Python
269 lines
11 KiB
Python
"""도엽 표시본 잘라 두기 — 평면도 = 노선 띠 · 배수유역도 = 유역 + 여유(509-1 · 509-6 · 서버 단독 ③).
|
||
|
||
도엽 사본은 약 6.7×8.3 km 라 노선 편집 · 배수유역도가 통째로 받으면 그릴 꼭짓점의 9 할이
|
||
화면에 안 드는 헛일이다(요구 316). 화면마다 **보는 범위**만 잘라 보낸다 — 형상은 그대로 ·
|
||
범위 밖만 사라짐. 분석(배수유역 · 관 자리)은 원본 도엽을 읽으므로 이 사본과 상관없다.
|
||
|
||
- 범위(사업지 m · 바깥으로 `SNAP_M` 눈금에 맞춤 → 노선을 조금 옮겨도 같은 열쇠)
|
||
- `route` = 저장 노선 bbox + 띠 `ROUTE_BAND_M` + 여유 `ROUTE_SPARE_M` — 편집이 이 밖으로
|
||
나가면 화면이 놓은 뒤 편집 중 노선 bbox(`route_bounds` · 508-2)로 다시 받는다
|
||
(`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 parse_route_bbox(text: str | None) -> Rect | None:
|
||
"""화면이 보낸 편집 중 노선 bbox(`x0,y0,x1,y1` m) — 숫자 넷 · 차례 맞음만 · 아니면 None."""
|
||
try:
|
||
values = [float(part) for part in (text or "").split(",")]
|
||
except ValueError:
|
||
return None
|
||
if len(values) != 4 or not all(math.isfinite(v) for v in values):
|
||
return None
|
||
if values[0] > values[2] or values[1] > values[3]:
|
||
return None
|
||
return values[0], values[1], values[2], values[3]
|
||
|
||
|
||
def clip_rect_m(
|
||
project_root: Path, scope: str, meta: dict[str, Any], route_bounds: Rect | None = None
|
||
) -> Rect | None:
|
||
"""범위 사각형(사업지 m · 눈금 맞춤). 노선이 없으면(또는 배수유역도에 유역이 없으면) None.
|
||
`route_bounds` = 저장 노선 대신 쓸 노선 bbox(편집 중 · 508-2)."""
|
||
route = route_bounds or _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, route_bounds: Rect | None = None
|
||
) -> Path:
|
||
"""`scope` 범위로 자른 표시본 경로. 범위를 못 정하거나 실패하면 `source` 그대로.
|
||
열쇠 = 사본 지문 + 범위 — 같은 범위면 노선을 어디서 재든 같은 저장본(다시 안 셈)."""
|
||
try:
|
||
meta = _read_meta(project_root / "B04_PreProcess" / "processed")
|
||
rect_m = (
|
||
clip_rect_m(project_root, scope, meta, route_bounds)
|
||
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
|