diff --git a/B05_Profile/B05_Profile_Engine_Sections.py b/B05_Profile/B05_Profile_Engine_Sections.py index bbaf3707f..108484692 100644 --- a/B05_Profile/B05_Profile_Engine_Sections.py +++ b/B05_Profile/B05_Profile_Engine_Sections.py @@ -23,6 +23,17 @@ from B05_Profile.B05_Profile_Engine_Sections_Core import ( SectionGenerationOptions, generate_sections, ) +from B05_Profile.B05_Profile_Engine_Sections_Names import ( # noqa: F401 — 부르는 곳이 여기서 씀 + CROSS_KEY_PATTERN, + cross_filename, + cross_files_in_order, + cross_key, + cross_station, + find_cross_file, + parse_cross_key, + prune_stale_cross_files, + valid_cross_names, +) from B05_Profile.B05_Profile_Structures_Repository import load_structures from B05_Profile.B05_Profile_Structures_Schema import ( structure_type_map, @@ -118,82 +129,6 @@ def _read_design_curve_table(project_root: Path) -> dict[str, Any]: return data if isinstance(data, dict) else {} -# ── 횡단 이름 규칙(한 벌) — 파일 이름 · B07 도면 id · 찾기가 모두 이것을 씀 ──────────── -#: 이름 속 측점 글 — 정수 m `00120` · 소수 측점 `00119.80` -CROSS_KEY_PATTERN = r"(\d+(?:\.\d{2})?)" - - -def cross_station(chainage_m: float) -> int | float: - """측점 → 이름 열쇠 값 — 정수 m(0.005m 안)면 int(옛 이름 · 키 그대로) · 아니면 소수 2자리.""" - value = round(float(chainage_m), 2) - whole = round(value) - return int(whole) if abs(value - whole) < 0.005 else value - - -def cross_key(chainage_m: float) -> str: - """이름 속 측점 글 — `00120`(정수 · 옛 이름 그대로) · `00119.80`(소수 측점).""" - station = cross_station(chainage_m) - return f"{station:05d}" if isinstance(station, int) else f"{station:08.2f}" - - -def parse_cross_key(text: str) -> int | float: - """`cross_key` 되읽기.""" - return cross_station(float(text)) - - -def cross_filename(chainage_m: float) -> str: - """측점 chainage에 대응하는 횡단면 파일명(단일 규칙).""" - return f"cross_{cross_key(chainage_m)}m.json" - - -def find_cross_file(cross_dir: Path, chainage_m: float) -> Path: - """읽기용 — 지금 이름이 없으면 옛 정수 m 이름(소수 측점이 정수로 붙던 옛 프로젝트).""" - path = cross_dir / cross_filename(chainage_m) - legacy = cross_dir / f"cross_{int(round(float(chainage_m))):05d}m.json" - return path if path.exists() or not legacy.exists() else legacy - - -def valid_cross_names(cross_dir: Path, stations: list[Any]) -> set[str]: - """stations 에 맞는 cross_*.json 파일명 집합 — 읽기 전용(지우지 않음 · GET 이 씀). - - 측점 정보가 비면 빈 집합(호출부는 빈 집합이면 필터를 생략한다). - """ - # 옛 프로젝트는 소수 측점 파일이 정수 m 이름이라 찾기 규칙(find_cross_file)으로 가림 - return { - find_cross_file(cross_dir, station["chainage_m"]).name - for station in stations - if isinstance(station, dict) and isinstance(station.get("chainage_m"), (int, float)) - } - - -def cross_files_in_order(cross_dir: Path, stations: list[Any]) -> list[Path]: - """잔재를 걸러 읽을 cross_*.json — **측점 차례**(이름 차례면 `01080.16` 이 `01080` 앞).""" - valid = valid_cross_names(cross_dir, stations) - - def chainage(path: Path) -> tuple[float, str]: - match = re.fullmatch(rf"cross_{CROSS_KEY_PATTERN}m", path.stem) - return (float(match.group(1)) if match else float("inf"), path.name) - - files = [path for path in cross_dir.glob("cross_*.json") if not valid or path.name in valid] - return sorted(files, key=chainage) - - -def prune_stale_cross_files(cross_dir: Path, stations: list[Any]) -> set[str]: - """stations에 없는 잔재 cross_*.json을 삭제하고 유효 파일명 집합을 반환한다. - - 쓰는 쪽(측점 만들기 · 측점 목록 고침)만 부른다 — 조회(GET)는 `valid_cross_names`. - 측점 정보가 비어 있으면 오삭제를 피하기 위해 아무것도 지우지 않고 - 빈 집합을 반환한다(호출부는 빈 집합이면 필터를 생략한다). - """ - valid = valid_cross_names(cross_dir, stations) - 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", []) @@ -297,7 +232,12 @@ def relabel_pipe_stations( def resolve_extra_stations( - project_root: Path, pipes: list[PipePoint], xy: Sequence[Sequence[float]] | None = None + project_root: Path, + pipes: list[PipePoint], + xy: Sequence[Sequence[float]] | None = None, + *, + structures: list[Any] | None = None, + plus: list[float] | None = None, ) -> tuple[tuple[float, str], ...]: """구조물(비정규) 측점의 **단일 공급자** — 관 정본 + 구조물 정본 A군에서 파생한다. @@ -314,12 +254,14 @@ def resolve_extra_stations( 거울 시험 = `resources/tester/test_322_3_plan_mirror.py` · `test_272_1_station_list.py`. 추가 측점(368-2) = `plus_stations.json` · 이름 "". `xy` = 지금 노선 — 노선 밖(시점 앞 · 종점 뒤 · `route_outside`) 관 · 구조물은 뺌(459). + `structures` · `plus` = 파일 대신 따라간 사본(476 추천 초기 종단 · 파일 안 씀). """ from B05_Profile.B05_Profile_Plus_Stations import plus_chainages from common_util.common_util_route_follow import route_outside try: - structures = load_structures(str(project_root))[1] + if structures is None: + structures = load_structures(str(project_root))[1] except Exception: logger.exception("B05: 구조물 정본을 읽지 못했습니다 (관 유래 측점만 씁니다)") structures = [] @@ -332,7 +274,9 @@ def resolve_extra_stations( ] return tuple( (float(station["chainage_m"]), str(station["title"])) - for station in plan_stations(pipes, structures, plus_chainages(project_root)) + for station in plan_stations( + pipes, structures, plus_chainages(project_root) if plus is None else plus + ) ) @@ -450,6 +394,55 @@ def _annotate_pavement_suggestions( station["pavement_grade_limit_pct"] = round(unpaved_limit, 2) +def build_sections( + polyline: list[list[float]], + sampler: Any, + pipes: list[PipePoint], + extras: tuple[tuple[float, str], ...], + *, + options: SectionGenerationOptions | None, + grade_options: GradeDesignOptions | None, + source_snapshot: dict[str, Any], + crs: str | None, + design_curves: list[dict[str, Any]], + station_snap: bool, + station_snap_step: float | None, +) -> tuple[dict[str, Any], dict[str, Any] | None]: + """측점 틀 · 지반 · 1차 계획선(438) · 포장 제안 — 파일 안 씀 · (결과, 계획선 요약). + + 종횡단 생성(`run_section_generation`)과 추천 후보 초기 종단(476 · `..._RouteInitial_Initial`) + 이 같은 셈 한 벌.""" + result = generate_sections( + polyline, + sampler, + replace(options or SectionGenerationOptions(), extra_stations=extras), + source_snapshot=source_snapshot, + crs=crs, + design_curves=design_curves, + station_snap=station_snap, + station_snap_step=station_snap_step, + ) + 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], + # 시설 제원 최소 여유(관경+토피 등)로 계획선 들어올림 — 사용자 스위치(기본 해제 · 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 포장 제안 판정 실패 (종횡단은 유지)") + return result, grade_summary + + def run_section_generation( project_root: Path, route_data_path: str, @@ -487,10 +480,13 @@ def run_section_generation( extras = () logger.info("B05 종횡단: 구조물 측점 %d건 파생(관 %d건)", len(extras), len(pipes)) - result = generate_sections( + result, grade_summary = build_sections( polyline, sampler, - replace(options or SectionGenerationOptions(), extra_stations=extras), + pipes, + extras, + options=options, + grade_options=grade_options, source_snapshot={"filter": filter_key, "method": method, "smooth": smooth}, crs=crs, # 곡선부 확폭의 정본 — 설계 곡선표(시·종점·반지름)를 그대로 쓴다. 없으면 빈 목록이라 @@ -505,29 +501,6 @@ def run_section_generation( 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 = { diff --git a/B05_Profile/B05_Profile_Engine_Sections_Names.py b/B05_Profile/B05_Profile_Engine_Sections_Names.py new file mode 100644 index 000000000..60605bb82 --- /dev/null +++ b/B05_Profile/B05_Profile_Engine_Sections_Names.py @@ -0,0 +1,82 @@ +"""횡단 이름 규칙 한 벌 — 파일 이름 · B07 도면 id · 찾기 · 잔재 거름. + +`B05_Profile_Engine_Sections` 에서 나눔(700줄) · 그 모듈이 다시 내보냄.""" + +import re +from pathlib import Path +from typing import Any + +# ── 횡단 이름 규칙(한 벌) — 파일 이름 · B07 도면 id · 찾기가 모두 이것을 씀 ──────────── +#: 이름 속 측점 글 — 정수 m `00120` · 소수 측점 `00119.80` +CROSS_KEY_PATTERN = r"(\d+(?:\.\d{2})?)" + + +def cross_station(chainage_m: float) -> int | float: + """측점 → 이름 열쇠 값 — 정수 m(0.005m 안)면 int(옛 이름 · 키 그대로) · 아니면 소수 2자리.""" + value = round(float(chainage_m), 2) + whole = round(value) + return int(whole) if abs(value - whole) < 0.005 else value + + +def cross_key(chainage_m: float) -> str: + """이름 속 측점 글 — `00120`(정수 · 옛 이름 그대로) · `00119.80`(소수 측점).""" + station = cross_station(chainage_m) + return f"{station:05d}" if isinstance(station, int) else f"{station:08.2f}" + + +def parse_cross_key(text: str) -> int | float: + """`cross_key` 되읽기.""" + return cross_station(float(text)) + + +def cross_filename(chainage_m: float) -> str: + """측점 chainage에 대응하는 횡단면 파일명(단일 규칙).""" + return f"cross_{cross_key(chainage_m)}m.json" + + +def find_cross_file(cross_dir: Path, chainage_m: float) -> Path: + """읽기용 — 지금 이름이 없으면 옛 정수 m 이름(소수 측점이 정수로 붙던 옛 프로젝트).""" + path = cross_dir / cross_filename(chainage_m) + legacy = cross_dir / f"cross_{int(round(float(chainage_m))):05d}m.json" + return path if path.exists() or not legacy.exists() else legacy + + +def valid_cross_names(cross_dir: Path, stations: list[Any]) -> set[str]: + """stations 에 맞는 cross_*.json 파일명 집합 — 읽기 전용(지우지 않음 · GET 이 씀). + + 측점 정보가 비면 빈 집합(호출부는 빈 집합이면 필터를 생략한다). + """ + # 옛 프로젝트는 소수 측점 파일이 정수 m 이름이라 찾기 규칙(find_cross_file)으로 가림 + return { + find_cross_file(cross_dir, station["chainage_m"]).name + for station in stations + if isinstance(station, dict) and isinstance(station.get("chainage_m"), (int, float)) + } + + +def cross_files_in_order(cross_dir: Path, stations: list[Any]) -> list[Path]: + """잔재를 걸러 읽을 cross_*.json — **측점 차례**(이름 차례면 `01080.16` 이 `01080` 앞).""" + valid = valid_cross_names(cross_dir, stations) + + def chainage(path: Path) -> tuple[float, str]: + match = re.fullmatch(rf"cross_{CROSS_KEY_PATTERN}m", path.stem) + return (float(match.group(1)) if match else float("inf"), path.name) + + files = [path for path in cross_dir.glob("cross_*.json") if not valid or path.name in valid] + return sorted(files, key=chainage) + + +def prune_stale_cross_files(cross_dir: Path, stations: list[Any]) -> set[str]: + """stations에 없는 잔재 cross_*.json을 삭제하고 유효 파일명 집합을 반환한다. + + 쓰는 쪽(측점 만들기 · 측점 목록 고침)만 부른다 — 조회(GET)는 `valid_cross_names`. + 측점 정보가 비어 있으면 오삭제를 피하기 위해 아무것도 지우지 않고 + 빈 집합을 반환한다(호출부는 빈 집합이면 필터를 생략한다). + """ + valid = valid_cross_names(cross_dir, stations) + 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 diff --git a/B05_Profile/B05_Profile_Plus_Stations.py b/B05_Profile/B05_Profile_Plus_Stations.py index 7f21321a2..911c93462 100644 --- a/B05_Profile/B05_Profile_Plus_Stations.py +++ b/B05_Profile/B05_Profile_Plus_Stations.py @@ -75,6 +75,14 @@ def follow_plus_stations(project_root: str | Path, new_xy: Sequence[Sequence[flo rows = load_plus_stations(project_root) if not rows: return 0 + followed = followed_plus_rows(rows, new_xy) + if followed["moved"]: + atomic_write_json(str(plus_stations_path(project_root)), {"stations": followed["rows"]}) + return int(followed["moved"]) + + +def followed_plus_rows(rows: list[dict[str, Any]], new_xy: Sequence[Sequence[float]]) -> dict: + """추가 측점을 새 노선에 투영(저장 안 함) → `{rows, moved}` — 체인 · 추천 초기 종단(476).""" results = run_calc( ["facility.plus_follow"], { @@ -89,7 +97,4 @@ def follow_plus_stations(project_root: str | Path, new_xy: Sequence[Sequence[flo ) if results is None: raise RuntimeError("추가 측점 따라감 실패 — 계산 Node 입구(facility.plus_follow)") - followed = results["facility.plus_follow"] - if followed["moved"]: - atomic_write_json(str(plus_stations_path(project_root)), {"stations": followed["rows"]}) - return int(followed["moved"]) + return results["facility.plus_follow"] diff --git a/B05_Profile/B05_Profile_Route_Save.py b/B05_Profile/B05_Profile_Route_Save.py index d309e903f..fda51025b 100644 --- a/B05_Profile/B05_Profile_Route_Save.py +++ b/B05_Profile/B05_Profile_Route_Save.py @@ -22,7 +22,7 @@ API — `POST /api/projects/{id}/route/replan`(`B05_Profile_Router_Replan.replan (`B05_Profile_Route_Save_Parts`) · 배수유역 저장분은 노선 지문이 달라 「낡음」(B04 `_resolve`). - 계획고 = 바뀐 자리만 새 지반 첫 계획선 · 같은 자리(옆 1 m 안) 사용자 계획고 지킴 (`B05_Profile_Engine_Grade_Carry` · 옛 · 새 **계획 폴리라인** — 화면 planline 과 같은 선) · - 추천 [적용](`as_initial`) = 옮기지 않음 — 균형 종단(`apply_profile_target`) 또는 1차 계획선(438). + 추천 [적용](`as_initial`) = 옮기지 않음 — 균형 종단(`apply_profile_target`) · 1차 계획선(438). - 초기값은 여기서 안 뜸 — 부르는 쪽이 `as_initial` 일 때만(자동설계 체인 직후 + 추천 [적용]). """ @@ -165,6 +165,31 @@ def _xy(project_root: Path, route_data_path: str | None) -> list[list[float]]: return [] +def followed_pipes( + old_pipes: list, old_xy: list[list[float]], points: list[dict[str, float]], streams: list +) -> tuple[list, int, list]: + """관 따라감 + 새 세류선 교차점 기본 관(저장 안 함) → (짝, 교차 수, 새 노선 정점) — 체인 · + 추천 초기 종단(476) 한 벌.""" + from B04_PreProcess.B04_PreProcess_Engine_Watershed_Analyze import _base_pipes + from common_util.common_util_drainage_pipes import fill_pipe_coordinates + + vertices = vertices_of([(p["x"], p["y"]) for p in points]) + fill_pipe_coordinates(old_pipes, vertices_of(old_xy)) + pairs = follow_pipes(old_pipes, vertices) + total = float(vertices[-1].chainage_m) + crossing = add_crossing_pipes(pairs, _base_pipes(vertices, streams), old_xy, total) + return pairs, crossing, vertices + + +def section_seed(request: Any, old_section_options: dict[str, Any] | None) -> dict[str, Any]: + """체인 종횡단 생성 옵션(건너뛴 첫 생성이 남겼을 값) — 체인 · 추천 초기 종단(476) 한 벌.""" + from B05_Profile.B05_Profile_Router import _section_options + + seed = asdict(_section_options(request, old_section_options)) + seed.pop("extra_stations", None) + return seed + + async def _place_pipes( project_id: UUID, stored_path: str, @@ -178,22 +203,13 @@ async def _place_pipes( 노선은 범위를 안 잘라 누가거리가 갈림 · 세류선만 거기서. 좌표 없는 옛 관은 옛 노선 자리로 먼저 채움(459). """ - from B04_PreProcess.B04_PreProcess_Engine_Watershed_Analyze import _base_pipes from B04_PreProcess.B04_PreProcess_Router_Watershed import _prepare - from common_util.common_util_drainage_pipes import ( - fill_pipe_coordinates, - route_signature, - save_pipe_points, - ) + from common_util.common_util_drainage_pipes import route_signature, save_pipe_points from common_util.common_util_route_follow import route_outside prepared = await _prepare(project_id) streams = [] if isinstance(prepared, JSONResponse) else prepared["streams"] # 도엽 없음 - vertices = vertices_of([(p["x"], p["y"]) for p in points]) - fill_pipe_coordinates(old_pipes, vertices_of(old_xy)) - pairs = follow_pipes(old_pipes, vertices) - total = float(vertices[-1].chainage_m) - crossing = add_crossing_pipes(pairs, _base_pipes(vertices, streams), old_xy, total) + pairs, crossing, vertices = followed_pipes(old_pipes, old_xy, points, streams) await asyncio.to_thread( save_pipe_points, stored_path, @@ -259,7 +275,7 @@ async def _carry_profile( def carries_cross_inputs(initial: bool) -> bool: - """횡단 손 입력을 새 측점(같은 자리)에 넘기나 — 추천 [적용]도 넘김(475-3 · 사용자 답 대기 476).""" + """횡단 손 입력을 새 측점(같은 자리)에 넘기나 — [적용]도 넘김(475-3 · 사용자 답 대기 476).""" return True @@ -376,11 +392,7 @@ async def run_route_save_chain( from B03_FileInput.B03_FileInput_Service_Chain import _log_steps, redesign_solve_request from B05_Profile.B05_Profile_Engine_Sections import _STAGE_SUBDIR from B05_Profile.B05_Profile_Plus_Stations import follow_plus_stations, plus_stations_path - from B05_Profile.B05_Profile_Router import ( - _section_options, - confirm_latest_route, - solve_route_record, - ) + from B05_Profile.B05_Profile_Router import confirm_latest_route, solve_route_record from B05_Profile.B05_Profile_Structures_Repository import load_structures, structures_file_path from B05_Profile.B05_Profile_Structures_Router import follow_structures_to_route from B06_Section.B06_Section_Repository import ( @@ -456,8 +468,7 @@ async def run_route_save_chain( pairs, pipes = await _place_pipes(project_id, stored_path, old_pipes, old_xy, points) marks.append(("관 따라감 · 교차점", time.perf_counter())) - seed = asdict(_section_options(request, old_section_options)) - seed.pop("extra_stations", None) + seed = section_seed(request, old_section_options) made = await regenerate_route_sections( project_id, new_route_id, float(seed["cross_half_width_m"]), stored_options=seed ) diff --git a/B05_Profile/B05_Profile_Router_RouteInitial.py b/B05_Profile/B05_Profile_Router_RouteInitial.py index c2936fbd2..19579feec 100644 --- a/B05_Profile/B05_Profile_Router_RouteInitial.py +++ b/B05_Profile/B05_Profile_Router_RouteInitial.py @@ -355,15 +355,22 @@ async def refresh_expected_route(project_id: UUID) -> None: def _set_base_pvi(path: Path, stations: list[float], plan: list[float]) -> int: """종단 정본의 1차 변화점(`base_pvi`) 계획고를 `stations` · `plan` 직선 보간으로 바꿔 쓴다.""" - import json + from common_util.common_util_json import atomic_write_json + longitudinal = json.loads(path.read_text(encoding="utf-8")) + count = seat_base_pvi(longitudinal, stations, plan) + atomic_write_json(path, longitudinal) + return count + + +def seat_base_pvi(longitudinal: dict[str, Any], stations: list[float], plan: list[float]) -> int: + """1차 변화점 계획고를 균형 종단에 얹음(제자리 · 파일 안 씀) — [적용] 체인 · 추천 초기 종단(476) + 한 벌.""" import numpy as np from B05_Profile.B05_Profile_Engine_Grade_Alignment import AlignmentPolicy from B05_Profile.B05_Profile_Engine_RouteInitial_Profile import held_pass_through - from common_util.common_util_json import atomic_write_json - longitudinal = json.loads(path.read_text(encoding="utf-8")) alignment = longitudinal["profile_alignment"] base = alignment["base_pvi"] base_s = np.array([float(item["chainage_m"]) for item in base]) @@ -374,7 +381,6 @@ def _set_base_pvi(path: Path, stations: list[float], plan: list[float]) -> int: ) for item, value in zip(base, base_z, strict=True): item["elevation_m"] = round(float(value), 6) # 직선 판정(426) 1e-5 m 안 - atomic_write_json(path, longitudinal) return len(base) diff --git a/B05_Profile/B05_Profile_Structures_Router.py b/B05_Profile/B05_Profile_Structures_Router.py index 9b1059aa3..f639d071b 100644 --- a/B05_Profile/B05_Profile_Structures_Router.py +++ b/B05_Profile/B05_Profile_Structures_Router.py @@ -121,6 +121,26 @@ async def _route_xy(project_id: UUID, root: str) -> list[list[float]] | None: return None +def followed_structures( + project_root: Path, + items: list[Any], + old_route_data_path: str | None, + new_xy: list[dict[str, float]], +) -> list[Any]: + """구조물을 새 노선 땅 위 자리로 옮김(제자리 · 저장 안 함) — 노선 저장 체인 · 추천 초기 + 종단(476) 한 벌. 좌표 없는 옛 자료는 옛 노선으로 먼저 채움.""" + from B05_Profile.B05_Profile_Engine_Sections import _load_route_polyline + + if old_route_data_path: + try: + fill_structure_coordinates( + items, _load_route_polyline(project_root, old_route_data_path), only_missing=True + ) + except (OSError, ValueError): + logger.warning("B05 구조물 따라감: 옛 노선을 못 읽어 좌표 없는 구조물은 그대로") + return follow_structures(items, [(point["x"], point["y"]) for point in new_xy]) + + def follow_structures_to_route( project_root: Path, old_route_data_path: str | None, new_xy: list[dict[str, float]] ) -> dict[str, int]: @@ -129,21 +149,12 @@ def follow_structures_to_route( 좌표가 없는 옛 자료는 옛 노선(아직 안 바뀐 경로 파일)으로 먼저 채운다. 실패는 비치명 — 구조물은 옛 누가거리 그대로 남는다(지금까지와 같음). """ - from B05_Profile.B05_Profile_Engine_Sections import _load_route_polyline - root = str(project_root) revision, items = load_structures(root) if not items: return {"moved": 0, "far": 0} - if old_route_data_path: - try: - fill_structure_coordinates( - items, _load_route_polyline(project_root, old_route_data_path), only_missing=True - ) - except (OSError, ValueError): - logger.warning("B05 구조물 따라감: 옛 노선을 못 읽어 좌표 없는 구조물은 그대로") before = [(item.chainage_m, item.start_m, item.end_m) for item in items] - result = follow_structures(items, [(point["x"], point["y"]) for point in new_xy]) + result = followed_structures(project_root, items, old_route_data_path, new_xy) moved = sum( before[index] != (item.chainage_m, item.start_m, item.end_m) for index, item in enumerate(items)