eomsangdonandClaude Opus 5
313a7dde43
사용자 확정(2026-09-02): 설계일자는 확정일 자동이 아니라 **사용자가 넣는다**. 도면번호는 Claude 판단으로 목록 순번을 쓴다. 설계일자 - `013_company_assets.sql` 로 `projects.design_date DATE` 추가(같은 마이그레이션에 로고·서명 공유 테이블 자리도 함께 만들었음 — 배선은 다음 커밋). - B01 프로젝트 수정 모달에 날짜 칸. 공용 `createInputField` 가 `type="date"` 를 받게 넓힘. - 표제란에는 도면 표기 관행대로 `2026. 09. 02.` 꼴로 적는다. 도면번호 - 단건 조회는 목록 순서를 모른다 — 알려면 횡단 장 계획을 다시 계산해야 하고 도면을 열 때마다 그러면 비싸다. **목록이 매긴 번호를 manifest 에 적어 두고** 단건 조회가 읽는다. 목록은 화면 진입 때 늘 먼저 뜨므로 성립하고, 목록이 바뀌면 다음 조회에서 다시 적힌다. - `add_title_fields()` 신설 — 도면을 읽는 도중에 아는 값을 문맥에 얹는 창구. 검증: `pytest tmp/tests/ -q` 139 passed / 0 failed, 루트 `tsc --noEmit` 통과. 실사용자 경로(5174, wdw) — 대시보드 [수정] 모달 라벨 10개(설계일자 칸 추가 확인) → `2026-09-02` 입력 → [확인] → 종단면도 표제란에 `2026. 09. 02.`. 도면번호는 목록 13건 기준 종단 1 · 유역도 6 · 용지도 13 이 도각에 그대로 실림 (표지는 도각을 두르지 않아 번호 칸이 없다). Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
801 lines
34 KiB
Python
801 lines
34 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_blank_drawing,
|
|
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_Engine_Template import add_title_fields
|
|
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"
|
|
|
|
# 아직 내용을 만들지 않은 도면 — 도각만 실어 연다(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)
|
|
|
|
|
|
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)
|
|
]
|
|
# 횡단 장 계획이 실패해도 나머지 도면은 목록에 남긴다 — 예전에는 `cross_sections/`
|
|
# 하나가 없으면 표지·종단면도·토적도·유역도까지 함께 사라졌다(2026-09-01 지적).
|
|
try:
|
|
sheets = _cross_sheet_plan(project_root, longitudinal_path, designs)
|
|
except (FileNotFoundError, ValueError, OSError) as exc:
|
|
logger.warning("B07 횡단 장 계획 실패 — 나머지 도면만 싣는다: %s", exc)
|
|
sheets = []
|
|
for sheet in sheets:
|
|
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")),
|
|
)
|
|
)
|
|
# 아직 내용을 만들지 않은 도면도 **빈 도각으로 열린다**(2026-09-01 사용자 지시).
|
|
# 목록의 절반이 눌리지 않는 회색 버튼이면 고장난 것처럼 보인다.
|
|
for drawing_id, label in BLANK_DRAWINGS:
|
|
drawings.append(DesignDrawingItem(id=drawing_id, kind="blank", label=label))
|
|
_store_drawing_numbers(project_root, drawings)
|
|
return drawings
|
|
|
|
|
|
def _store_drawing_numbers(project_root: Path, drawings: list[DesignDrawingItem]) -> None:
|
|
"""도면번호(목록 순번)를 manifest 에 적어 둔다.
|
|
|
|
단건 조회는 목록 순서를 모른다 — 알려면 횡단 장 계획을 다시 계산해야 하고, 그것을 도면
|
|
열 때마다 하면 비싸다. 목록은 화면에 들어올 때 늘 먼저 뜨므로, 그때 매긴 번호를 적어
|
|
두고 단건 조회는 그것을 읽는다. 목록이 바뀌면 다음 조회에서 다시 적힌다.
|
|
"""
|
|
manifest = _read_manifest(project_root)
|
|
entries = manifest["drawings"]
|
|
changed = False
|
|
for number, item in enumerate(drawings, start=1):
|
|
entry = entries.setdefault(item.id, {})
|
|
if entry.get("number") != number:
|
|
entry["number"] = number
|
|
changed = True
|
|
if changed:
|
|
_write_manifest(project_root, manifest)
|
|
|
|
|
|
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, {})
|
|
number = manifest_entry.get("number")
|
|
if isinstance(number, int):
|
|
add_title_fields({"도면번호": str(number)})
|
|
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 in BLANK_LABELS:
|
|
# 아직 내용을 만들지 않은 도면 — 도각만 실어 연다. 확정 대상이 아니다.
|
|
label = BLANK_LABELS[drawing_id]
|
|
return "blank", label, build_blank_drawing(drawing_id, label), False, None
|
|
|
|
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
|