"""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