eomsangdonandClaude Opus 5
9fb453cfca
좌표계 기록 이전 저장본을 되돌리는 폴백을 넣으면서 변환을 하나로 묶었다. 그래서 도엽 배경(등고선·세류선)까지 저장본 좌표계로 되돌아가, 라벨과 실좌표계가 갈린 사업지에서 배경이 1,000km 밖에 놓였다. 도곽이 그 거리까지 벌어지면 유역도는 사실상 빈 화면이 된다 (다른 PC 보고: 배수유역도 그림 자체가 없음). 변환을 둘로 나눈다. - 배경·노선: `context.crs`(사업지 .prj 좌표계) — 노선이 그 좌표계에 있다. - 세부유역: 그 파일을 쓸 때 쓴 좌표계 — 옛 저장본은 노선 CSV의 EPSG 라벨로 쓰였고, 같은 라벨로 되돌려야 원래 미터 좌표가 나온다(왕복이 상쇄된다). 되찾을 근거조차 없을 때(노선 파일 없음 등)를 위해 안전장치를 뒀다. 노선에서 50km 넘게 떨어진 유역은 빼고 경고를 남긴다 — 도면 전체를 잃는 대신 배경·노선은 그린다. 폴백에 쓰는 route_geometry import 는 함수 안으로 내렸다(순환 import 여지 제거). 검증: tmp/tests 44 passed(신규 test_watershed_source_crs.py 2건). c1bb453f 유역도 API 재확인 — 엔티티 1,597개(등고선 1028·계류 460·배수유역 20·도각 82· 유역 제원 4·노선 1·표제 2)로 수정 전과 같다. Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
745 lines
31 KiB
Python
745 lines
31 KiB
Python
"""B07 도면 조립 지원 — 파일·매니페스트 읽기, 도면 목록·장 배치, 확정 저장.
|
|
|
|
라우터에서 HTTP 처리와 무관한 동기 헬퍼만 떼어 놓은 모듈이다(단일 파일 700줄 제한).
|
|
"""
|
|
|
|
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,
|
|
build_cross_drawing,
|
|
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_Cover import build_cover_drawing
|
|
from B07_DesignDetail.B07_DesignDetail_Engine_Cad_Long import (
|
|
build_longitudinal_drawing,
|
|
longitudinal_chunks,
|
|
)
|
|
from B07_DesignDetail.B07_DesignDetail_Engine_Cad_MassHaul import build_mass_haul_drawing
|
|
from B07_DesignDetail.B07_DesignDetail_Engine_Cad_Sheet import (
|
|
CROSS_SHEET_ID,
|
|
build_cross_sheet,
|
|
plan_cross_sheets,
|
|
section_block_size,
|
|
)
|
|
from B07_DesignDetail.B07_DesignDetail_Engine_Cad_Table import (
|
|
QUANTITY_VALUE_KEYS,
|
|
)
|
|
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$")
|
|
_LONG_ID = re.compile(r"^longitudinal(?:_(\d+))?$")
|
|
# 노선 전장에 한 장씩만 나오는 도면 — 라우터가 원본 자료를 따로 실어 넘긴다.
|
|
MASS_HAUL_ID = "mass_haul"
|
|
WATERSHED_ID = "watershed"
|
|
COVER_ID = "cover"
|
|
|
|
|
|
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,
|
|
designs: dict[int, dict[str, Any]] | None = None,
|
|
) -> list[DesignDrawingItem]:
|
|
longitudinal = _read_json(longitudinal_path)
|
|
station_by_chainage = _station_map(longitudinal)
|
|
manifest_drawings = _read_manifest(project_root)["drawings"]
|
|
# 종단도: 측점 30개 초과 시 30개 단위 분할 도면을 각각 목록에 노출한다(N-1-1).
|
|
drawings = [
|
|
DesignDrawingItem(
|
|
id=str(chunk["id"]),
|
|
kind="longitudinal",
|
|
label=str(chunk["label"]),
|
|
confirmed=bool(manifest_drawings.get(str(chunk["id"]), {}).get("confirmed")),
|
|
)
|
|
for chunk in longitudinal_chunks(longitudinal)
|
|
]
|
|
for sheet in _cross_sheet_plan(project_root, longitudinal_path, designs):
|
|
chainages = sheet["chainages"]
|
|
first = station_by_chainage.get(chainages[0], {})
|
|
span = f"{chainages[0]}m"
|
|
if len(chainages) > 1:
|
|
span = f"{chainages[0]}~{chainages[-1]}m"
|
|
drawings.append(
|
|
DesignDrawingItem(
|
|
id=sheet["id"],
|
|
kind="cross",
|
|
label=f"{sheet['number']}장 ({span})",
|
|
chainage_m=float(first.get("chainage_m", chainages[0])),
|
|
confirmed=bool(manifest_drawings.get(sheet["id"], {}).get("confirmed")),
|
|
)
|
|
)
|
|
# 노선 전장 1장짜리 도면 — 자료가 없으면 여는 시점에 404로 알린다(목록에는 항상 둔다).
|
|
for drawing_id, kind, label in (
|
|
(COVER_ID, "cover", "표지"),
|
|
(MASS_HAUL_ID, "mass_haul", "토적도(유토곡선)"),
|
|
(WATERSHED_ID, "watershed", "유역도(배수 유역도)"),
|
|
):
|
|
drawings.append(
|
|
DesignDrawingItem(
|
|
id=drawing_id,
|
|
kind=kind,
|
|
label=label,
|
|
confirmed=bool(manifest_drawings.get(drawing_id, {}).get("confirmed")),
|
|
)
|
|
)
|
|
return drawings
|
|
|
|
|
|
def _cross_sheet_plan(
|
|
project_root: Path,
|
|
longitudinal_path: Path,
|
|
designs: dict[int, dict[str, Any]] | None = None,
|
|
) -> list[dict[str, Any]]:
|
|
"""측점 도면을 A1 장으로 나눈 계획 — 목록과 도면 생성이 같은 결과를 쓴다.
|
|
|
|
한 장에 몇 개가 들어가는지는 단면 블록 크기가 정한다(작성 척도 1/100 +
|
|
설계선이 원지반과 갈라지는 구간). 설계 지정이 없으면 원지반 기준 폭이 된다.
|
|
"""
|
|
longitudinal = _read_json(longitudinal_path)
|
|
blocks: list[tuple[int, float, float]] = []
|
|
for path in sorted(_cross_files(longitudinal_path, longitudinal)):
|
|
match = _CROSS_ID.fullmatch(path.stem)
|
|
if not match:
|
|
continue
|
|
chainage = int(match.group(1))
|
|
source = _read_json(path)
|
|
design = (designs or {}).get(chainage)
|
|
design_line = _cross_design_line(longitudinal_path, source, design)
|
|
width, height = section_block_size(source, design_line)
|
|
blocks.append((chainage, width, height))
|
|
return plan_cross_sheets(blocks)
|
|
|
|
|
|
def _quantity_table(source: dict[str, Any]) -> dict[str, float | None]:
|
|
"""횡단면 원본에서 편집 가능한 수량 산출표 값을 구조화한다 (납품 양식 키).
|
|
|
|
center_z→지반고, design_elevation_m→계획고, 절토고/성토고는 파생 초기값.
|
|
나머지 항목은 source["quantities"]에 같은 키가 있으면 읽고 없으면 None으로
|
|
두어 CAD 테이블에서 사용자가 채운다.
|
|
"""
|
|
|
|
def num(value: Any) -> float | None:
|
|
return float(value) if isinstance(value, (int, float)) else None
|
|
|
|
ground = num(source.get("center_z"))
|
|
planned = num(source.get("planned_elevation_m", source.get("design_elevation_m")))
|
|
cut = max(ground - planned, 0.0) if ground is not None and planned is not None else None
|
|
fill = max(planned - ground, 0.0) if ground is not None and planned is not None else None
|
|
quantities = source.get("quantities") if isinstance(source.get("quantities"), dict) else {}
|
|
|
|
table: dict[str, float | None] = {
|
|
"ground": ground,
|
|
"planned": planned,
|
|
"cut": cut,
|
|
"fill": fill,
|
|
}
|
|
for key in QUANTITY_VALUE_KEYS:
|
|
table.setdefault(key, num(quantities.get(key)))
|
|
return table
|
|
|
|
|
|
def _route_cross_frame(
|
|
longitudinal_path: Path, longitudinal: dict[str, Any]
|
|
) -> dict[str, float] | None:
|
|
"""노선 전체 횡단면의 기준 레이아웃 {half, half_height}를 구한다.
|
|
|
|
측점별 단면 높이(최대-최소 표고)와 폭의 노선 최대값 — 모든 횡단도가 자기
|
|
콘텐츠 중심 기준으로 같은 크기 레이아웃에 배치되도록 하는 기준값(테이블
|
|
y 고정용). 실패 시 None(측점 자체 범위 폴백).
|
|
"""
|
|
half = half_height = 0.0
|
|
found = False
|
|
try:
|
|
for path in _cross_files(longitudinal_path, longitudinal):
|
|
source = _read_json(path)
|
|
min_e: float | None = None
|
|
max_e: float | None = None
|
|
for sample in source.get("samples", []):
|
|
if not isinstance(sample, dict) or not sample.get("valid", False):
|
|
continue
|
|
offset = sample.get("offset_m")
|
|
elevation = sample.get("elevation_m")
|
|
if isinstance(offset, (int, float)):
|
|
half = max(half, abs(float(offset)))
|
|
if isinstance(elevation, (int, float)):
|
|
value = float(elevation)
|
|
min_e = value if min_e is None else min(min_e, value)
|
|
max_e = value if max_e is None else max(max_e, value)
|
|
if min_e is not None and max_e is not None:
|
|
half_height = max(half_height, (max_e - min_e) / 2.0)
|
|
found = True
|
|
except (OSError, ValueError, json.JSONDecodeError, FileNotFoundError):
|
|
return None
|
|
if not found:
|
|
return None
|
|
# 여유: 측구 깊이·사면 연장 등 설계선이 지반 포락선을 소폭 벗어나는 분 반영.
|
|
return {"half": half or 12.0, "half_height": half_height + 1.5}
|
|
|
|
|
|
def _cross_design_line(
|
|
longitudinal_path: Path, source: dict[str, Any], stored_design: dict[str, Any] | None
|
|
) -> list[Any] | None:
|
|
"""횡단 CAD 계획선용 design_line을 정한다: 저장 설계 우선, 없으면 기본값 계산."""
|
|
if isinstance(stored_design, dict) and isinstance(stored_design.get("design_line"), list):
|
|
return stored_design["design_line"]
|
|
try:
|
|
longitudinal = _read_json(longitudinal_path)
|
|
design = compute_cross_design(
|
|
source.get("samples", []),
|
|
design_elevation_from_longitudinal(longitudinal, float(source.get("chainage_m", 0.0))),
|
|
ground_type="soil",
|
|
section_mode="left_cut",
|
|
)
|
|
return design["design_line"]
|
|
except (ValueError, KeyError, OSError, json.JSONDecodeError):
|
|
return None
|
|
|
|
|
|
def _cross_section_input(
|
|
longitudinal_path: Path,
|
|
longitudinal: dict[str, Any],
|
|
chainage: int,
|
|
design: dict[str, Any] | None,
|
|
) -> dict[str, Any] | None:
|
|
"""한 측점의 장 배치 입력(원본·계획선·수량표·제목)을 만든다."""
|
|
path = longitudinal_path.parent.parent / "cross_sections" / f"cross_{chainage:05d}m.json"
|
|
if not path.is_file():
|
|
return None
|
|
source = _read_json(path)
|
|
interval = infer_station_interval(longitudinal.get("stations") or [])
|
|
return {
|
|
"chainage": chainage,
|
|
"source": source,
|
|
"design": design,
|
|
"design_line": _cross_design_line(longitudinal_path, source, design),
|
|
"quantity_table": _quantity_table(source),
|
|
"title": station_no_label(float(source.get("chainage_m", chainage)), interval),
|
|
}
|
|
|
|
|
|
def _read_cross_sheet(
|
|
project_root: Path,
|
|
longitudinal_path: Path,
|
|
drawing_id: str,
|
|
stored_designs: dict[str, Any] | None,
|
|
) -> tuple[str, str, dict[str, Any], bool, dict[str, float | None] | None]:
|
|
"""횡단 장 하나를 만든다. stored_designs는 {측점: 설계 지정} 묶음이다."""
|
|
designs = stored_designs if isinstance(stored_designs, dict) else {}
|
|
plan = _cross_sheet_plan(project_root, longitudinal_path, designs)
|
|
sheet = next((item for item in plan if item["id"] == drawing_id), None)
|
|
if sheet is None:
|
|
raise FileNotFoundError("요청한 횡단 장을 찾을 수 없습니다.")
|
|
longitudinal = _read_json(longitudinal_path)
|
|
sections = [
|
|
section
|
|
for section in (
|
|
_cross_section_input(longitudinal_path, longitudinal, chainage, designs.get(chainage))
|
|
for chainage in sheet["chainages"]
|
|
)
|
|
if section is not None
|
|
]
|
|
label = f"횡단면도 {sheet['number']}장"
|
|
return "cross", label, build_cross_sheet(sheet, sections), False, None
|
|
|
|
|
|
def _read_drawing(
|
|
project_root: Path,
|
|
longitudinal_path: Path,
|
|
drawing_id: str,
|
|
stored_design: dict[str, Any] | None = None,
|
|
) -> tuple[str, str, dict[str, Any], bool, dict[str, float | None] | None]:
|
|
"""(kind, label, drawing, confirmed, quantity_table)를 반환한다.
|
|
|
|
quantity_table은 횡단도에서만 채워지며, 확정본은 manifest에 저장된 사용자
|
|
편집값을 우선하고 없으면 원본에서 파생한 초기값을 계산한다. 횡단도의 계획선은
|
|
stored_design(없으면 기본값)에서 만든다.
|
|
"""
|
|
manifest_entry = _read_manifest(project_root)["drawings"].get(drawing_id, {})
|
|
saved_path = _design_root(project_root) / "drawings" / f"{drawing_id}.json"
|
|
if manifest_entry.get("confirmed") and saved_path.is_file():
|
|
saved = _read_json(saved_path)
|
|
# 포맷 버전이 다르면(테이블·레이어 구성 변경 전 저장본) 캐시를 버리고
|
|
# 아래에서 원본 기준으로 재생성한다. 확정 상태도 무효로 응답해 재확정 유도.
|
|
if saved.get("format") == DRAWING_FORMAT:
|
|
if drawing_id in (COVER_ID, MASS_HAUL_ID, WATERSHED_ID):
|
|
kind = drawing_id # id와 kind가 같은 단장 도면
|
|
else:
|
|
kind = "longitudinal" if _LONG_ID.fullmatch(drawing_id) else "cross"
|
|
label = str(manifest_entry.get("label") or drawing_id)
|
|
stored_table = manifest_entry.get("quantity_table")
|
|
table = stored_table if kind == "cross" and isinstance(stored_table, dict) else None
|
|
return kind, label, saved, True, table
|
|
if drawing_id == COVER_ID:
|
|
# 표지는 설계 자료를 쓰지 않는다 — 템플릿 한 장이 곧 도면이다.
|
|
return "cover", "표지", build_cover_drawing(drawing_id), False, None
|
|
|
|
if drawing_id == MASS_HAUL_ID:
|
|
# stored_design = 확정 종단 DB row의 mass_haul 산출물(라우터가 실어 준다).
|
|
if not isinstance(stored_design, dict):
|
|
raise FileNotFoundError("확정 종단에 유토곡선 산출물이 없습니다.")
|
|
longitudinal = _read_json(longitudinal_path)
|
|
return (
|
|
"mass_haul",
|
|
"토적도(유토곡선)",
|
|
build_mass_haul_drawing(longitudinal, stored_design, drawing_id),
|
|
False,
|
|
None,
|
|
)
|
|
|
|
if drawing_id == WATERSHED_ID:
|
|
# stored_design = watershed_source()가 모아 준 노선·유역·배경 좌표(사업지 CRS).
|
|
if not isinstance(stored_design, dict):
|
|
raise FileNotFoundError("배수 유역 산출물이 없습니다.")
|
|
longitudinal = _read_json(longitudinal_path)
|
|
interval = infer_station_interval(longitudinal.get("stations") or [])
|
|
return (
|
|
"watershed",
|
|
"유역도(배수 유역도)",
|
|
build_watershed_drawing(
|
|
drawing_id,
|
|
stored_design.get("route_xy") or [],
|
|
stored_design.get("basins") or [],
|
|
stored_design.get("contours") or [],
|
|
stored_design.get("streams") or [],
|
|
interval,
|
|
),
|
|
False,
|
|
None,
|
|
)
|
|
|
|
if _LONG_ID.fullmatch(drawing_id):
|
|
source = _read_json(longitudinal_path)
|
|
chunk = next(
|
|
(item for item in longitudinal_chunks(source) if item["id"] == drawing_id), None
|
|
)
|
|
if chunk is None:
|
|
raise FileNotFoundError("요청한 종단도 분할 도면을 찾을 수 없습니다.")
|
|
return (
|
|
"longitudinal",
|
|
str(chunk["label"]),
|
|
build_longitudinal_drawing(source, drawing_id, chunk),
|
|
False,
|
|
None,
|
|
)
|
|
|
|
if CROSS_SHEET_ID.fullmatch(drawing_id):
|
|
return _read_cross_sheet(project_root, longitudinal_path, drawing_id, stored_design)
|
|
|
|
if not _CROSS_ID.fullmatch(drawing_id):
|
|
raise ValueError("올바르지 않은 도면 ID입니다.")
|
|
path = longitudinal_path.parent.parent / "cross_sections" / f"{drawing_id}.json"
|
|
if not path.is_file():
|
|
raise FileNotFoundError("요청한 횡단도를 찾을 수 없습니다.")
|
|
source = _read_json(path)
|
|
label = str(source.get("label") or drawing_id)
|
|
design_line = _cross_design_line(longitudinal_path, source, stored_design)
|
|
quantity_table = _quantity_table(source)
|
|
# 수량표 제목행 No. 표기: 종단 측점 간격 기준 (납품 도면 양식)
|
|
longitudinal = _read_json(longitudinal_path)
|
|
interval = infer_station_interval(longitudinal.get("stations") or [])
|
|
title = station_no_label(float(source.get("chainage_m", 0.0)), interval)
|
|
# 계획고(로컬좌표 기준) + 노선 공통 범위 — 측점 이동 시 화면 배치 고정.
|
|
design_elevation = design_elevation_from_longitudinal(
|
|
longitudinal, float(source.get("chainage_m", 0.0))
|
|
)
|
|
frame = _route_cross_frame(longitudinal_path, longitudinal)
|
|
return (
|
|
"cross",
|
|
label,
|
|
build_cross_drawing(
|
|
source,
|
|
drawing_id,
|
|
design_line,
|
|
stored_design,
|
|
quantity_table,
|
|
title,
|
|
design_elevation,
|
|
frame,
|
|
),
|
|
False,
|
|
quantity_table,
|
|
)
|
|
|
|
|
|
def _store_confirmed_drawing(
|
|
project_root: Path,
|
|
item: DesignDrawingItem,
|
|
drawing: dict[str, Any],
|
|
expected_ids: set[str],
|
|
quantity_table: dict[str, Any] | None = None,
|
|
quantity_tables: dict[str, Any] | None = None,
|
|
) -> bool:
|
|
if not isinstance(drawing.get("entities"), list) or not isinstance(drawing.get("layers"), list):
|
|
raise ValueError("CAD 도면 스키마가 올바르지 않습니다.")
|
|
# CAD 앱 직렬화본에는 format이 없으므로 저장 시 현재 포맷 버전을 스탬프한다.
|
|
drawing = {"format": DRAWING_FORMAT, **drawing}
|
|
drawings_dir = _design_root(project_root) / "drawings"
|
|
drawings_dir.mkdir(parents=True, exist_ok=True)
|
|
path = drawings_dir / f"{item.id}.json"
|
|
temporary = path.with_suffix(".tmp")
|
|
temporary.write_text(json.dumps(drawing, ensure_ascii=False, indent=2), encoding="utf-8")
|
|
temporary.replace(path)
|
|
|
|
manifest = _read_manifest(project_root)
|
|
entry: dict[str, Any] = {
|
|
"kind": item.kind,
|
|
"label": item.label,
|
|
"confirmed": True,
|
|
"file": f"drawings/{item.id}.json",
|
|
}
|
|
if item.kind == "cross" and isinstance(quantity_table, dict):
|
|
entry["quantity_table"] = quantity_table
|
|
if isinstance(quantity_tables, dict) and quantity_tables:
|
|
# 장 도면: 측점별 수량표를 그대로 보관한다(뒷단계 수량산출이 측점 단위).
|
|
entry["quantity_tables"] = quantity_tables
|
|
manifest["drawings"][item.id] = entry
|
|
_write_manifest(project_root, manifest)
|
|
confirmed_ids = {
|
|
item_id for item_id, entry in manifest["drawings"].items() if entry.get("confirmed")
|
|
}
|
|
return expected_ids.issubset(confirmed_ids)
|
|
|
|
|
|
def _invalidate_drawing(project_root: Path, drawing_id: str) -> None:
|
|
manifest = _read_manifest(project_root)
|
|
entry = manifest["drawings"].get(drawing_id)
|
|
if entry:
|
|
entry["confirmed"] = False
|
|
_write_manifest(project_root, manifest)
|
|
|
|
|
|
def _recompute_confirmed_design(
|
|
longitudinal_path: Path, cross_stem: str, designation: dict[str, Any]
|
|
) -> dict[str, Any]:
|
|
"""B06 지정값과 현재 계획고로 절·성토 단면적을 재계산해 확정치(status=confirmed)로 만든다.
|
|
|
|
B07 CAD에는 아직 편집 가능한 설계선이 없으므로, 저장된 지정값(지반유형·단면유형·
|
|
측구위치)과 계획고로 동일 엔진을 재실행해 확정 시점 값을 고정한다.
|
|
"""
|
|
longitudinal = _read_json(longitudinal_path)
|
|
cross_path = longitudinal_path.parent.parent / "cross_sections" / f"{cross_stem}.json"
|
|
source = _read_json(cross_path)
|
|
samples = source.get("samples")
|
|
if not isinstance(samples, list):
|
|
raise ValueError("횡단 상세 파일 형식이 올바르지 않습니다.")
|
|
design_elevation = design_elevation_from_longitudinal(
|
|
longitudinal, float(source.get("chainage_m", 0.0))
|
|
)
|
|
design = compute_cross_design(
|
|
samples,
|
|
design_elevation,
|
|
ground_type=designation["ground_type"],
|
|
section_mode=designation["section_mode"],
|
|
ditch_side=designation.get("ditch_side"),
|
|
)
|
|
design["status"] = "confirmed"
|
|
return design
|