"""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_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" 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_features(path: Path) -> list[dict[str, Any]]: """GeoJSON 피처 목록만 읽는다. 파일이 없거나 깨졌으면 빈 목록.""" 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 [] features = payload.get("features") if isinstance(payload, dict) else None return features if isinstance(features, list) else [] 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_line_to_box( line: list[tuple[float, float]], box: tuple[float, float, float, float] ) -> list[list[tuple[float, float]]]: """범위 안에 든 연속 구간만 조각으로 잘라 낸다(경계 한 점씩 물려 끊긴 티를 줄인다). 도엽 등고선 한 줄은 도엽 끝까지 이어지므로, 걸치면 통째로 남기면 도면이 A1을 넘는다. """ min_x, min_y, max_x, max_y = box runs: list[list[tuple[float, float]]] = [] current: list[tuple[float, float]] = [] for index, point in enumerate(line): x, y = point if min_x <= x <= max_x and min_y <= y <= max_y: if not current and index > 0: current.append(line[index - 1]) # 바깥쪽 직전 점까지 물린다 current.append(point) elif current: current.append(point) runs.append(current) current = [] if current: runs.append(current) return [run for run in runs if len(run) >= 2] def watershed_source(context: Any) -> dict[str, Any]: """유역도 입력(노선·세부유역·등고선·세류선)을 사업지 CRS(m)로 모은다. 저장본은 전부 WGS84라 여기서 미터 좌표로 되돌린다(B04가 저장할 때와 반대 방향). """ to_metric = Transformer.from_crs("EPSG:4326", f"EPSG:{context.epsg}", 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)) route_xy = [(vertex.x, vertex.y) for vertex in context.vertices] basins: list[dict[str, Any]] = [] for feature in _geojson_features(detail_basins_path(context.stored_path)): properties = feature.get("properties") or {} if properties.get("kind") != "detail_basin": continue rings = _geometry_lines(feature.get("geometry")) if not rings: continue basins.append({"ring": [metric(point) for point in rings[0]], "props": properties}) # 배경(등고선·세류선)은 **여러 도엽을 합쳐 받은 뒤 도곽 크기로 절취**한다 # (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 ( (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 (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 == 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