eomsangdon/Aislo
1
0
Fork 0
Code Issues Pull Requests 2 Actions Packages Projects Releases Wiki Activity

refactor(B07): 도면 조립 지원 700줄 초과 분리 — 유역도·입출력

Browse Source 
808줄 한 파일을 셋으로 나눔 (동작 불변, 순수 이동).
- `B07_DesignDetail_Router_Support.py` 532줄 — 도면 목록·장 배치·확정 저장·재계산
- `B07_DesignDetail_Router_Support_Basin.py` 279줄 — 유역도 배경(GeoJSON·좌표 환산·도곽 클리핑)
- `B07_DesignDetail_Router_Support_Io.py` 88줄 — 원본 JSON·매니페스트 입출력, 측점 대응표

떼어낸 `watershed_source`·`clip_line_to_box` 는 본체에서 `as` 로 다시 내보내
호출부(`B07_DesignDetail_Router.py`)의 import 경로가 그대로임.

검증: 라우트 7개 그대로, 공용 브라우저 `design-drawings` 200 · `frame-template` 200,
ruff check 통과, tmp/tests 378 passed

Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
This commit is contained in:
eomsangdon
2026-09-04 10:34:18 +09:00
co-authored by Claude Opus 5
parent 3ec6527cc2
commit 8be60007e7
3 changed files with 386 additions and 295 deletions
Download Patch File Download Diff File Expand all files Collapse all files
B07_DesignDetail/B07_DesignDetail_Router_Support.py
+19 -295
View File
@@ -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,
B07_DesignDetail/B07_DesignDetail_Router_Support_Basin.py
+279
View File
@@ -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"],
}
B07_DesignDetail/B07_DesignDetail_Router_Support_Io.py
+88
View File
@@ -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)