diff --git a/B07_DesignDetail/B07_DesignDetail_Router_Support.py b/B07_DesignDetail/B07_DesignDetail_Router_Support.py index 6d69433e..bc045ce5 100644 --- a/B07_DesignDetail/B07_DesignDetail_Router_Support.py +++ b/B07_DesignDetail/B07_DesignDetail_Router_Support.py @@ -5,15 +5,10 @@ import json import logging -import math import re from pathlib import Path from typing import Any -from pyproj import Transformer - -from B04_PreProcess.B04_PreProcess_Router_Watershed import CONTOUR_FILE, STREAM_FILE -from B05_Profile.B05_Profile_Engine_Sections import prune_stale_cross_files from B06_Section.B06_Section_Engine_Design import compute_cross_design from B07_DesignDetail.B07_DesignDetail_Engine_Cad import ( DRAWING_FORMAT, @@ -21,7 +16,7 @@ from B07_DesignDetail.B07_DesignDetail_Engine_Cad import ( infer_station_interval, station_no_label, ) -from B07_DesignDetail.B07_DesignDetail_Engine_Cad_Basin import build_watershed_drawing, map_area_mm +from B07_DesignDetail.B07_DesignDetail_Engine_Cad_Basin import build_watershed_drawing from B07_DesignDetail.B07_DesignDetail_Engine_Cad_Cover import ( build_blank_drawing, build_cover_drawing, @@ -41,15 +36,31 @@ from B07_DesignDetail.B07_DesignDetail_Engine_Cad_Table import ( QUANTITY_VALUE_KEYS, ) from B07_DesignDetail.B07_DesignDetail_Engine_Template import add_title_fields + +# 유역도 배경·파일 입출력 조각은 700줄 제한으로 떼어냈다(2026-09-04). +# 여기서 그대로 다시 내보내 호출부(`B07_DesignDetail_Router.py`)의 import 경로는 불변이다. +from B07_DesignDetail.B07_DesignDetail_Router_Support_Basin import ( + clip_line_to_box as clip_line_to_box, +) +from B07_DesignDetail.B07_DesignDetail_Router_Support_Basin import ( + watershed_source as watershed_source, +) +from B07_DesignDetail.B07_DesignDetail_Router_Support_Io import ( + _cross_files, + _design_root, + _read_json, + _read_manifest, + _station_map, + _write_manifest, +) from B07_DesignDetail.B07_DesignDetail_Schema import ( DesignDrawingItem, ) -from common_util.common_util_drainage_pipes import detail_basins_path from common_util.common_util_route_profile import design_elevation_from_longitudinal -from config.config_system import DRAWING_SCALE_BASIN logger = logging.getLogger(__name__) + _STAGE_DIR = "B07_DesignDetail" _CROSS_ID = re.compile(r"^cross_(\d+)m$") @@ -73,293 +84,6 @@ BLANK_DRAWINGS: tuple[tuple[str, str], ...] = ( BLANK_LABELS: dict[str, str] = dict(BLANK_DRAWINGS) -def _read_json(path: Path) -> dict[str, Any]: - payload = json.loads(path.read_text(encoding="utf-8")) - if not isinstance(payload, dict): - raise ValueError("도면 원본 JSON 형식이 올바르지 않습니다.") - return payload - - -def _cross_files(longitudinal_path: Path, longitudinal: dict[str, Any]) -> list[Path]: - cross_dir = longitudinal_path.parent.parent / "cross_sections" - if not cross_dir.is_dir(): - raise FileNotFoundError("B06 횡단면 파일을 찾을 수 없습니다.") - stations = longitudinal.get("stations") - valid_names = prune_stale_cross_files(cross_dir, stations if isinstance(stations, list) else []) - files = sorted(cross_dir.glob("cross_*.json")) - if valid_names: - files = [path for path in files if path.name in valid_names] - return files - - -def _station_map(longitudinal: dict[str, Any]) -> dict[int, dict[str, Any]]: - stations = longitudinal.get("stations", []) - if not isinstance(stations, list): - return {} - return { - round(float(station.get("chainage_m", 0))): station - for station in stations - if isinstance(station, dict) - } - - -def _design_root(project_root: Path) -> Path: - return project_root / _STAGE_DIR - - -def _read_manifest(project_root: Path) -> dict[str, Any]: - path = _design_root(project_root) / "manifest.json" - if not path.is_file(): - return {"drawings": {}} - payload = _read_json(path) - return payload if isinstance(payload.get("drawings"), dict) else {"drawings": {}} - - -def _write_manifest(project_root: Path, manifest: dict[str, Any]) -> None: - stage_root = _design_root(project_root) - stage_root.mkdir(parents=True, exist_ok=True) - path = stage_root / "manifest.json" - temporary = path.with_suffix(".tmp") - temporary.write_text(json.dumps(manifest, ensure_ascii=False, indent=2), encoding="utf-8") - temporary.replace(path) - - -def _geojson_payload(path: Path) -> dict[str, Any]: - """GeoJSON 전체를 읽는다. 파일이 없거나 깨졌으면 빈 dict.""" - if not path.is_file(): - return {} - try: - payload = json.loads(path.read_text(encoding="utf-8")) - except (OSError, ValueError): - logger.warning("B07 유역도: GeoJSON을 읽지 못했습니다 — %s", path) - return {} - return payload if isinstance(payload, dict) else {} - - -def _geojson_features(path: Path) -> list[dict[str, Any]]: - """GeoJSON 피처 목록만 읽는다. 파일이 없거나 깨졌으면 빈 목록.""" - features = _geojson_payload(path).get("features") - return features if isinstance(features, list) else [] - - -def _basins_crs(context: Any, payload: dict[str, Any]) -> str: - """저장본을 미터로 되돌릴 좌표계. - - 저장할 때 쓴 좌표계를 파일이 갖고 있으면 그 값이다. 없으면 좌표계 기록 이전에 저장된 - 파일이라 노선 CSV의 EPSG 라벨로 쓰였다 — 그 라벨로 되돌려야 왕복이 맞는다 - (2026-09-01). 라벨을 못 읽으면 현재 사업지 좌표계로 둔다. - """ - from common_util.common_util_route_geometry import ( - find_planned_route_file, - read_planned_route, - ) - - stored = payload.get("crs_input") - if isinstance(stored, str) and stored: - return stored - route_file = find_planned_route_file(context.project_root / "B03_FileInput" / "input") - planned = read_planned_route(route_file) if route_file else None - if planned is not None and planned.epsg: - logger.warning( - "B07 유역도: 좌표계 기록이 없는 옛 저장본 — 노선 CSV 라벨 EPSG:%s로 되돌립니다. " - "B04에서 유역을 다시 확정하면 현재 좌표계로 새로 남습니다.", - planned.epsg, - ) - return f"EPSG:{planned.epsg}" - return context.crs - - -def _geometry_lines(geometry: Any) -> list[list[tuple[float, float]]]: - """LineString·MultiLineString·Polygon을 점열 목록으로 편다.""" - if not isinstance(geometry, dict): - return [] - kind = geometry.get("type") - coordinates = geometry.get("coordinates") - if not isinstance(coordinates, list) or not coordinates: - return [] - if kind == "LineString": - return [[(float(p[0]), float(p[1])) for p in coordinates if isinstance(p, list)]] - if kind in ("MultiLineString", "Polygon"): - return [ - [(float(p[0]), float(p[1])) for p in part if isinstance(p, list)] - for part in coordinates - if isinstance(part, list) - ] - return [] - - -def _clip_segment( - start: tuple[float, float], - end: tuple[float, float], - box: tuple[float, float, float, float], -) -> tuple[tuple[float, float], tuple[float, float]] | None: - """선분에서 상자 안에 드는 부분만 돌려준다(Liang-Barsky). 겹치지 않으면 None.""" - min_x, min_y, max_x, max_y = box - dx = end[0] - start[0] - dy = end[1] - start[1] - t0, t1 = 0.0, 1.0 - for numerator, denominator in ( - (start[0] - min_x, -dx), - (max_x - start[0], dx), - (start[1] - min_y, -dy), - (max_y - start[1], dy), - ): - if denominator == 0.0: - if numerator < 0.0: - return None # 경계와 나란한데 바깥이다 - continue - t = numerator / denominator - if denominator < 0.0: - if t > t1: - return None - t0 = max(t0, t) - else: - if t < t0: - return None - t1 = min(t1, t) - return ( - (start[0] + t0 * dx, start[1] + t0 * dy), - (start[0] + t1 * dx, start[1] + t1 * dy), - ) - - -def clip_line_to_box( - line: list[tuple[float, float]], box: tuple[float, float, float, float] -) -> list[list[tuple[float, float]]]: - """범위 안에 든 부분만 조각으로 잘라 낸다 — 끝점은 경계 위에 정확히 놓인다. - - 도엽 등고선 한 줄은 도엽 끝까지 이어지므로, 걸치면 통째로 남기면 도면이 A1을 넘는다. - 예전 구현은 경계 바깥 점을 하나씩 물고 나와 외곽선이 들쭉날쭉했다(2026-08-31 사용자 - 지적). 이제 교차점을 계산해 자르므로 절취면이 곧게 떨어진다. - """ - runs: list[list[tuple[float, float]]] = [] - current: list[tuple[float, float]] = [] - for start, end in zip(line, line[1:]): - piece = _clip_segment(start, end, box) - if piece is None: - current = [] - continue - head, tail = piece - if head == tail: - continue # 경계에 점으로만 닿았다 - if current and current[-1] == head: - current.append(tail) - else: - current = [head, tail] - runs.append(current) - return [run for run in runs if len(run) >= 2] - - -# 세부유역이 노선에서 이만큼 넘게 떨어져 있으면 좌표계를 잘못 되돌린 것으로 본다. -# 임도 한 노선이 담는 유역은 길어야 수 km라 오검출 여지가 없다. -_BASIN_MAX_DISTANCE_M = 50_000.0 - - -def _too_far_from_route( - ring: list[tuple[float, float]], route_xy: list[tuple[float, float]] -) -> bool: - """되돌린 유역이 노선 근처에 없으면 True — 좌표계를 되찾지 못한 저장본이다.""" - if not ring or not route_xy: - return False - cx = sum(x for x, _ in ring) / len(ring) - cy = sum(y for _, y in ring) / len(ring) - return min(math.hypot(cx - x, cy - y) for x, y in route_xy) > _BASIN_MAX_DISTANCE_M - - -def watershed_source(context: Any) -> dict[str, Any]: - """유역도 입력(노선·세부유역·등고선·세류선)을 사업지 CRS(m)로 모은다. - - 저장본은 전부 WGS84라 여기서 미터 좌표로 되돌린다(B04가 저장할 때와 반대 방향). - - 되돌리는 좌표계가 둘이다. **배경(도엽 등고선·세류선)은 사업지 좌표계**로 돌린다 — - 노선(`context.vertices`)이 그 좌표계에 있으므로 같은 자리에 겹쳐야 한다. **세부유역은 - 그 파일을 쓸 때 쓴 좌표계**로 돌린다 — 좌표계 기록 이전 저장본은 노선 CSV의 EPSG - 라벨로 쓰였고, 그 라벨로 되돌려야 원래 미터 좌표가 나온다(2026-09-01). - """ - basins_payload = _geojson_payload(detail_basins_path(context.stored_path)) - to_metric = Transformer.from_crs("EPSG:4326", context.crs, always_xy=True) - to_basin_metric = Transformer.from_crs( - "EPSG:4326", _basins_crs(context, basins_payload), always_xy=True - ) - - def metric(point: tuple[float, float]) -> tuple[float, float]: - x, y = to_metric.transform(point[0], point[1]) - return (float(x), float(y)) - - def basin_metric(point: tuple[float, float]) -> tuple[float, float]: - x, y = to_basin_metric.transform(point[0], point[1]) - return (float(x), float(y)) - - route_xy = [(vertex.x, vertex.y) for vertex in context.vertices] - basins: list[dict[str, Any]] = [] - dropped = 0 - for feature in basins_payload.get("features") or []: - properties = feature.get("properties") or {} - if properties.get("kind") != "detail_basin": - continue - rings = _geometry_lines(feature.get("geometry")) - if not rings: - continue - ring = [basin_metric(point) for point in rings[0]] - # 저장본 좌표계를 못 되찾으면 유역이 노선에서 수백 km 밖으로 떨어진다. 그대로 두면 - # 도곽이 그 거리까지 벌어져 도면이 통째로 빈 화면이 된다 — 유역만 버리고 배경·노선은 - # 그린다(2026-09-01 다른 PC 보고: 유역도 그림 자체가 없음). - if _too_far_from_route(ring, route_xy): - dropped += 1 - continue - basins.append({"ring": ring, "props": properties}) - if dropped: - logger.warning( - "B07 유역도: 노선에서 %.0fkm 넘게 떨어진 세부유역 %d개를 뺐습니다 — " - "저장본 좌표계를 되찾지 못했습니다. B04에서 유역을 다시 확정하세요.", - _BASIN_MAX_DISTANCE_M / 1000.0, - dropped, - ) - - # 배경(등고선·세류선)은 **여러 도엽을 합쳐 받은 뒤 도곽 크기로 절취**한다 - # (2026-08-30 사용자 지시 — 노선이 도엽 경계에 걸릴 수 있어 주변 도엽까지 받아 둔다). - # 도엽 등고선 한 줄은 도엽 끝까지 이어지므로 "걸치면 통째로"는 도면이 A1을 넘긴다 - # (실측 430x871 mm). 절취 범위 = 도곽 안 지형 영역(정보표·제목 제외)을 축척으로 되돌린 크기. - usable_w_mm, usable_h_mm = map_area_mm() - half_w_m = usable_w_mm / 2.0 * DRAWING_SCALE_BASIN / 1000.0 - half_h_m = usable_h_mm / 2.0 * DRAWING_SCALE_BASIN / 1000.0 - extent = [*route_xy, *(point for basin in basins for point in basin["ring"])] - if extent: - center_x = (min(x for x, _y in extent) + max(x for x, _y in extent)) / 2.0 - center_y = (min(y for _x, y in extent) + max(y for _x, y in extent)) / 2.0 - # 노선·유역이 도곽보다 크면 그쪽을 우선한다 — 배경만 잘리고 주제는 다 보인다. - min_x = min(center_x - half_w_m, min(x for x, _y in extent)) - max_x = max(center_x + half_w_m, max(x for x, _y in extent)) - min_y = min(center_y - half_h_m, min(y for _x, y in extent)) - max_y = max(center_y + half_h_m, max(y for _x, y in extent)) - else: - min_x = min_y = -math.inf - max_x = max_y = math.inf - - box = (min_x, min_y, max_x, max_y) - - def clip(line: list[tuple[float, float]]) -> list[list[tuple[float, float]]]: - return clip_line_to_box(line, box) - - sheet_dir = Path(context.project_root) / "B04_PreProcess" / "processed" - background: dict[str, list[list[tuple[float, float]]]] = {} - for key, filename in (("contours", CONTOUR_FILE), ("streams", STREAM_FILE)): - lines: list[list[tuple[float, float]]] = [] - for feature in _geojson_features(sheet_dir / filename): - for part in _geometry_lines(feature.get("geometry")): - converted = [metric(point) for point in part] - if len(converted) >= 2: - lines.extend(clip(converted)) - background[key] = lines - - return { - "route_xy": route_xy, - "basins": basins, - "contours": background["contours"], - "streams": background["streams"], - } - - def _drawing_list( project_root: Path, longitudinal_path: Path, diff --git a/B07_DesignDetail/B07_DesignDetail_Router_Support_Basin.py b/B07_DesignDetail/B07_DesignDetail_Router_Support_Basin.py new file mode 100644 index 00000000..f229cf17 --- /dev/null +++ b/B07_DesignDetail/B07_DesignDetail_Router_Support_Basin.py @@ -0,0 +1,279 @@ +"""B07 유역도 배경 자료 — GeoJSON 읽기·좌표계 환산·도곽 클리핑. + +지원 모듈이 700줄을 넘어 떼어냈다(2026-09-04). 도면 목록·확정 저장은 본체에 남는다. +""" + +import json +import logging +import math +import re +from pathlib import Path +from typing import Any + +from pyproj import Transformer + +from B04_PreProcess.B04_PreProcess_Router_Watershed import CONTOUR_FILE, STREAM_FILE +from B07_DesignDetail.B07_DesignDetail_Engine_Cad_Basin import map_area_mm +from common_util.common_util_drainage_pipes import detail_basins_path +from config.config_system import DRAWING_SCALE_BASIN + +logger = logging.getLogger(__name__) + + +_STAGE_DIR = "B07_DesignDetail" + +_CROSS_ID = re.compile(r"^cross_(\d+)m$") +_LONG_ID = re.compile(r"^longitudinal(?:_(\d+))?$") +# 노선 전장에 한 장씩만 나오는 도면 — 라우터가 원본 자료를 따로 실어 넘긴다. +MASS_HAUL_ID = "mass_haul" +WATERSHED_ID = "watershed" +COVER_ID = "cover" + +# 아직 내용을 만들지 않은 도면 — 도각만 실어 연다(2026-09-01 사용자 지시). +# 화면 순서(DRAWING_GROUPS)와 같은 이름을 쓴다. +BLANK_DRAWINGS: tuple[tuple[str, str], ...] = ( + ("blank_plan_terrain", "계획평면도(지형)"), + ("blank_plan_route", "계획평면도(노선배치도)"), + ("blank_plan_layout", "계획평면도(배치도)"), + ("blank_plan_lidar", "계획평면도(라이다)"), + ("blank_cross_standard", "표준 횡단면도"), + ("blank_standard", "표준도"), + ("blank_landuse", "용지도"), +) +BLANK_LABELS: dict[str, str] = dict(BLANK_DRAWINGS) + + +def _geojson_payload(path: Path) -> dict[str, Any]: + """GeoJSON 전체를 읽는다. 파일이 없거나 깨졌으면 빈 dict.""" + if not path.is_file(): + return {} + try: + payload = json.loads(path.read_text(encoding="utf-8")) + except (OSError, ValueError): + logger.warning("B07 유역도: GeoJSON을 읽지 못했습니다 — %s", path) + return {} + return payload if isinstance(payload, dict) else {} + + +def _geojson_features(path: Path) -> list[dict[str, Any]]: + """GeoJSON 피처 목록만 읽는다. 파일이 없거나 깨졌으면 빈 목록.""" + features = _geojson_payload(path).get("features") + return features if isinstance(features, list) else [] + + +def _basins_crs(context: Any, payload: dict[str, Any]) -> str: + """저장본을 미터로 되돌릴 좌표계. + + 저장할 때 쓴 좌표계를 파일이 갖고 있으면 그 값이다. 없으면 좌표계 기록 이전에 저장된 + 파일이라 노선 CSV의 EPSG 라벨로 쓰였다 — 그 라벨로 되돌려야 왕복이 맞는다 + (2026-09-01). 라벨을 못 읽으면 현재 사업지 좌표계로 둔다. + """ + from common_util.common_util_route_geometry import ( + find_planned_route_file, + read_planned_route, + ) + + stored = payload.get("crs_input") + if isinstance(stored, str) and stored: + return stored + route_file = find_planned_route_file(context.project_root / "B03_FileInput" / "input") + planned = read_planned_route(route_file) if route_file else None + if planned is not None and planned.epsg: + logger.warning( + "B07 유역도: 좌표계 기록이 없는 옛 저장본 — 노선 CSV 라벨 EPSG:%s로 되돌립니다. " + "B04에서 유역을 다시 확정하면 현재 좌표계로 새로 남습니다.", + planned.epsg, + ) + return f"EPSG:{planned.epsg}" + return context.crs + + +def _geometry_lines(geometry: Any) -> list[list[tuple[float, float]]]: + """LineString·MultiLineString·Polygon을 점열 목록으로 편다.""" + if not isinstance(geometry, dict): + return [] + kind = geometry.get("type") + coordinates = geometry.get("coordinates") + if not isinstance(coordinates, list) or not coordinates: + return [] + if kind == "LineString": + return [[(float(p[0]), float(p[1])) for p in coordinates if isinstance(p, list)]] + if kind in ("MultiLineString", "Polygon"): + return [ + [(float(p[0]), float(p[1])) for p in part if isinstance(p, list)] + for part in coordinates + if isinstance(part, list) + ] + return [] + + +def _clip_segment( + start: tuple[float, float], + end: tuple[float, float], + box: tuple[float, float, float, float], +) -> tuple[tuple[float, float], tuple[float, float]] | None: + """선분에서 상자 안에 드는 부분만 돌려준다(Liang-Barsky). 겹치지 않으면 None.""" + min_x, min_y, max_x, max_y = box + dx = end[0] - start[0] + dy = end[1] - start[1] + t0, t1 = 0.0, 1.0 + for numerator, denominator in ( + (start[0] - min_x, -dx), + (max_x - start[0], dx), + (start[1] - min_y, -dy), + (max_y - start[1], dy), + ): + if denominator == 0.0: + if numerator < 0.0: + return None # 경계와 나란한데 바깥이다 + continue + t = numerator / denominator + if denominator < 0.0: + if t > t1: + return None + t0 = max(t0, t) + else: + if t < t0: + return None + t1 = min(t1, t) + return ( + (start[0] + t0 * dx, start[1] + t0 * dy), + (start[0] + t1 * dx, start[1] + t1 * dy), + ) + + +def clip_line_to_box( + line: list[tuple[float, float]], box: tuple[float, float, float, float] +) -> list[list[tuple[float, float]]]: + """범위 안에 든 부분만 조각으로 잘라 낸다 — 끝점은 경계 위에 정확히 놓인다. + + 도엽 등고선 한 줄은 도엽 끝까지 이어지므로, 걸치면 통째로 남기면 도면이 A1을 넘는다. + 예전 구현은 경계 바깥 점을 하나씩 물고 나와 외곽선이 들쭉날쭉했다(2026-08-31 사용자 + 지적). 이제 교차점을 계산해 자르므로 절취면이 곧게 떨어진다. + """ + runs: list[list[tuple[float, float]]] = [] + current: list[tuple[float, float]] = [] + for start, end in zip(line, line[1:]): + piece = _clip_segment(start, end, box) + if piece is None: + current = [] + continue + head, tail = piece + if head == tail: + continue # 경계에 점으로만 닿았다 + if current and current[-1] == head: + current.append(tail) + else: + current = [head, tail] + runs.append(current) + return [run for run in runs if len(run) >= 2] + + +# 세부유역이 노선에서 이만큼 넘게 떨어져 있으면 좌표계를 잘못 되돌린 것으로 본다. +# 임도 한 노선이 담는 유역은 길어야 수 km라 오검출 여지가 없다. +_BASIN_MAX_DISTANCE_M = 50_000.0 + + +def _too_far_from_route( + ring: list[tuple[float, float]], route_xy: list[tuple[float, float]] +) -> bool: + """되돌린 유역이 노선 근처에 없으면 True — 좌표계를 되찾지 못한 저장본이다.""" + if not ring or not route_xy: + return False + cx = sum(x for x, _ in ring) / len(ring) + cy = sum(y for _, y in ring) / len(ring) + return min(math.hypot(cx - x, cy - y) for x, y in route_xy) > _BASIN_MAX_DISTANCE_M + + +def watershed_source(context: Any) -> dict[str, Any]: + """유역도 입력(노선·세부유역·등고선·세류선)을 사업지 CRS(m)로 모은다. + + 저장본은 전부 WGS84라 여기서 미터 좌표로 되돌린다(B04가 저장할 때와 반대 방향). + + 되돌리는 좌표계가 둘이다. **배경(도엽 등고선·세류선)은 사업지 좌표계**로 돌린다 — + 노선(`context.vertices`)이 그 좌표계에 있으므로 같은 자리에 겹쳐야 한다. **세부유역은 + 그 파일을 쓸 때 쓴 좌표계**로 돌린다 — 좌표계 기록 이전 저장본은 노선 CSV의 EPSG + 라벨로 쓰였고, 그 라벨로 되돌려야 원래 미터 좌표가 나온다(2026-09-01). + """ + basins_payload = _geojson_payload(detail_basins_path(context.stored_path)) + to_metric = Transformer.from_crs("EPSG:4326", context.crs, always_xy=True) + to_basin_metric = Transformer.from_crs( + "EPSG:4326", _basins_crs(context, basins_payload), always_xy=True + ) + + def metric(point: tuple[float, float]) -> tuple[float, float]: + x, y = to_metric.transform(point[0], point[1]) + return (float(x), float(y)) + + def basin_metric(point: tuple[float, float]) -> tuple[float, float]: + x, y = to_basin_metric.transform(point[0], point[1]) + return (float(x), float(y)) + + route_xy = [(vertex.x, vertex.y) for vertex in context.vertices] + basins: list[dict[str, Any]] = [] + dropped = 0 + for feature in basins_payload.get("features") or []: + properties = feature.get("properties") or {} + if properties.get("kind") != "detail_basin": + continue + rings = _geometry_lines(feature.get("geometry")) + if not rings: + continue + ring = [basin_metric(point) for point in rings[0]] + # 저장본 좌표계를 못 되찾으면 유역이 노선에서 수백 km 밖으로 떨어진다. 그대로 두면 + # 도곽이 그 거리까지 벌어져 도면이 통째로 빈 화면이 된다 — 유역만 버리고 배경·노선은 + # 그린다(2026-09-01 다른 PC 보고: 유역도 그림 자체가 없음). + if _too_far_from_route(ring, route_xy): + dropped += 1 + continue + basins.append({"ring": ring, "props": properties}) + if dropped: + logger.warning( + "B07 유역도: 노선에서 %.0fkm 넘게 떨어진 세부유역 %d개를 뺐습니다 — " + "저장본 좌표계를 되찾지 못했습니다. B04에서 유역을 다시 확정하세요.", + _BASIN_MAX_DISTANCE_M / 1000.0, + dropped, + ) + + # 배경(등고선·세류선)은 **여러 도엽을 합쳐 받은 뒤 도곽 크기로 절취**한다 + # (2026-08-30 사용자 지시 — 노선이 도엽 경계에 걸릴 수 있어 주변 도엽까지 받아 둔다). + # 도엽 등고선 한 줄은 도엽 끝까지 이어지므로 "걸치면 통째로"는 도면이 A1을 넘긴다 + # (실측 430x871 mm). 절취 범위 = 도곽 안 지형 영역(정보표·제목 제외)을 축척으로 되돌린 크기. + usable_w_mm, usable_h_mm = map_area_mm() + half_w_m = usable_w_mm / 2.0 * DRAWING_SCALE_BASIN / 1000.0 + half_h_m = usable_h_mm / 2.0 * DRAWING_SCALE_BASIN / 1000.0 + extent = [*route_xy, *(point for basin in basins for point in basin["ring"])] + if extent: + center_x = (min(x for x, _y in extent) + max(x for x, _y in extent)) / 2.0 + center_y = (min(y for _x, y in extent) + max(y for _x, y in extent)) / 2.0 + # 노선·유역이 도곽보다 크면 그쪽을 우선한다 — 배경만 잘리고 주제는 다 보인다. + min_x = min(center_x - half_w_m, min(x for x, _y in extent)) + max_x = max(center_x + half_w_m, max(x for x, _y in extent)) + min_y = min(center_y - half_h_m, min(y for _x, y in extent)) + max_y = max(center_y + half_h_m, max(y for _x, y in extent)) + else: + min_x = min_y = -math.inf + max_x = max_y = math.inf + + box = (min_x, min_y, max_x, max_y) + + def clip(line: list[tuple[float, float]]) -> list[list[tuple[float, float]]]: + return clip_line_to_box(line, box) + + sheet_dir = Path(context.project_root) / "B04_PreProcess" / "processed" + background: dict[str, list[list[tuple[float, float]]]] = {} + for key, filename in (("contours", CONTOUR_FILE), ("streams", STREAM_FILE)): + lines: list[list[tuple[float, float]]] = [] + for feature in _geojson_features(sheet_dir / filename): + for part in _geometry_lines(feature.get("geometry")): + converted = [metric(point) for point in part] + if len(converted) >= 2: + lines.extend(clip(converted)) + background[key] = lines + + return { + "route_xy": route_xy, + "basins": basins, + "contours": background["contours"], + "streams": background["streams"], + } diff --git a/B07_DesignDetail/B07_DesignDetail_Router_Support_Io.py b/B07_DesignDetail/B07_DesignDetail_Router_Support_Io.py new file mode 100644 index 00000000..b4ebb452 --- /dev/null +++ b/B07_DesignDetail/B07_DesignDetail_Router_Support_Io.py @@ -0,0 +1,88 @@ +"""B07 도면 조립 지원 — 원본 JSON·매니페스트 읽기/쓰기와 측점 대응표. + +지원 모듈이 700줄을 넘어 떼어냈다(2026-09-04). 유역도 조각과 본체가 함께 쓴다. +""" + +import json +import logging +import re +from pathlib import Path +from typing import Any + +from B05_Profile.B05_Profile_Engine_Sections import prune_stale_cross_files + +logger = logging.getLogger(__name__) + + +_STAGE_DIR = "B07_DesignDetail" + +_CROSS_ID = re.compile(r"^cross_(\d+)m$") +_LONG_ID = re.compile(r"^longitudinal(?:_(\d+))?$") +# 노선 전장에 한 장씩만 나오는 도면 — 라우터가 원본 자료를 따로 실어 넘긴다. +MASS_HAUL_ID = "mass_haul" +WATERSHED_ID = "watershed" +COVER_ID = "cover" + +# 아직 내용을 만들지 않은 도면 — 도각만 실어 연다(2026-09-01 사용자 지시). +# 화면 순서(DRAWING_GROUPS)와 같은 이름을 쓴다. +BLANK_DRAWINGS: tuple[tuple[str, str], ...] = ( + ("blank_plan_terrain", "계획평면도(지형)"), + ("blank_plan_route", "계획평면도(노선배치도)"), + ("blank_plan_layout", "계획평면도(배치도)"), + ("blank_plan_lidar", "계획평면도(라이다)"), + ("blank_cross_standard", "표준 횡단면도"), + ("blank_standard", "표준도"), + ("blank_landuse", "용지도"), +) +BLANK_LABELS: dict[str, str] = dict(BLANK_DRAWINGS) + + +def _read_json(path: Path) -> dict[str, Any]: + payload = json.loads(path.read_text(encoding="utf-8")) + if not isinstance(payload, dict): + raise ValueError("도면 원본 JSON 형식이 올바르지 않습니다.") + return payload + + +def _cross_files(longitudinal_path: Path, longitudinal: dict[str, Any]) -> list[Path]: + cross_dir = longitudinal_path.parent.parent / "cross_sections" + if not cross_dir.is_dir(): + raise FileNotFoundError("B06 횡단면 파일을 찾을 수 없습니다.") + stations = longitudinal.get("stations") + valid_names = prune_stale_cross_files(cross_dir, stations if isinstance(stations, list) else []) + files = sorted(cross_dir.glob("cross_*.json")) + if valid_names: + files = [path for path in files if path.name in valid_names] + return files + + +def _station_map(longitudinal: dict[str, Any]) -> dict[int, dict[str, Any]]: + stations = longitudinal.get("stations", []) + if not isinstance(stations, list): + return {} + return { + round(float(station.get("chainage_m", 0))): station + for station in stations + if isinstance(station, dict) + } + + +def _design_root(project_root: Path) -> Path: + return project_root / _STAGE_DIR + + +def _read_manifest(project_root: Path) -> dict[str, Any]: + path = _design_root(project_root) / "manifest.json" + if not path.is_file(): + return {"drawings": {}} + payload = _read_json(path) + return payload if isinstance(payload.get("drawings"), dict) else {"drawings": {}} + + +def _write_manifest(project_root: Path, manifest: dict[str, Any]) -> None: + stage_root = _design_root(project_root) + stage_root.mkdir(parents=True, exist_ok=True) + path = stage_root / "manifest.json" + temporary = path.with_suffix(".tmp") + temporary.write_text(json.dumps(manifest, ensure_ascii=False, indent=2), encoding="utf-8") + temporary.replace(path)