# B04_PreProcess_Engine_Extent.py # 전처리에서 내려받을 범위와 기준 좌표를 정한다. # # 배경(2026-08-01 사용자 지시): 배경 지도·수치지형도 도엽의 기준을 라이다 범위 한가운데로 # 잡으면, 계획노선이 라이다 범위를 벗어날 때 배경과 도엽이 노선을 덮지 못한다. # 계획노선(B03 CSV)의 시점·종점을 기준으로 삼고, 없을 때만 라이다 범위로 되돌린다. from __future__ import annotations import json import logging from pathlib import Path from typing import Any logger = logging.getLogger(__name__) _ROUTE_CSV_GLOB = "B03_FileInput/input/csv/*.csv" def read_planned_route(project_root: Path) -> dict[str, Any] | None: """B03 계획노선 CSV를 읽어 좌표계·범위·시점·종점을 돌려준다. 없거나 형식이 어긋나면 None.""" from B03_FileInput.B03_FileInput_Engine_Analyze import analyze_planned_route_csv for csv_path in sorted(project_root.glob(_ROUTE_CSV_GLOB)): try: return analyze_planned_route_csv(csv_path) except (OSError, ValueError) as exc: logger.warning("B04 계획노선 CSV 해석 실패: %s (%s)", csv_path.name, exc) return None def _to_target_crs( points: list[tuple[float, float]], source_epsg: int | None, target_epsg: str ) -> list[tuple[float, float]]: """계획노선 좌표를 라이다 좌표계로 옮긴다. 좌표계가 같거나 알 수 없으면 그대로 쓴다.""" if source_epsg is None: return points source = f"EPSG:{source_epsg}" if source.upper() == target_epsg.upper(): return points from pyproj import Transformer transformer = Transformer.from_crs(source, target_epsg, always_xy=True) return [transformer.transform(x, y) for x, y in points] def download_extent( project_root: Path, las_bounds: dict[str, list[float]], target_epsg: str, ) -> dict[str, list[float]]: """배경 지도가 반드시 덮어야 할 범위 = 라이다 범위 ∪ 계획노선 범위. 여유폭은 여기서 미터로 더하지 않는다. 내려받기 쪽에서 이 범위를 덮는 타일을 정한 뒤 바깥으로 `SURFACE_MAP_MARGIN_TILES` 겹만큼 주변 셀을 더 받는다(2026-08-01 사용자 지시). las_bounds/반환값 모두 {"x": [최소, 최대], "y": [...], "z": [...]} 꼴(라이다 좌표계). """ x_min, x_max = float(las_bounds["x"][0]), float(las_bounds["x"][1]) y_min, y_max = float(las_bounds["y"][0]), float(las_bounds["y"][1]) route = read_planned_route(project_root) if route: bounds = route["bounds"] corners = [ (float(bounds["x_min"]), float(bounds["y_min"])), (float(bounds["x_max"]), float(bounds["y_max"])), ] try: moved = _to_target_crs(corners, route.get("epsg"), target_epsg) except Exception as exc: logger.warning("B04 계획노선 좌표 변환 실패 — 라이다 범위만 사용 (%s)", exc) moved = [] for x, y in moved: x_min, x_max = min(x_min, x), max(x_max, x) y_min, y_max = min(y_min, y), max(y_max, y) return { "x": [x_min, x_max], "y": [y_min, y_max], "z": list(las_bounds.get("z", [0.0, 0.0])), } def satellite_extent( project_root: Path, las_bounds: dict[str, list[float]], target_epsg: str, ) -> dict[str, list[float]]: """배경 지도를 확보할 범위 = 계획노선이 걸치는 **기준 도엽**의 도곽 범위. 수치지형도 도엽과 같은 눈금으로 맞춘다 — 도엽 1매면 1매 크기, 2~3매에 걸치면 그만큼. 주변 도엽은 받지 않는다(2026-08-01 사용자 지시). 도엽 번호를 얻지 못하면 라이다∪계획노선 범위로 되돌아간다. """ from pyproj import Transformer from .B04_PreProcess_Engine_MapSheet import sheet5k_to_bounds, sheets_for_points points = sheet_reference_points_wgs84(project_root, las_bounds, target_epsg) sheets = sheets_for_points(points) if not sheets: return download_extent(project_root, las_bounds, target_epsg) lon_min = lat_min = float("inf") lon_max = lat_max = float("-inf") for sheet_no in sheets: s_lon_min, s_lat_min, s_lon_max, s_lat_max = sheet5k_to_bounds(sheet_no) lon_min, lon_max = min(lon_min, s_lon_min), max(lon_max, s_lon_max) lat_min, lat_max = min(lat_min, s_lat_min), max(lat_max, s_lat_max) to_target = Transformer.from_crs("EPSG:4326", target_epsg, always_xy=True) x0, y0 = to_target.transform(lon_min, lat_min) x1, y1 = to_target.transform(lon_max, lat_max) logger.info("B04 배경 지도 기준 도엽 %d매: %s", len(sheets), ", ".join(sheets)) return { "x": [min(x0, x1), max(x0, x1)], "y": [min(y0, y1), max(y0, y1)], "z": list(las_bounds.get("z", [0.0, 0.0])), } def planned_route_bounds(project_root: Path, target_epsg: str) -> dict[str, float] | None: """계획노선(B03 CSV)의 평면 범위를 프로젝트 좌표계로 돌려준다. 없으면 None. 지도(2D) 초기 화면을 도로 기준으로 맞출 때 쓴다 — 화면 쪽은 도로 범위만 알면 된다. """ route = read_planned_route(project_root) if not route: return None bounds = route["bounds"] corners = [ (float(bounds["x_min"]), float(bounds["y_min"])), (float(bounds["x_max"]), float(bounds["y_max"])), ] try: moved = _to_target_crs(corners, route.get("epsg"), target_epsg) except Exception as exc: logger.warning("B04 계획노선 범위 변환 실패 (%s)", exc) return None xs = [point[0] for point in moved] ys = [point[1] for point in moved] return {"x_min": min(xs), "x_max": max(xs), "y_min": min(ys), "y_max": max(ys)} def project_epsg_from_prj(project_root: Path) -> str: """프로젝트 PRJ에서 좌표계를 읽는다. 없으면 중부원점(EPSG:5186). PRJ가 둘 이상 올라오므로(노선 세트 + 지형) 지형 PRJ를 고른다 — `find_project_prj`. 판정은 작업 좌표계 창구(`resolve_project_crs`) 하나로 모았다(2026-09-03). """ from common_util.common_util_crs import resolve_project_crs return resolve_project_crs(project_root) def map_meta_covers(meta_path: Path, extent: dict[str, list[float]]) -> bool: """저장된 배경 지도가 필요한 범위를 이미 덮고 있는가. 파일 존재 여부만 보면 계획노선이 바뀌거나 여유 셀 설정을 바꿔도 옛 사진을 계속 쓴다 (B03 업로드 → 전처리 경로는 `rebuild=False`라 더더욱 다시 받지 않는다, 2026-08-01). """ if not meta_path.is_file(): return False try: meta = json.loads(meta_path.read_text(encoding="utf-8")) return ( float(meta["x_min"]) <= extent["x"][0] and float(meta["x_max"]) >= extent["x"][1] and float(meta["y_min"]) <= extent["y"][0] and float(meta["y_max"]) >= extent["y"][1] ) except (OSError, ValueError, KeyError, TypeError): return False def sheet_reference_points_wgs84( project_root: Path, las_bounds: dict[str, list[float]], target_epsg: str, ) -> list[tuple[float, float]]: """도엽 선정 기준 좌표(위도, 경도) 목록. ① 계획노선 시점·종점 → ② 라이다 범위 중심(폴백). 시점과 종점이 서로 다른 도엽에 걸치면 호출측이 두 도엽의 주변을 모두 확보한다. """ from pyproj import Transformer to_wgs84 = Transformer.from_crs(target_epsg, "EPSG:4326", always_xy=True) route = read_planned_route(project_root) if route: ends = [ (float(route["start_point"][0]), float(route["start_point"][1])), (float(route["end_point"][0]), float(route["end_point"][1])), ] try: moved = _to_target_crs(ends, route.get("epsg"), target_epsg) return [(lat, lon) for lon, lat in (to_wgs84.transform(x, y) for x, y in moved)] except Exception as exc: logger.warning("B04 계획노선 기준 좌표 산출 실패 — 라이다 중심 사용 (%s)", exc) center_x = (float(las_bounds["x"][0]) + float(las_bounds["x"][1])) / 2.0 center_y = (float(las_bounds["y"][0]) + float(las_bounds["y"][1])) / 2.0 lon, lat = to_wgs84.transform(center_x, center_y) return [(lat, lon)]