"""B05 경로 계산 후 종횡단을 생성하는 엔진 오케스트레이터. 확정 경로 GeoJSON과 확정 지표면 모델 sampler로 종단·횡단을 생성하고, 종단은 longitudinal/, 각 횡단은 cross_sections/ 아래 파일로 저장한다. DB 기록용 데이터(경로·요약·측점별 파일)를 준비해 반환한다. 라우터에서 asyncio.to_thread로 호출한다. """ import json import logging from dataclasses import replace from pathlib import Path from typing import Any from B05_Profile.B05_Profile_Engine_Grade import GradeDesignOptions, design_grade_line from B05_Profile.B05_Profile_Engine_Grade_Profile import ( design_alignment_profile, design_ground_following_profile, design_pipe_anchored_profile, ) from B05_Profile.B05_Profile_Engine_Sections_Core import ( SectionGenerationOptions, generate_sections, ) from B05_Profile.B05_Profile_Structures_Repository import load_structures from B05_Profile.B05_Profile_Structures_Schema import ( is_station_planting_type, structure_type_map, ) from common_util.common_util_drainage_pipes import ( PIPE_FACILITY_FORD_PAVEMENT, PIPE_FACILITY_PIPE, PIPE_FACILITY_REVET, PipePoint, load_pipe_points_file, pipe_anchor_clearances, route_signature, ) from common_util.common_util_json import atomic_write_json from common_util.common_util_route_geometry import RouteVertex from common_util.common_util_surface_sampler import build_surface_sampler from config.config_system import ( DRAINAGE_CACHE_DIRNAME, DRAINAGE_EDITS_DIRNAME, DRAINAGE_PIPE_POINTS_FILENAME, FOREST_ROAD_PROFILE_CRITERIA, PIPE_DEFAULT_DIAMETER_MM, PIPE_DEFAULT_TYPE, ) logger = logging.getLogger(__name__) _STAGE_SUBDIR = Path("B06_Section") _MODELS_SUBDIR = Path("B04_PreProcess") / "models" def _load_route_polyline(project_root: Path, route_data_path: str) -> list[list[float]]: """B05가 저장한 경로 GeoJSON에서 3D 폴리라인 좌표열을 로드한다.""" geojson_path = project_root / Path(route_data_path) if not geojson_path.is_file(): raise FileNotFoundError(f"경로 GeoJSON을 찾을 수 없습니다: {route_data_path}") geo = json.loads(geojson_path.read_text(encoding="utf-8")) coords = geo.get("geometry", {}).get("coordinates") if not coords or len(coords) < 2: raise ValueError("경로 GeoJSON에 유효한 LineString 좌표가 없습니다.") return [[float(c[0]), float(c[1]), float(c[2]) if len(c) > 2 else 0.0] for c in coords] def cross_filename(chainage_m: float) -> str: """측점 chainage에 대응하는 횡단면 파일명(단일 규칙).""" return f"cross_{int(round(float(chainage_m))):05d}m.json" def prune_stale_cross_files(cross_dir: Path, stations: list[Any]) -> set[str]: """stations에 없는 잔재 cross_*.json을 삭제하고 유효 파일명 집합을 반환한다. 측점 정보가 비어 있으면 오삭제를 피하기 위해 아무것도 지우지 않고 빈 집합을 반환한다(호출부는 빈 집합이면 필터를 생략한다). """ valid = { cross_filename(station["chainage_m"]) for station in stations if isinstance(station, dict) and isinstance(station.get("chainage_m"), (int, float)) } if not valid: return valid for path in cross_dir.glob("cross_*.json"): if path.name not in valid: path.unlink(missing_ok=True) return valid def _cross_summary(cross_section: dict[str, Any]) -> dict[str, Any]: """횡단면 상세에서 DB data 컬럼에 저장할 요약을 만든다.""" samples = cross_section.get("samples", []) valid_z = [ s["elevation_m"] for s in samples if s.get("valid") and s.get("elevation_m") is not None ] return { "chainage_m": cross_section.get("chainage_m"), "center_z": cross_section.get("center_z"), "azimuth_deg": cross_section.get("azimuth_deg"), "sample_count": len(samples), "min_elevation_m": min(valid_z) if valid_z else None, "max_elevation_m": max(valid_z) if valid_z else None, } def _load_pipe_points(project_root: Path, polyline: list[list[float]]) -> list[PipePoint]: """배수유역도가 확정한 계곡 통과 시설(관 지점 정본)을 읽는다. 계획선 정착과 구조물 측점 생성이 **같은 한 번의 읽기**를 쓴다 — 따로 읽으면 계획선이 물린 자리와 측점 자리가 어긋난다. 관 지점 파일에는 저장 당시 노선 지문이 함께 있다. 지문이 달라도 저장분에 좌표가 있으면 **이 계획선에 투영해 이월한다** — B04가 관 자리를 정한 선(계획노선 정본)과 여기 계획선은 같은 자리를 지나면서 연장이 다르다(실측 350.11m vs 354.83m). 그래서 지문은 거의 항상 달랐고 관이 통째로 빠졌다(2026-08-30 사용자 지적). 좌표가 없는 구 저장분만 버린다. """ vertices = [ RouteVertex( x=float(p[0]), y=float(p[1]), z=float(p[2]) if len(p) > 2 else 0.0, chainage_m=0.0 ) for p in polyline ] path = ( project_root / "B04_PreProcess" / DRAINAGE_CACHE_DIRNAME / DRAINAGE_EDITS_DIRNAME / DRAINAGE_PIPE_POINTS_FILENAME ) if not path.is_file(): return [] points = load_pipe_points_file(path, route_signature(vertices), vertices) if points is None: # 계획선 정착과 구조물 측점이 함께 빠지므로 남긴다(침묵 실패 금지). logger.warning("B05 계획선: 저장된 관 지점을 쓰지 못했습니다 (좌표 없는 구 저장분).") return [] return points def resolve_extra_stations( project_root: Path, pipes: list[PipePoint] ) -> tuple[tuple[float, str], ...]: """구조물(비정규) 측점의 **단일 공급자** — 관 정본 + 구조물 정본 A군에서 파생한다. 종전에는 이 목록을 서버에 알려 주는 곳이 프론트 [임시저장] 한 줄뿐이었다. 그래서 초기화·재생성이 종횡단 파일을 다시 쓸 때마다 구조물 측점이 통째로 사라졌다 (2026-08-28 사용자 보고). 정본 파일은 초기화에도 살아남으므로 여기서 매번 다시 만든다 — B05·B06은 같은 응답 하나를 보는 한 페이지라 공급자도 하나여야 한다. 물넘이포장은 비정규 측점으로 올리지 않는다 — 종단 그래프 하단 구조물 표시로만 다룬다 (2026-08-28 사용자 확정). 라벨 형식은 프론트 structureLabel()과 같아야 한다 — 재진입 이관이 라벨 문자열로 종류를 되짚기 때문이다(B05_Profile_Structures_Migration). """ types = structure_type_map() derived: list[tuple[float, str]] = [] for pipe in pipes: if pipe.facility == PIPE_FACILITY_FORD_PAVEMENT: continue if pipe.facility == PIPE_FACILITY_PIPE: values = pipe.options or {} kind = str(values.get("pipe_kind") or PIPE_DEFAULT_TYPE) diameter = int(float(values.get("pipe_diameter_mm") or PIPE_DEFAULT_DIAMETER_MM)) label = f"{kind} D{diameter}" else: definition = types.get(pipe.facility) if definition is None: continue label = definition.name # 독립 기슭막이는 구간(시작~종료)이라 **시작·기준·종료** 세 측점을 심는다 — # 길수록 횡단도가 여러 장 나온다(구 D군 구간형과 같은 규칙). 나머지 관 시설은 # 기준 한 곳(관이 노선을 가로지르는 점형). if pipe.facility == PIPE_FACILITY_REVET: marks = (pipe.start_m, pipe.chainage_m, pipe.end_m) else: marks = (pipe.chainage_m,) for mark in marks: if mark is not None: derived.append((float(mark), label)) # 구조물 정본에서 측점이 필요한 것들. 관 정본이 관리하는 타입(managed_by)은 # structures.json에 저장되지 않아 중복되지 않는다. # · 점형(A군 노출형 횡단수로·개거): 기준 측점 한 곳. # · 구간형(D군 기슭막이): **시작·기준·종료** — 길이가 길수록 횡단도가 여러 장 # 나와야 한다(2026-08-28 사용자). 격자와 같은 정수 미터면 generate_sections가 # 격자로 스냅해 파일이 겹치지 않는다. try: for structure in load_structures(str(project_root))[1]: definition = types.get(structure.type_id) if definition is None or definition.managed_by: continue if not is_station_planting_type(definition): continue if definition.placement == "interval": marks = (structure.start_m, structure.chainage_m, structure.end_m) else: marks = (structure.anchor_m(),) for mark in marks: if mark is not None: derived.append((float(mark), definition.name)) except Exception: logger.exception("B05: 구조물 정본을 읽지 못했습니다 (관 유래 측점만 씁니다)") return tuple(sorted(derived)) def _append_design_profiles( longitudinal: dict[str, Any], grade_options: GradeDesignOptions | None, station_interval_m: float | None = None, pipe_chainages: list[float] | None = None, pipe_clearances: dict[float, float] | None = None, ) -> dict[str, Any] | None: """종단 계획선을 산출해 longitudinal에 붙이고 요약을 반환한다. 1순위는 **전체 측점 폴리라인**(`design_ground_following_profile`) — 모든 측점의 계획고를 원지반고에 맞추고 종단곡선은 넣지 않는다(2026-09-02 사용자 확정). 실패하면 2순위 배관 정착 선형(`design_pipe_anchored_profile`, 옛 1차), 3순위 지반 추종 직선 분할 선형(`design_alignment_profile`), 그마저 실패하면 구 균형 최적화 계획선으로 폴백하고(편집 불가), 모두 실패해도 종횡단 생성 자체는 유지한다. """ longitudinal.setdefault("design_profiles", []) if grade_options is None: return None alignment = None profile = None try: alignment, profile = design_ground_following_profile( longitudinal, grade_options, station_interval_m=station_interval_m ) except (ValueError, KeyError, ArithmeticError): logger.exception("B05 전체 측점 폴리라인 산출 실패 — 배관 정착 선형으로 대체") if profile is None and pipe_chainages: try: alignment, profile = design_pipe_anchored_profile( longitudinal, grade_options, pipe_chainages, station_interval_m=station_interval_m, pipe_clearances=pipe_clearances, ) except (ValueError, KeyError, ArithmeticError): logger.exception("B05 배관 정착 계획선 산출 실패 — 직선 분할 선형으로 대체") if profile is None: try: alignment, profile = design_alignment_profile( longitudinal, grade_options, station_interval_m=station_interval_m ) except (ValueError, KeyError, ArithmeticError): logger.exception("B05 계획선 선형 산출 실패 — 균형 최적화 계획선으로 대체") try: profile = design_grade_line(longitudinal, grade_options) except (ValueError, KeyError, ArithmeticError): logger.exception("B05 종단 계획선 산출 실패 (종횡단은 유지)") return None if alignment is not None: longitudinal["profile_alignment"] = alignment longitudinal["design_profiles"].append(profile) return {"id": profile["id"], **profile["summary"]} def _annotate_pavement_suggestions( longitudinal: dict[str, Any], grade_options: GradeDesignOptions | None ) -> None: """계획선 국소 종단경사가 비포장 법정 상한을 넘는 측점에 포장 제안을 표기한다. 근거: 「임도설치 및 관리 등에 관한 규정」[별표 1-2] — 기준(설계속도×지형) 초과 구간은 노면포장 시에 한해 상한(`paved_exception_grade_pct`)까지 허용된다. stations에 `pavement_suggested`(bool)와 판정 근거(경사·상한)를 남기며, B06이 포장 기본값과 법정 근거 문구 표기에 그대로 사용한다. 실패해도 종횡단은 유지한다. """ if grade_options is None: return profiles = longitudinal.get("design_profiles") or [] samples = profiles[0].get("samples", []) if profiles else [] points = [ (float(s["chainage_m"]), float(s["elevation_m"])) for s in samples if isinstance(s.get("chainage_m"), (int, float)) and isinstance(s.get("elevation_m"), (int, float)) ] stations = longitudinal.get("stations") if len(points) < 2 or not isinstance(stations, list): return criteria = FOREST_ROAD_PROFILE_CRITERIA["design_speed"].get(grade_options.design_speed_kph) if not criteria: return terrain = grade_options.terrain_type if grade_options.terrain_type else "normal" unpaved_limit = float( criteria["max_grade_pct"].get(terrain, criteria["max_grade_pct"]["normal"]) ) def local_grade_pct(chainage: float) -> float: """측점을 감싸는 인접 계획선 구간들의 경사 중 최댓값(절댓값 %)을 돌려준다.""" worst = 0.0 for index in range(1, len(points)): c0, z0 = points[index - 1] c1, z1 = points[index] if c1 < chainage - 1e-6 or c0 > chainage + 1e-6: continue span = c1 - c0 if span <= 1e-9: continue worst = max(worst, abs((z1 - z0) / span) * 100.0) return worst for station in stations: chainage = station.get("chainage_m") if not isinstance(chainage, (int, float)): continue grade_pct = local_grade_pct(float(chainage)) station["pavement_suggested"] = grade_pct > unpaved_limit + 1e-6 station["pavement_grade_pct"] = round(grade_pct, 2) station["pavement_grade_limit_pct"] = round(unpaved_limit, 2) def run_section_generation( project_root: Path, route_data_path: str, filter_key: str, method: str, smooth: bool, *, options: SectionGenerationOptions | None = None, grade_options: GradeDesignOptions | None = None, grade_overrides: dict[str, Any] | None = None, crs: str | None = None, ) -> dict[str, Any]: """종횡단을 생성·저장하고 DB 기록용 데이터를 반환한다. 비정규(구조물) 측점은 정본 파일에서 해석하며 **호출자 옵션의 extra_stations는 무시한다** (단일 공급자 — resolve_extra_stations). 반환 dict: - longitudinal: {file_path, data(요약)} - cross_sections: [{chainage_m, sequence_num, data(요약), cross_section_file_path}] - result: generate_sections 원본 결과 """ polyline = _load_route_polyline(project_root, route_data_path) models_dir = project_root / _MODELS_SUBDIR sampler = build_surface_sampler(models_dir, filter_key, method, smooth) # 구조물(비정규) 측점의 단일 공급자 — 정본 파일에서 매번 다시 만든다. 호출자가 넘긴 # extra_stations는 쓰지 않는다: 이 옵션은 아래 long_summary["options"]로 DB에 되쓰이는 # 값이라, 되먹이면 B04에서 지운 관이 파일 정본을 이기고 되살아난다. pipes = _load_pipe_points(project_root, polyline) try: extras = resolve_extra_stations(project_root, pipes) except Exception: logger.exception("B05 구조물 측점 해석 실패 (규칙 격자만 생성)") extras = () logger.info("B05 종횡단: 구조물 측점 %d건 파생(관 %d건)", len(extras), len(pipes)) result = generate_sections( polyline, sampler, replace(options or SectionGenerationOptions(), extra_stations=extras), source_snapshot={"filter": filter_key, "method": method, "smooth": smooth}, crs=crs, ) stage_root = project_root / _STAGE_SUBDIR long_dir = stage_root / "longitudinal" cross_dir = stage_root / "cross_sections" long_dir.mkdir(parents=True, exist_ok=True) cross_dir.mkdir(parents=True, exist_ok=True) # 종단면 저장 (계획선은 저장 직전에 종단 데이터에 붙인다) grade_summary = _append_design_profiles( result["longitudinal"], grade_options, (result.get("options") or {}).get("station_interval_m"), # 1차 계획선의 변화점 = 배수유역도가 확정한 배관 배치 측점(측점 생성과 같은 목록). pipe_chainages=[pipe.chainage_m for pipe in pipes], # 시설 제원이 요구하는 최소 여유(관경+토피 등) — 계획선이 그만큼 들려야 관·구체가 # 들어갈 자리가 생긴다(2026-08-23 사용자 지시). 다만 이 들어올림이 배수 자리마다 # 계획고를 잡아당겨 제어가 어려워, 적용 여부를 사용자 스위치로 뺐다(2026-09-01 # 사용자 지시 — 기본 해제). 산식은 그대로이고 켜고 끄는 것만 바뀐다. pipe_clearances=( dict(pipe_anchor_clearances(pipes)) if grade_options and grade_options.enforce_pipe_clearance else None ), ) # 계획선 경사 기반 포장 제안(법정 상한 초과 측점) — 비치명적. try: _annotate_pavement_suggestions(result["longitudinal"], grade_options) except Exception: logger.exception("B05 포장 제안 판정 실패 (종횡단은 유지)") long_file = long_dir / "longitudinal.json" atomic_write_json(long_file, result["longitudinal"]) long_summary = { "length_m": result["longitudinal"]["length_m"], "station_count": result["summary"]["station_count"], "invalid_samples": result["summary"]["invalid_longitudinal_samples"], # 사용자 선택값의 단일 소스(DB): 재생성·재탐색 시 이 값을 우선 사용한다. "options": result["options"], # 해석이 끝난 기준값은 기록·표시용으로만 보관한다. 이 값을 재계산 시 # 폴백으로 되쓰면 등급·지형을 바꿔도 옛 기본값이 법정값을 이겨 갱신되지 않는다. "grade_options": grade_options.as_dict() if grade_options else None, # 재계산 폴백에 쓰는 단일 소스: 사용자가 명시적으로 입력한 값만 담는다. "grade_overrides": { key: value for key, value in (grade_overrides or {}).items() if value is not None }, "grade_summary": grade_summary, } # 측점별 횡단면 저장 (detail 조회가 폴더 전체를 glob하므로 이전 실행 잔재를 먼저 비운다) for stale_file in cross_dir.glob("cross_*.json"): stale_file.unlink() cross_records: list[dict[str, Any]] = [] for seq, cross_section in enumerate(result["cross_sections"]): chainage = float(cross_section["chainage_m"]) cross_file = cross_dir / cross_filename(chainage) atomic_write_json(cross_file, cross_section) cross_records.append( { "chainage_m": chainage, "sequence_num": seq, "data": _cross_summary(cross_section), "cross_section_file_path": cross_file.relative_to(project_root).as_posix(), } ) return { "longitudinal": { "file_path": long_file.relative_to(project_root).as_posix(), "data": long_summary, }, "cross_sections": cross_records, "result": result, } def generate_irregular_sections( project_root: Path, route_data_path: str, filter_key: str, method: str, smooth: bool, *, extra_stations: tuple[tuple[float, str], ...], options: SectionGenerationOptions | None = None, crs: str | None = None, ) -> list[dict[str, Any]]: """비정규 측점만 샘플링해 **횡단 파일을 쓰고** 측점 dict 목록을 돌려준다. 기존 종단 계획선·규칙 측점·횡단은 건드리지 않는다. 격자 측점과 똑같은 접선·오프셋 샘플링을 거치되(`generate_sections` 재사용) kind="irregular"인 것만 골라내 각각 `cross_*.json`으로 저장한다. B06 상세 조회가 이 폴더를 glob하고 종단 파일의 stations로 필터링하므로, 이 파일들 + 종단 stations 병합만으로 다음 페이지에 횡단이 나타난다(DB 불필요). """ if not extra_stations: return [] base = options or SectionGenerationOptions() merged_options = SectionGenerationOptions( station_interval_m=base.station_interval_m, cross_half_width_m=base.cross_half_width_m, cross_sample_interval_m=base.cross_sample_interval_m, long_sample_interval_m=base.long_sample_interval_m, include_endpoint=base.include_endpoint, extra_stations=tuple(extra_stations), ) polyline = _load_route_polyline(project_root, route_data_path) sampler = build_surface_sampler(project_root / _MODELS_SUBDIR, filter_key, method, smooth) result = generate_sections( polyline, sampler, merged_options, source_snapshot={"filter": filter_key, "method": method, "smooth": smooth}, crs=crs, ) irregular_stations = [ station for station in result["longitudinal"]["stations"] if station.get("kind") == "irregular" ] if not irregular_stations: return [] cross_dir = project_root / _STAGE_SUBDIR / "cross_sections" cross_dir.mkdir(parents=True, exist_ok=True) for cross_section in result["cross_sections"]: if cross_section.get("kind") != "irregular": continue cross_file = cross_dir / cross_filename(float(cross_section["chainage_m"])) atomic_write_json(cross_file, cross_section) return irregular_stations