"""B05 종단 계획선 선형 산출 오케스트레이터 (지반 추종 직선 분할 + 편집 재구성). [[B05_wf2_Route_Engine_Grade]] 가 확정한 설계 기준과 [[B05_wf2_Route_Engine_Grade_Solver]] 의 수치 계산, [[B05_wf2_Route_Engine_Grade_Alignment]] 의 기하 파생을 묶어 `design_profiles` 배열에 넣을 계획선 한 벌을 만든다. 두 진입점이 있다. - `design_alignment_profile()` : 노선 계산 직후. 직선 분할 DP부터 새로 푼다. - `rebuild_alignment_profile()`: 사용자 편집 확정 시. **저장된 자동 선형(base_pvi)과 정책을 그대로 재사용**하고 편집 델타만 다시 얹는다. DP를 다시 돌리면 기준선이 흔들려 "원복" 이 원래 위치로 돌아가지 않기 때문이다. """ from collections import Counter from typing import Any import numpy as np from B05_wf2_Route.B05_wf2_Route_Engine_Grade import ( GradeDesignOptions, detect_main_direction, ground_profile, ) from B05_wf2_Route.B05_wf2_Route_Engine_Grade_Alignment import ( ALIGNMENT_SCHEMA_VERSION, AlignmentPolicy, build_alignment, ) from B05_wf2_Route.B05_wf2_Route_Engine_Grade_Solver import ( grade_limits, integration_weights, solve_alignment_elevations, station_breakpoints, ) ALIGNMENT_PROFILE_ID = "design_grade_line" ALIGNMENT_BASIS = "station_alignment" def infer_station_interval(stations: list[dict[str, Any]]) -> float: """측점 목록에서 가장 흔한 간격을 기준 측점간격으로 본다. 사용자가 추가한 비기준 측점(+18, +15 등)이 섞여 있어도 최빈값이 기준 간격이다. """ gaps: list[float] = [] for previous, current in zip(stations[:-1], stations[1:]): gap = round(float(current["chainage_m"]) - float(previous["chainage_m"]), 1) if gap > 0: gaps.append(gap) if not gaps: return 20.0 return float(Counter(gaps).most_common(1)[0][0]) def _profile_entry( alignment: dict[str, Any], options: GradeDesignOptions, direction: str, balanced: bool, warnings: list[str], ) -> dict[str, Any]: """`design_profiles` 배열 계약(기존 스키마)에 맞춰 계획선 한 벌을 만든다. B06 횡단 계획고와 B07 CAD 계획선 레이어는 `samples`만 사용하므로, 선형 구조가 바뀌어도 하류 단계는 그대로 동작한다. """ balance = alignment["balance"] grades = [abs(segment["grade_percent"]) for segment in alignment["segments"]] return { "schema_version": ALIGNMENT_SCHEMA_VERSION, "id": ALIGNMENT_PROFILE_ID, "name": "계획선", "basis": ALIGNMENT_BASIS, "criteria": {**options.as_dict(), "resolved_main_direction": direction}, # 구 스키마 호환: 변화점 목록을 pvis 이름으로도 노출한다. "pvis": alignment["pvi"], "samples": alignment["samples"], "stations": alignment["stations"], "balance_segments": [ { "index": 0, "start_chainage_m": alignment["segments"][0]["from_m"] if alignment["segments"] else 0.0, "end_chainage_m": alignment["segments"][-1]["to_m"] if alignment["segments"] else 0.0, "cut_area_m2": balance["cut_area_m2"], "fill_area_m2": balance["fill_area_m2"], "balance_error_m2": balance["net_area_m2"], } ], "summary": { "cut_area_m2": balance["cut_area_m2"], "fill_area_m2": balance["fill_area_m2"], "balance_error_m2": balance["net_area_m2"], "imbalance_percent": balance["imbalance_percent"], "tolerance_percent": balance["tolerance_percent"], "max_grade_pct": round(max(grades), 6) if grades else 0.0, "vertical_curve_count": sum(1 for curve in alignment["curves"] if not curve["omitted"]), "pvi_count": len(alignment["pvi"]), "balance_segment_count": 1, "balanced": bool(balanced and balance["within_tolerance"]), "main_direction": direction, "suggested_elevation_offset_m": None, "edited_station_count": len(alignment["edits"]["station_offsets"]), "warnings": warnings, }, } def design_alignment_profile( longitudinal: dict[str, Any], options: GradeDesignOptions, *, station_interval_m: float | None = None, edits: dict[str, Any] | None = None, ) -> tuple[dict[str, Any], dict[str, Any]]: """지반 종단을 직선 분할로 근사해 계획선 선형을 새로 만든다. 반환값은 (선형 구조, `design_profiles` 항목) 이다. """ options.validate() chainage, ground = ground_profile(longitudinal) total = float(chainage[-1]) if total <= 0: raise ValueError("종단 연장이 0이어서 계획선을 만들 수 없습니다.") stations = list(longitudinal.get("stations") or []) interval = float(station_interval_m or 0) or infer_station_interval(stations) policy = AlignmentPolicy.from_config( station_interval_m=interval, max_grade_pct=options.max_grade_pct, curve_skip_delta_pct=options.vertical_curve_skip_delta_pct, paved=options.paved, ) warnings = list(options.warnings) fixed = ( float(ground[0]) + options.start_elevation_offset_m, float(ground[-1]) + options.end_elevation_offset_m, ) rise = fixed[1] - fixed[0] direction, note = ( detect_main_direction(ground, rise) if options.main_direction == "auto" else (options.main_direction, None) ) if note: warnings.append(note) up_limit, down_limit = grade_limits( options.max_grade_pct, options.max_reverse_grade_pct, direction ) limit = up_limit if rise >= 0 else down_limit if abs(rise) / total > limit + 1e-9: raise ValueError( f"시·종점 고도차({rise:.2f}m)를 연장 {total:.1f}m에서 기준 기울기 " f"{limit * 100:.1f}% 이내로 연결할 수 없습니다." ) station_chainages = np.array( [float(station["chainage_m"]) for station in stations], dtype=np.float64 ) base_s = station_breakpoints( chainage, ground, station_chainages if len(station_chainages) else chainage, penalty_m2=policy.pvi_penalty_m2, min_segment_stations=policy.min_segment_stations, ) base_z, balanced = solve_alignment_elevations( base_s, chainage, ground, integration_weights(chainage), fixed, up_limit, down_limit, balance_tolerance_percent=policy.balance_tolerance_percent, ) alignment = build_alignment( base_s=base_s, base_z=base_z, chainage=chainage, ground=ground, stations=stations, policy=policy, edits=edits, ) warnings.extend(alignment["warnings"]) return alignment, _profile_entry(alignment, options, direction, balanced, warnings) def rebuild_alignment_profile( longitudinal: dict[str, Any], edits: dict[str, Any] | None, ) -> tuple[dict[str, Any], dict[str, Any]]: """저장된 자동 선형을 기준으로 사용자 편집만 다시 얹는다. 직선 분할 DP를 다시 돌리지 않으므로 편집 델타의 기준선(base_pvi)이 고정되고, 편집 키를 지우면 최초 자동 선형으로 정확히 되돌아간다. """ stored = longitudinal.get("profile_alignment") if not isinstance(stored, dict) or not stored.get("base_pvi"): raise ValueError("저장된 계획선 선형이 없어 편집을 반영할 수 없습니다.") chainage, ground = ground_profile(longitudinal) base = stored["base_pvi"] base_s = np.array([float(item["chainage_m"]) for item in base], dtype=np.float64) base_z = np.array([float(item["elevation_m"]) for item in base], dtype=np.float64) policy = AlignmentPolicy.from_dict(stored.get("policy") or {}) alignment = build_alignment( base_s=base_s, base_z=base_z, chainage=chainage, ground=ground, stations=list(longitudinal.get("stations") or []), policy=policy, edits=edits, ) previous = (longitudinal.get("design_profiles") or [{}])[0] criteria = previous.get("criteria") or {} options = GradeDesignOptions( max_grade_pct=float(criteria.get("max_grade_pct") or policy.max_grade_pct), max_reverse_grade_pct=float(criteria.get("max_reverse_grade_pct") or 5.0), min_vertical_radius_m=float(criteria.get("min_vertical_radius_m") or 100.0), min_curve_length_m=float(criteria.get("min_curve_length_m") or 20.0), min_tangent_length_m=float(criteria.get("min_tangent_length_m") or 20.0), vertical_curve_skip_delta_pct=policy.curve_skip_delta_pct, design_speed_kph=int(criteria.get("design_speed_kph") or 20), terrain_type=str(criteria.get("terrain_type") or "normal"), paved=policy.paved, main_direction=str(criteria.get("main_direction") or "auto"), ) direction = str(criteria.get("resolved_main_direction") or "none") return alignment, _profile_entry( alignment, options, direction, alignment["balance"]["within_tolerance"], alignment["warnings"], )