diff --git a/B03_FileInput/B03_FileInput_Service_Chain.py b/B03_FileInput/B03_FileInput_Service_Chain.py index a2f589bea..a85f53710 100644 --- a/B03_FileInput/B03_FileInput_Service_Chain.py +++ b/B03_FileInput/B03_FileInput_Service_Chain.py @@ -392,6 +392,53 @@ async def _ensure_initial_polyline( logger.exception("초기 계획노선 폴리라인 생성 실패(계속 진행): %s", project_id) +def redesign_solve_request( + params: dict[str, Any], selection: dict[str, Any], surface_model_id: int +) -> Any: + """stage 2 params(사용자 저장분) + 새 확정 지표면 → as_planned 풀기 요청(노선 저장도 씀).""" + from B05_Profile.B05_Profile_Schema import RouteSolveRequest + + points = params.get("points") or {} + options = params.get("options") or {} + return RouteSolveRequest( + filter_key=str(selection.get("source_filter") or params.get("filter_key")), + method=str(selection.get("method") or params.get("method") or "dtm"), + smooth=bool(selection.get("smooth", params.get("smooth", False))), + surface_model_id=surface_model_id, + # 사용자가 고친 계획노선을 **그대로** 쓴다 — 다시 풀면 탐색 제약(종단경사· + # 최소곡선반지름·회피지역)에 걸려 체인이 통째로 멈춘다(2026-09-06 실측: + # 「세그먼트 1 (BP → CP1) 경로 탐색 실패」로 종횡단만 옛 노선에 남았음). + # PLAN 0-3 에서 자동탐색은 접기로 했고 노선은 사용자가 직접 고친다. + algorithm="as_planned", + bp=points["bp"], + ep=points["ep"], + cp=points.get("cp") or [], + ap=points.get("ap") or [], + fp=points.get("fp") or [], + station_interval_m=params.get("station_interval_m"), + cross_half_width_m=params.get("cross_half_width_m"), + cross_sample_interval_m=params.get("cross_sample_interval_m"), + long_sample_interval_m=params.get("long_sample_interval_m"), + **{key: options.get(key) for key in ("grade_class",) if options.get(key)}, + paved=bool(options.get("paved", False)), + terrain_type=str(options.get("terrain_type") or "special"), + main_direction=str(options.get("main_direction") or "auto"), + min_curve_radius_m=options.get("min_curve_radius_m"), + max_uphill_grade=options.get("max_uphill_grade"), + max_downhill_grade=options.get("max_downhill_grade"), + min_uphill_grade=options.get("min_uphill_grade"), + min_downhill_grade=options.get("min_downhill_grade"), + weights=options.get("weights"), + allow_avoid_pass_through=bool(options.get("allow_avoid_pass_through", False)), + max_grade_pct=params.get("max_grade_pct"), + min_vertical_radius_m=params.get("min_vertical_radius_m"), + min_tangent_length_m=params.get("min_tangent_length_m"), + balance_segment_length_m=params.get("balance_segment_length_m"), + start_elevation_offset_m=params.get("start_elevation_offset_m"), + end_elevation_offset_m=params.get("end_elevation_offset_m"), + ) + + async def run_redesign_chain( project_id: UUID, surface_model_id: int, @@ -421,7 +468,6 @@ async def run_redesign_chain( confirm_latest_route, solve_route_record, ) - from B05_Profile.B05_Profile_Schema import RouteSolveRequest from B06_Section.B06_Section_Repository import ( get_cross_section_designs, get_latest_section_options, @@ -478,7 +524,6 @@ async def run_redesign_chain( ) params = (stage2 or {}).get("params") or {} points = params.get("points") or {} - options = params.get("options") or {} if not points.get("bp") or not points.get("ep"): logger.warning( "재확정 체인 중단(stage 2 params에 제어점 없음): project_id=%s", project_id @@ -488,43 +533,7 @@ async def run_redesign_chain( marks.append(("준비·사용자입력 회수", time.perf_counter())) # 2) B05 재계산 — 지표면 관련 값만 새 확정 선택으로 교체, 나머지는 사용자 저장분. - request = RouteSolveRequest( - filter_key=str(selection.get("source_filter") or params.get("filter_key")), - method=str(selection.get("method") or params.get("method") or "dtm"), - smooth=bool(selection.get("smooth", params.get("smooth", False))), - surface_model_id=surface_model_id, - # 사용자가 고친 계획노선을 **그대로** 쓴다 — 다시 풀면 탐색 제약(종단경사· - # 최소곡선반지름·회피지역)에 걸려 체인이 통째로 멈춘다(2026-09-06 실측: - # 「세그먼트 1 (BP → CP1) 경로 탐색 실패」로 종횡단만 옛 노선에 남았음). - # PLAN 0-3 에서 자동탐색은 접기로 했고 노선은 사용자가 직접 고친다. - algorithm="as_planned", - bp=points["bp"], - ep=points["ep"], - cp=points.get("cp") or [], - ap=points.get("ap") or [], - fp=points.get("fp") or [], - station_interval_m=params.get("station_interval_m"), - cross_half_width_m=params.get("cross_half_width_m"), - cross_sample_interval_m=params.get("cross_sample_interval_m"), - long_sample_interval_m=params.get("long_sample_interval_m"), - **{key: options.get(key) for key in ("grade_class",) if options.get(key)}, - paved=bool(options.get("paved", False)), - terrain_type=str(options.get("terrain_type") or "special"), - main_direction=str(options.get("main_direction") or "auto"), - min_curve_radius_m=options.get("min_curve_radius_m"), - max_uphill_grade=options.get("max_uphill_grade"), - max_downhill_grade=options.get("max_downhill_grade"), - min_uphill_grade=options.get("min_uphill_grade"), - min_downhill_grade=options.get("min_downhill_grade"), - weights=options.get("weights"), - allow_avoid_pass_through=bool(options.get("allow_avoid_pass_through", False)), - max_grade_pct=params.get("max_grade_pct"), - min_vertical_radius_m=params.get("min_vertical_radius_m"), - min_tangent_length_m=params.get("min_tangent_length_m"), - balance_segment_length_m=params.get("balance_segment_length_m"), - start_elevation_offset_m=params.get("start_elevation_offset_m"), - end_elevation_offset_m=params.get("end_elevation_offset_m"), - ) + request = redesign_solve_request(params, selection, surface_model_id) # 종횡단은 관을 정한 뒤 한 번만 만든다(342-2) — 여기서 만들면 아래 4c 가 통째로 덮는 헛일. solve_result: Any = await solve_route_record(project_id, request, generate_sections=False) if isinstance(solve_result, JSONResponse): diff --git a/B05_Profile/B05_Profile_Route_Save.py b/B05_Profile/B05_Profile_Route_Save.py new file mode 100644 index 000000000..d1e344d79 --- /dev/null +++ b/B05_Profile/B05_Profile_Route_Save.py @@ -0,0 +1,496 @@ +"""노선 저장 서버 길(346-3) — 노선 초안을 정본으로: 노선 → 풀기 → 관 · 구조물 따라감 → +종 · 횡단 한 번 → 계획고 옮김 → 횡단 이월 → B06 확정(서버 재계산 · 3D). 배수유역은 체인에서 뗌. + +API — `POST /api/projects/{id}/route/replan`(`B05_Profile_Router_Replan.replan_route`): + 요청 `RouteSaveRequest` = 옛 칸(`vertices` · `initial` · `balanced_profile` · `terrain` · + `station_snap`) + + `draft` — 계약 `RouteDraft`(`A00_Common/design/design_pipeline_types.ts`) 그대로 + `{rev, nodes: [[x, y]…], curveOn: [bool…], curveRadius: [m | null…], curveLock, curveArc, + stationSnap, terrain}` · 오면 `vertices` · `station_snap` · `terrain` 대신 이것. + `as_initial` — 추천 노선 [적용](초기값 다시 뜸) · 손으로 고친 노선 저장은 false(초기값 그대로). + 응답 200 `{status: "success", project_id, route_id, total_length_m, vertex_count, current_initial, + initial_refreshed, pipes: {kept, crossing, concave}, structures: {moved, far}, cross_dropped, + drainage_stale: true, seconds}`. + 오류 400(노드 둘 미만 · 초안 칸 길이 어긋남) · 404(프로젝트 없음) · + 409 `{status: "error", message}`(체인 실패 — 노선 · 관 · 구조물 · 종횡단 파일 · 새 노선 행을 + 옛것으로 되살린 뒤). + +규칙: + - 관 정본을 지우지 않음 — 좌표를 따라감 · 새 세류선 교차점 기본 관 · 오목 지점 보충만 더함 + (`B05_Profile_Route_Save_Parts`) · 배수유역 저장분은 노선 지문이 달라 「낡음」(B04 `_resolve`). + - 계획고 = 바뀐 구간만 새 지반 첫 계획선 · 그 밖 사용자 계획고 지킴 + (`B05_Profile_Engine_Grade_Carry`) · 추천 「종단 반영」 켬이면 균형 종단 + (`apply_profile_target`). + - 초기값은 여기서 안 뜸 — 부르는 쪽이 `as_initial` 일 때만(자동설계 체인 직후 + 추천 [적용]). +""" + +from __future__ import annotations + +import asyncio +import json +import logging +import shutil +import tempfile +import time +from dataclasses import asdict +from pathlib import Path +from typing import Any, Literal +from uuid import UUID + +import numpy as np +from fastapi.responses import JSONResponse +from pydantic import BaseModel, Field + +from B05_Profile.B05_Profile_Route_Save_Parts import ( + add_crossing_pipes, + carry_cross_inputs, + concave_chainages, + fill_concave_pipes, + follow_pipes, + station_keys, + vertices_of, +) +from B05_Profile.B05_Profile_Schema import RouteReplanRequest, RouteVertexInput +from config.config_db import get_db_pool + +logger = logging.getLogger(__name__) + + +class RouteDraftInput(BaseModel): + """노선 초안 — 계약 `RouteDraft` 모양 그대로(칸 이름 camelCase).""" + + rev: int = 0 + nodes: list[tuple[float, float]] = Field(default_factory=list) + curveOn: list[bool] = Field(default_factory=list) # noqa: N815 — 계약 칸 이름 + curveRadius: list[float | None] = Field(default_factory=list) # noqa: N815 + curveLock: list[str | None] = Field(default_factory=list) # noqa: N815 + curveArc: list[float | None] = Field(default_factory=list) # noqa: N815 + stationSnap: bool = False # noqa: N815 + terrain: Literal["las", "sheet"] | None = None + + +class RouteSaveRequest(RouteReplanRequest): + """노선 저장 요청 — 옛 [노선 반영] 칸 + 초안(`draft`) · 추천 [적용](`as_initial`).""" + + draft: RouteDraftInput | None = None + as_initial: bool = False + + def resolved(self) -> RouteReplanRequest: + """초안이 오면 노드 · 곡선 · 스냅 · 지형을 옛 칸으로 편다. 칸 길이가 어긋나면 ValueError.""" + if self.draft is None: + return self + draft = self.draft + count = len(draft.nodes) + for name in ("curveOn", "curveRadius"): + values = getattr(draft, name) + if values and len(values) != count: + raise ValueError( + f"노선 초안 {name} 길이({len(values)})가 노드 수({count})와 다릅니다." + ) + vertices = [ + RouteVertexInput( + x=x, + y=y, + curve=draft.curveOn[index] if draft.curveOn else True, + radius_m=draft.curveRadius[index] if draft.curveRadius else None, + ) + for index, (x, y) in enumerate(draft.nodes) + ] + return self.model_copy( + update={ + "vertices": vertices, + "station_snap": draft.stationSnap, + "terrain": draft.terrain, + } + ) + + +class _Backup: + """체인이 고치는 파일을 손에 쥠 — 실패하면 그대로 되돌림(노선 · 관 · 구조물 · 종횡단).""" + + def __init__(self, project_root: Path, paths: list[Path], folders: list[Path]) -> None: + self.held = tempfile.mkdtemp(prefix="route_save_", dir=str(project_root)) + self.files = {path: path.read_bytes() if path.is_file() else None for path in paths} + self.folders: dict[Path, Path | None] = {} + for index, folder in enumerate(folders): + copy = Path(self.held) / f"dir{index}" + self.folders[folder] = shutil.copytree(folder, copy) if folder.is_dir() else None + + def restore(self) -> None: + for path, saved in self.files.items(): + if saved is None: + path.unlink(missing_ok=True) + else: + path.parent.mkdir(parents=True, exist_ok=True) + path.write_bytes(saved) + for folder, copy in self.folders.items(): + shutil.rmtree(folder, ignore_errors=True) + if copy is not None: + shutil.copytree(copy, folder) + + def drop(self) -> None: + shutil.rmtree(self.held, ignore_errors=True) + + +class _Fail(RuntimeError): + """체인 한 단계 실패 — 사유를 그대로 화면까지(까닭을 덮지 않음).""" + + +def _reason(response: JSONResponse) -> str: + try: + return bytes(response.body).decode("utf-8", "replace")[:300] + except Exception: # noqa: BLE001 — 사유는 로그 · 응답용 + return f"status={response.status_code}" + + +def _read_json(path: Path) -> dict[str, Any]: + try: + return json.loads(path.read_text(encoding="utf-8")) + except (OSError, ValueError): + return {} + + +def _xy(project_root: Path, route_data_path: str | None) -> list[list[float]]: + from B05_Profile.B05_Profile_Engine_Sections import _load_route_polyline + + if not route_data_path: + return [] + try: + return [[x, y] for x, y, _ in _load_route_polyline(project_root, route_data_path)] + except (OSError, ValueError): + return [] + + +def _ground_along(project_root: Path, selection: dict[str, Any], vertices: list) -> tuple: + """새 노선 지반 종단(2 m) — 오목 지점 찾기. 지표면을 못 열면 빈 배열(보충 건너뜀).""" + from shapely.geometry import LineString + + from B05_Profile.B05_Profile_Route_Save_Parts import CONCAVE_STEP_M + from common_util.common_util_surface_sampler import build_surface_sampler + + try: + sampler = build_surface_sampler( + project_root / "B04_PreProcess" / "models", + str(selection.get("source_filter")), + str(selection.get("method") or "dtm"), + bool(selection.get("smooth")), + ) + except (FileNotFoundError, KeyError, OSError, ValueError): + logger.warning("노선 저장: 지표면을 못 열어 오목 지점 보충을 건너뜀") + return np.zeros(0), np.zeros(0) + line = LineString([(vertex.x, vertex.y) for vertex in vertices]) + chainage = np.arange(0.0, line.length + 1e-9, CONCAVE_STEP_M) + xy = np.array([line.interpolate(float(s)).coords[0] for s in chainage], dtype=np.float64) + z, valid = sampler.sample_xy(xy) + return chainage[valid], np.asarray(z, dtype=np.float64)[valid] + + +async def _place_pipes( + project_id: UUID, + project_root: Path, + stored_path: str, + selection: dict[str, Any], + old_pipes: list, + old_xy: list[list[float]], + points: list[dict[str, float]], +) -> tuple[list, dict[str, int]]: + """관 따라감 → 새 교차점 기본 관 → 오목 지점 보충 → 정본 저장(새 노선 지문).""" + 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 route_signature, save_pipe_points + + prepared = await _prepare(project_id) + if isinstance(prepared, JSONResponse): # 도엽 · 노선 파일 없음 — 따라감만 + vertices, streams = vertices_of([(p["x"], p["y"]) for p in points]), [] + else: + vertices, streams = prepared["vertices"], prepared["streams"] + pairs = follow_pipes(old_pipes, vertices) + total = float(vertices[-1].chainage_m) + crossing = add_crossing_pipes(pairs, _base_pipes(vertices, streams), old_xy, total) + chainage, ground = await asyncio.to_thread(_ground_along, project_root, selection, vertices) + concave = fill_concave_pipes(pairs, concave_chainages(chainage, ground), vertices, old_xy) + await asyncio.to_thread( + save_pipe_points, + stored_path, + route_signature(vertices), + [pipe for _, pipe in pairs], + vertices, + ) + return pairs, {"kept": len(old_pipes), "crossing": crossing, "concave": concave} + + +async def _carry_profile( + project_root: Path, + relative: str, + route_id: int, + old_alignment: dict[str, Any], + old_xy: list[list[float]], + new_xy: list[list[float]], +) -> None: + """바뀐 구간만 새 지반 첫 계획선 · 그 밖 사용자 계획고 · 종단곡선 R 은 길이 차만큼 밀어 지킴.""" + from B05_Profile.B05_Profile_Engine_Grade_Carry import carry_profile_edits, changed_span + from B05_Profile.B05_Profile_Repository import update_longitudinal_grade_summary + from B05_Profile.B05_Profile_Router import _apply_alignment_edits + from common_util.common_util_json import atomic_write_json + + if not old_alignment.get("base_pvi"): + return # 옛 선형 없음 — 새 지반 첫 계획선 그대로 + path = project_root / relative + longitudinal = _read_json(path) + alignment = longitudinal.get("profile_alignment") or {} + span = changed_span([tuple(p) for p in old_xy], [tuple(p) for p in new_xy]) + if span is None: # 같은 노선 — 옛 계획선 통째 + base, edits = old_alignment["base_pvi"], old_alignment.get("edits") or {} + else: + base, edits = carry_profile_edits( + old_alignment["base_pvi"], old_alignment.get("edits"), span, alignment["base_pvi"] + ) + alignment["base_pvi"] = base + await asyncio.to_thread(atomic_write_json, path, longitudinal) + _alignment, summary, _dropped = await asyncio.to_thread( + _apply_alignment_edits, project_root, relative, edits + ) + pool = get_db_pool() + async with pool.acquire() as connection: + await connection.begin() + await update_longitudinal_grade_summary( + connection, route_id=route_id, grade_summary=summary + ) + await connection.commit() + + +async def _carry_cross( + project_id: UUID, + project_root: Path, + route_id: int, + relative: str, + old_long: dict[str, Any], + old_designs: list[dict[str, Any]], + old_pipes: list, + old_structures: list, + pairs: list, + interval_m: float, + standard: dict[str, Any] | None, +) -> int: + """측점 열쇠로 옛 횡단 입력을 새 측점에 얹고 전 측점을 새 계획고로 셈 · 버린 수 돌려줌.""" + from B05_Profile.B05_Profile_Structures_Repository import load_structures + from B06_Section.B06_Section_Repository_Bulk import merge_cross_section_designs + from B06_Section.B06_Section_Router_Confirm import _rowless_station_chainages + from B06_Section.B06_Section_Router_Design import ( + CARRY_INPUT_KEYS, + STRUCTURE_ADJUST_KEYS, + compute_default_designs, + ) + + new_long = _read_json(project_root / relative) + _, new_structures = load_structures(str(project_root)) + old_keys = station_keys( + old_long.get("stations") or [], list(enumerate(old_pipes)), old_structures, interval_m + ) + new_keys = station_keys(new_long.get("stations") or [], pairs, new_structures, interval_m) + designs = { + round(float(row["chainage_m"]), 3): row["design"] + for row in old_designs + if isinstance(row.get("design"), dict) + } + inputs, adjusts, dropped = carry_cross_inputs( + old_keys, new_keys, designs, CARRY_INPUT_KEYS, STRUCTURE_ADJUST_KEYS + ) + chainages = await asyncio.to_thread(_rowless_station_chainages, project_root, relative, []) + entries = await asyncio.to_thread( + compute_default_designs, project_root, relative, chainages, standard, inputs + ) + entries = [(at, {**design, **adjusts.get(round(at, 3), {})}) for at, design in entries] + pool = get_db_pool() + async with pool.acquire() as connection: + await connection.begin() + try: + await merge_cross_section_designs( + connection, route_id=route_id, entries=entries, replace=True, project_id=project_id + ) + await connection.commit() + except Exception: + await connection.rollback() + raise + return dropped + + +async def _rollback(project_id: UUID, old_route_id: int, params: dict[str, Any]) -> None: + """새 노선 행(종횡단 함께 지워짐)을 지우고 stage 2 params 를 옛것으로 — 최신 노선 = 옛 노선.""" + from common_util.common_util_workflow_state import start_stage + + pool = get_db_pool() + async with pool.acquire() as connection: + await connection.begin() + try: + async with connection.cursor() as cursor: + await cursor.execute( + "DELETE FROM routes WHERE project_id = %s AND id > %s", + (str(project_id), old_route_id), + ) + await start_stage(cursor, str(project_id), 2, params) + await connection.commit() + except Exception: + await connection.rollback() + logger.exception("노선 저장 되살림(DB) 실패: project_id=%s", project_id) + + +async def run_route_save_chain( + project_id: UUID, + project_root: Path, + stored_path: str, + surface_model_id: int, + selection: dict[str, Any], + old_route: dict[str, Any], + points: list[dict[str, float]], + previous_params: dict[str, Any], + profile_target: tuple[list[float], list[float]] | None = None, +) -> dict[str, Any]: + """노선 저장 체인 — 성공이면 요약 dict · 실패면 `{"error": 사유}`(파일 · 행은 되살린 뒤).""" + import aiomysql + + 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_Router import ( + _section_options, + 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 ( + get_cross_section_designs, + get_latest_section_options, + get_longitudinal_section, + ) + from B06_Section.B06_Section_Router import regenerate_route_sections + from B06_Section.B06_Section_Router_Confirm import confirm_sections + from B06_Section.B06_Section_Router_Design import stored_standard_cross_section + from B06_Section.B06_Section_Schema import SectionConfirmRequest + from common_util.common_util_drainage_pipes import pipe_points_path, read_pipe_points_file + from common_util.common_util_initial_snapshot import clear_designing, mark_designing + from common_util.common_util_workflow_state import get_workflow_state + + marks = [("시작", time.perf_counter())] + old_route_id = int(old_route["id"]) + pool = get_db_pool() + async with pool.acquire() as connection: + async with connection.cursor(aiomysql.DictCursor) as cursor: + state = await get_workflow_state(cursor, str(project_id)) + old_long_row = await get_longitudinal_section(connection, project_id, old_route_id) + old_designs = await get_cross_section_designs(connection, old_route_id) + old_section_options = await get_latest_section_options(connection, project_id) + stage2 = next((s for s in state.get("stages", []) if int(s.get("stage_no", -1)) == 2), None) + params = (stage2 or {}).get("params") or {} + relative_old = str((old_long_row or {}).get("longitudinal_file_path") or "") + old_long = _read_json(project_root / relative_old) if relative_old else {} + old_xy = _xy(project_root, old_route.get("route_data_path")) + old_pipes = read_pipe_points_file(pipe_points_path(stored_path)) + _, old_structures = load_structures(str(project_root)) + stage_dir = project_root / _STAGE_SUBDIR + backup = _Backup( + project_root, + [ + project_root / str(old_route.get("route_data_path") or "B05_Profile/route/x"), + pipe_points_path(stored_path), + Path(structures_file_path(str(project_root))), + ], + [stage_dir / "longitudinal", stage_dir / "cross_sections"], + ) + mark_designing(project_root) + try: + # 구조물 = 땅 위 자리 따라감(옛 노선 파일이 아직 옛 것일 때 — 좌표 없는 옛 자료 채움) + try: + structures = await asyncio.to_thread( + follow_structures_to_route, project_root, old_route.get("route_data_path"), points + ) + except Exception: # noqa: BLE001 — 따라감 실패 = 옛 누가거리 그대로(비치명) + logger.exception("구조물 따라감 실패(옛 누가거리 유지): project_id=%s", project_id) + structures = {"moved": 0, "far": 0} + request = redesign_solve_request(params, selection, surface_model_id) + solved: Any = await solve_route_record(project_id, request, generate_sections=False) + if isinstance(solved, JSONResponse): + raise _Fail(f"노선 풀기 실패({solved.status_code}): {_reason(solved)}") + new_route_id = int(solved.route_id) + confirmed = await confirm_latest_route(project_id, None, mark_stage_complete=False) + if isinstance(confirmed, JSONResponse): + raise _Fail(f"노선 확정 실패({confirmed.status_code})") + new_xy = _xy(project_root, solved.route_data_path) + marks.append(("노선 풀기 · 구조물 따라감", time.perf_counter())) + + pairs, pipes = await _place_pipes( + project_id, project_root, stored_path, selection, old_pipes, old_xy, points + ) + marks.append(("관 따라감 · 교차점 · 오목 보충", time.perf_counter())) + + seed = asdict(_section_options(request, old_section_options)) + seed.pop("extra_stations", None) + made = await regenerate_route_sections( + project_id, new_route_id, float(seed["cross_half_width_m"]), stored_options=seed + ) + if isinstance(made, JSONResponse): + raise _Fail(f"종횡단 생성 실패({made.status_code}): {_reason(made)}") + async with pool.acquire() as connection: + new_long_row = await get_longitudinal_section(connection, project_id, new_route_id) + relative = str((new_long_row or {}).get("longitudinal_file_path") or "") + marks.append(("종횡단 한 번", time.perf_counter())) + + if profile_target is not None: + from B05_Profile.B05_Profile_Router_RouteInitial import apply_profile_target + + await apply_profile_target(project_id, new_route_id, profile_target) + else: + await _carry_profile( + project_root, + relative, + new_route_id, + old_long.get("profile_alignment") or {}, + old_xy, + new_xy, + ) + marks.append(("계획고 옮김", time.perf_counter())) + + standard = stored_standard_cross_section(old_long_row) + dropped = await _carry_cross( + project_id, + project_root, + new_route_id, + relative, + old_long, + old_designs, + old_pipes, + old_structures, + pairs, + float(seed["station_interval_m"]), + standard, + ) + marks.append(("횡단 이월(측점 열쇠)", time.perf_counter())) + + done = await confirm_sections( + project_id, + new_route_id, + SectionConfirmRequest(standard_cross_section=standard) if standard else None, + mark_stage_complete=False, + ) + if isinstance(done, JSONResponse): + raise _Fail(f"종횡단 확정 실패({done.status_code})") + marks.append(("B06 확정 · 서버 재계산 · 3D", time.perf_counter())) + except Exception as exc: # noqa: BLE001 — 어느 단계든 옛 노선으로 되살림 + if not isinstance(exc, _Fail): + logger.exception("노선 저장 체인 실패: project_id=%s", project_id) + await asyncio.to_thread(backup.restore) + await _rollback(project_id, old_route_id, previous_params) + return {"error": str(exc) if isinstance(exc, _Fail) else f"노선 저장 중 오류: {exc}"} + finally: + backup.drop() + clear_designing(project_root) + _log_steps("재확정 체인(노선 저장)", marks) + return { + "route_id": new_route_id, + "total_length_m": float(solved.total_length_m or 0.0), + "pipes": pipes, + "structures": structures, + "cross_dropped": dropped, + "drainage_stale": True, + "seconds": round(marks[-1][1] - marks[0][1], 2), + } diff --git a/B05_Profile/B05_Profile_Route_Save_Parts.py b/B05_Profile/B05_Profile_Route_Save_Parts.py new file mode 100644 index 000000000..f3ec307f5 --- /dev/null +++ b/B05_Profile/B05_Profile_Route_Save_Parts.py @@ -0,0 +1,224 @@ +"""노선 저장 길(346-3) 순수 부품 — 관 따라감 · 새 교차점 기본 관 · 오목 지점 보충 · 측점 열쇠 · +횡단 이월. + +DB · 파일 없이 값만 받아 값을 돌려준다(시험이 그대로 부름). 차례 · 저장 · 되살림은 +`B05_Profile_Route_Save.py`. + + - 관 = 좌표가 정본 — 새 노선에 투영해 누가거리만 다시(`project_pipe_points` · 정수 m). 지우지 않음 + (사용자 관 편집 지킴). + - 새 세류선 교차점(옛 노선 위에 없던 자리)에 관이 없으면 기본 관 하나(`_base_pipes` 결과를 받음). + - 관 최대 간격 보충 = 옛 노선에서 벗어난 자리의 **오목 지점**(지반 종단 낮은 곳)에만 — 격자 없음. + - 측점 열쇠(계약 `design_pipeline_types.ts` 머리) = 일반 `grid:k`(누가거리 ÷ 측점 간격 반올림) · + 관 `pipe:` · 구조물 `structure:`(구간 시작 · 끝은 `:start` · `:end`) · 시종점 + `end:BP` · `end:EP`. 관 id = 저장 파일 차례(노선 저장 한 번 안에서 옛 · 새가 같음). + - 횡단 이월 = 같은 열쇠 측점에 `CARRY_INPUT_KEYS` 일곱 · 관 · 구조물 열쇠면 + `STRUCTURE_ADJUST_KEYS` 도(구조물 조정은 구조물 따라) · 열쇠가 사라진 측점 수 = + 「노선 밖 고침」. +""" + +from __future__ import annotations + +import copy +import math +from typing import Any, Sequence + +import numpy as np +from shapely.geometry import LineString, Point + +from common_util.common_util_drainage_pipes import ( + PIPE_SOURCE_SPACING, + PIPE_SOURCE_STREAM, + PipePoint, + project_pipe_points, +) +from common_util.common_util_route_geometry import RouteVertex +from common_util.common_util_structure_station import plan_stations, snap_structure_station +from config.config_system import DRAINAGE_PIPE_MAX_SPACING_M, DRAINAGE_PIPE_MIN_SPACING_M + +# 옛 노선에서 이만큼(m) 넘게 떨어진 자리 = 노선이 바뀐 자리(교차점 · 보충 대상). +MOVED_AWAY_M = 0.5 +# 오목 지점을 찾을 지반 표본 간격(m) · 고름 창(표본 수) — DTM 잔물결을 오목으로 안 읽게. +# ponytail: 이동 평균 + 국소 최저 · 계곡이 얕아 놓치면 창을 줄이거나 흐름 강도와 섞을 것. +CONCAVE_STEP_M = 2.0 +CONCAVE_WINDOW = 5 +_SAME_M = 0.01 + + +def vertices_of(xy: Sequence[Sequence[float]]) -> list[RouteVertex]: + """평면 점열 → 누가거리 붙은 정점(z 0).""" + out: list[RouteVertex] = [] + walked = 0.0 + for index, point in enumerate(xy): + x, y = float(point[0]), float(point[1]) + if index: + walked += math.hypot(x - out[-1].x, y - out[-1].y) + out.append(RouteVertex(x=x, y=y, z=0.0, chainage_m=walked)) + return out + + +def follow_pipes( + old_pipes: list[PipePoint], vertices: list[RouteVertex] +) -> list[tuple[int, PipePoint]]: + """관이 새 노선 좌표를 따라감 → [(관 id = 옛 차례, 새 관)] 누가거리 순 · 옛 목록 그대로.""" + pairs = [(index, copy.deepcopy(pipe)) for index, pipe in enumerate(old_pipes)] + for _, pipe in pairs: + project_pipe_points([pipe], vertices) + pairs.sort(key=lambda pair: pair[1].chainage_m) + return pairs + + +def _moved_away(old_line: LineString | None, x: float, y: float) -> bool: + return old_line is None or old_line.distance(Point(x, y)) > MOVED_AWAY_M + + +def add_crossing_pipes( + pairs: list[tuple[int, PipePoint]], + crossings: Sequence[Any], + old_xy: Sequence[Sequence[float]] | None, + total_m: float, +) -> int: + """옛 노선에 없던 세류선 교차점에 관이 없으면 기본 관을 더함 · 더한 수.""" + old_line = LineString(old_xy) if old_xy and len(old_xy) >= 2 else None + added = 0 + for crossing in crossings: + if not _moved_away(old_line, crossing.x, crossing.y): + continue # 옛 교차점 — 사용자가 지운 관을 되살리지 않음 + at = snap_structure_station(float(crossing.chainage_m), [total_m]) + if any(abs(pipe.chainage_m - at) < DRAINAGE_PIPE_MIN_SPACING_M for _, pipe in pairs): + continue + pairs.append( + (-1, PipePoint(chainage_m=at, source=PIPE_SOURCE_STREAM, x=crossing.x, y=crossing.y)) + ) + added += 1 + pairs.sort(key=lambda pair: pair[1].chainage_m) + return added + + +def concave_chainages(chainage: np.ndarray, ground: np.ndarray) -> list[tuple[float, float]]: + """지반 종단의 오목 지점(이동 평균 국소 최저) → [(누가거리, 고른 지반고)].""" + if ground.size < 3: + return [] + kernel = np.ones(CONCAVE_WINDOW) / CONCAVE_WINDOW + pad = CONCAVE_WINDOW // 2 + smooth = np.convolve(np.pad(ground, pad, mode="edge"), kernel, mode="valid") + return [ + (float(chainage[i]), float(smooth[i])) + for i in range(1, smooth.size - 1) + if smooth[i] < smooth[i - 1] and smooth[i] <= smooth[i + 1] + ] + + +def fill_concave_pipes( + pairs: list[tuple[int, PipePoint]], + concave: list[tuple[float, float]], + vertices: list[RouteVertex], + old_xy: Sequence[Sequence[float]] | None, +) -> int: + """관 사이(시종점 포함)가 최대 간격을 넘으면 옛 노선 밖 오목 지점 중 가장 낮은 곳에 관.""" + line = LineString([(vertex.x, vertex.y) for vertex in vertices]) + old_line = LineString(old_xy) if old_xy and len(old_xy) >= 2 else None + total = float(line.length) + added = 0 + while True: + edges = [0.0, *(pipe.chainage_m for _, pipe in pairs), total] + best: tuple[float, float] | None = None + for start, end in zip(edges, edges[1:]): + if end - start <= DRAINAGE_PIPE_MAX_SPACING_M: + continue + for at, z in concave: + if not ( + start + DRAINAGE_PIPE_MIN_SPACING_M < at < end - DRAINAGE_PIPE_MIN_SPACING_M + ): + continue + point = line.interpolate(at) + if _moved_away(old_line, point.x, point.y) and (best is None or z < best[1]): + best = (at, z) + if best is None: + return added + at = snap_structure_station(best[0], [total]) + point = line.interpolate(at) + pairs.append( + (-1, PipePoint(chainage_m=at, source=PIPE_SOURCE_SPACING, x=point.x, y=point.y)) + ) + pairs.sort(key=lambda pair: pair[1].chainage_m) + concave = [item for item in concave if abs(item[0] - best[0]) > _SAME_M] + added += 1 + + +def _facility_keys( + pipes: Sequence[tuple[int, PipePoint]], structures: Sequence[Any] +) -> dict[float, str]: + """측점 자리(0.01 m) → 관 · 구조물 열쇠. 같은 자리면 관이 먼저(`plan_stations` 이름 규칙).""" + keys: dict[float, str] = {} + + def put(marks: list[dict[str, object]], head: str) -> None: + for mark in marks: + role = str(mark.get("role") or "anchor") + keys.setdefault( + round(float(mark["chainage_m"]), 2), head if role == "anchor" else f"{head}:{role}" + ) + + for pipe_id, pipe in pipes: + if pipe_id >= 0: + put(plan_stations([pipe], []), f"pipe:{pipe_id}") + for item in structures: + if getattr(item, "structure_id", None): + put(plan_stations([], [item]), f"structure:{item.structure_id}") + return keys + + +def station_keys( + stations: Sequence[dict[str, Any]], + pipes: Sequence[tuple[int, PipePoint]], + structures: Sequence[Any], + interval_m: float, +) -> dict[str, float]: + """측점 열쇠 → 누가거리. 관 · 구조물 자리 > 시종점 > 일반 격자(간격 배수인 것만).""" + facility = _facility_keys(pipes, structures) + values = [float(row["chainage_m"]) for row in stations if row.get("chainage_m") is not None] + out: dict[str, float] = {} + for index, chainage in enumerate(values): + key = facility.get(round(chainage, 2)) + if key is None and index == 0: + key = "end:BP" + elif key is None and index == len(values) - 1: + key = "end:EP" + elif key is None and interval_m > 0: + k = round(chainage / interval_m) + key = f"grid:{k}" if abs(chainage - k * interval_m) < _SAME_M else None + if key is not None: + out.setdefault(key, chainage) + return out + + +def carry_cross_inputs( + old_keys: dict[str, float], + new_keys: dict[str, float], + old_designs: dict[float, dict[str, Any]], + input_keys: Sequence[str], + adjust_keys: Sequence[str], +) -> tuple[dict[float, dict[str, Any]], dict[float, dict[str, Any]], int]: + """같은 열쇠 측점으로 횡단 입력을 넘김 → (입력{새 누가거리 0.001: 값}, 구조물 조정{…}, 버린 수). + + `old_designs` = 옛 노선 행 {누가거리 0.001: design}. 버린 수 = 설계가 있던 옛 측점 중 새 노선에 + 같은 열쇠가 없는 것(「노선 밖 고침」). + """ + inputs: dict[float, dict[str, Any]] = {} + adjusts: dict[float, dict[str, Any]] = {} + dropped = 0 + for key, old_at in old_keys.items(): + design = old_designs.get(round(old_at, 3)) + if not isinstance(design, dict): + continue + new_at = new_keys.get(key) + if new_at is None: + dropped += 1 + continue + carried = {name: design[name] for name in input_keys if design.get(name) is not None} + if carried: + inputs[round(new_at, 3)] = carried + if key.startswith(("pipe:", "structure:")): + adjust = {name: design[name] for name in adjust_keys if design.get(name) is not None} + if adjust: + adjusts[round(new_at, 3)] = adjust + return inputs, adjusts, dropped diff --git a/B05_Profile/B05_Profile_Router_Replan.py b/B05_Profile/B05_Profile_Router_Replan.py index 2dc46d594..428c49c03 100644 --- a/B05_Profile/B05_Profile_Router_Replan.py +++ b/B05_Profile/B05_Profile_Router_Replan.py @@ -14,17 +14,13 @@ 노선 초기화는 수정본 파일을 지우는 것이다 — 그러면 초기 폴리라인을 읽으므로 「초기본을 수정본으로 복사」와 결과가 같다. -[확인]을 눌렀을 때만 계산이 돈다. 재계산은 **초기 업로드 체인의 로직을 그대로 재사용**한다 -(2026-09-06 사용자 제안) — `run_redesign_chain`이 배수유역 다시 분석 → 기본 관 저장 → 관 -정착 계획선 재산출 → 종횡단 재생성 → 옛 측점 설계를 누가거리로 이월 → B05·B06 확정까지 한 -줄로 돈다. 여기서 하는 일은 그 앞에 **노선을 갈아 끼우는 것**뿐이다. - -시설 처리(2026-09-06 사용자 확정) — 횡단배수 지점은 노선 자리에 따라 생기고 없어지므로 -저장분을 버리고 **새로 계산한 값**을 쓴다. 그 밖의 구조물은 **옛 측점값을 그대로** 이어받는다 -(`structures.json`은 프로젝트 정본이라 손대지 않으면 그대로 남는다). +노선 저장(346-3) — 여기서 노선을 갈아 끼운 뒤 `B05_Profile_Route_Save.run_route_save_chain` 이 +풀기 → 관 · 구조물 좌표 따라감 → 종횡단 한 번 → 계획고 옮김 → 횡단 이월(측점 열쇠) → B06 확정까지 +돈다. 배수유역은 다시 안 셈(낡음) · 관 정본은 안 지움 · 초기값은 추천 [적용](`as_initial`)일 때만. +요청 · 응답 · 오류 모양 = 그 모듈 머리. GET /api/projects/{id}/route/plan → 예상노선·계획노선 정점(사업지 좌표계) - POST /api/projects/{id}/route/replan → 고친 계획노선으로 갈아 끼우고 재계산 + POST /api/projects/{id}/route/replan → 노선 초안 저장(노선 저장 체인) POST /api/projects/{id}/route/replan/reset → 계획노선을 예상노선으로 되돌리고 재계산 POST /api/projects/{id}/route/initial → 초기 계획노선 갈래 하나를 만들어 돌려줌(추천 저장본만 씀) """ @@ -45,6 +41,7 @@ from fastapi.responses import JSONResponse from B03_FileInput.B03_FileInput_Repository import get_project_storage_relative_path from B05_Profile.B05_Profile_Engine_RouteInitial import read_initial_choice, write_initial_choice from B05_Profile.B05_Profile_Engine_Sections import load_station_snap +from B05_Profile.B05_Profile_Route_Save import RouteSaveRequest from B05_Profile.B05_Profile_Schema import ( # noqa: F401 — 부르는 곳이 `replan.` 으로 씀 RouteInitialCriteria, RouteInitialRequest, @@ -340,22 +337,19 @@ async def _recompute( from B03_FileInput.B03_FileInput_Service_Chain import ( _log_steps, _planned_route_points_in_project_crs, - run_redesign_chain, ) from B04_PreProcess.B04_PreProcess_Service import ( find_surface_model_for_selection, switch_design_terrain, ) from B05_Profile.B05_Profile_Repository import get_latest_route - from common_util.common_util_drainage_pipes import clear_pipe_points + from B05_Profile.B05_Profile_Route_Save import run_route_save_chain from common_util.common_util_surface_confirmation import get_surface_confirmation_params from common_util.common_util_workflow_state import get_workflow_state, start_stage marks = [("시작", time.perf_counter())] - # 횡단배수 지점은 새 노선에서 새로 계산한다 — 저장분을 남기면 옛 자리가 살아난다. - await asyncio.to_thread(clear_pipe_points, stored_path) - + # 관 정본은 지우지 않음(346-3) — 사용자 관 편집을 지키고 좌표로 새 노선을 따라감(체인 안). pool = get_db_pool() async with pool.acquire() as connection: await connection.begin() # 고른 지형 = 설계 지형 — 뒤 단계가 모두 이 확정값을 읽음 @@ -408,7 +402,8 @@ async def _recompute( stage2 = next( (s for s in (state or {}).get("stages", []) if int(s.get("stage_no", -1)) == 2), None ) - params = dict((stage2 or {}).get("params") or {}) + previous_params = dict((stage2 or {}).get("params") or {}) # 실패 때 되살림 + params = dict(previous_params) params["points"] = { "bp": points[0], "ep": points[-1], @@ -418,23 +413,28 @@ async def _recompute( await start_stage(cursor, str(project_id), 2, params) await connection.commit() - marks.append(("재계산 준비(관 삭제·노선 읽기·stage 기록)", time.perf_counter())) - - # 구조물은 땅 위 자리(좌표)를 따라감(344-2) — 옛 노선 경로 파일은 체인 전이라 아직 옛 것 - from B05_Profile.B05_Profile_Structures_Router import follow_structures_to_route - - try: - old_path = (latest or {}).get("route_data_path") - await asyncio.to_thread(follow_structures_to_route, project_root, old_path, points) - except Exception: # 따라감 실패 = 옛 누가거리 그대로(지금까지와 같음) - logger.exception("구조물 따라감 실패(옛 누가거리 유지): project_id=%s", project_id) + marks.append(("재계산 준비(노선 읽기·stage 기록)", time.perf_counter())) target = None if balanced is not None: from B05_Profile.B05_Profile_Router_RouteInitial import balanced_target target = await balanced_target(project_id, project_root, points, balanced.criteria) - failure = await run_redesign_chain(project_id, int(surface_model_id), selection, target) + if not latest: + return {"error": "저장된 노선이 없어 노선을 저장할 수 없습니다(자동설계 먼저)."} + # 노선 저장 체인(346-3) — 배수유역 뗌 · 관 · 구조물 따라감 · 종횡단 한 번 · 계획고 옮김 + saved = await run_route_save_chain( + project_id, + project_root, + stored_path, + int(surface_model_id), + selection, + latest, + points, + previous_params, + target, + ) + failure = saved.get("error") marks.append(("재확정 체인", time.perf_counter())) _log_steps("노선 [확인] 재계산", marks) if failure and switched: # 노선과 함께 설계 지형도 되돌림 @@ -448,14 +448,7 @@ async def _recompute( # 부르는 쪽이 계획노선을 되돌리도록 사유를 올려 보낸다. return {"error": failure} - pool = get_db_pool() - async with pool.acquire() as connection: - made = await get_latest_route(connection, project_id) - return { - "route_id": int(made["id"]) if made else None, - "total_length_m": float(made.get("total_length_m") or 0.0) if made else None, - "vertex_count": len(points), - } + return {**saved, "vertex_count": len(points)} @router.get("/{project_id}/route/plan", response_model=None) @@ -530,10 +523,16 @@ async def read_route_plan(project_id: UUID) -> dict[str, Any] | JSONResponse: @router.post("/{project_id}/route/replan", response_model=None) -async def replan_route( - project_id: UUID, request: RouteReplanRequest -) -> dict[str, Any] | JSONResponse: - """고친 계획노선으로 갈아 끼우고 배수유역부터 다시 계산한다(모달 [확인]).""" +async def replan_route(project_id: UUID, body: RouteSaveRequest) -> dict[str, Any] | JSONResponse: + """노선 초안을 정본으로 저장한다(346-3 · 요청 · 응답 모양 = `B05_Profile_Route_Save` 머리). + + 배수유역은 다시 안 셈(낡음 표시) · 초기값은 추천 [적용](`as_initial`)일 때만 다시 뜬다. + """ + as_initial = bool(getattr(body, "as_initial", False)) # 옛 요청 꼴(파이썬 호출)도 받음 + try: + request = body.resolved() if isinstance(body, RouteSaveRequest) else body + except ValueError as error: + return JSONResponse(status_code=400, content={"status": "error", "message": str(error)}) if len(request.vertices) < 2: return JSONResponse( status_code=400, @@ -616,15 +615,18 @@ async def replan_route( if choice else await asyncio.to_thread(read_initial_choice, project_root) ) - # 확정 노선 = 새 초기값(PLAN 74-4) — [저장] 없이 이 결과를 [초기화]가 돌아갈 자리로 뜬다. - from B05_Profile.B05_Profile_Router_RouteInitial import refresh_initial_snapshot + # 초기값 = 자동설계 체인 직후 + 추천 노선 [적용] 때만(346-3 · 사용자 확정 10-16) — 손으로 고친 + # 노선 저장은 [초기화]가 돌아갈 자리를 안 바꿈. + if as_initial: + from B05_Profile.B05_Profile_Router_RouteInitial import refresh_initial_snapshot - await refresh_initial_snapshot(project_id, project_root, result.get("route_id")) + await refresh_initial_snapshot(project_id, project_root, result.get("route_id")) return { "status": "success", "project_id": str(project_id), **result, "current_initial": current, + "initial_refreshed": as_initial, } diff --git a/B05_Profile/B05_Profile_Router_RouteInitial.py b/B05_Profile/B05_Profile_Router_RouteInitial.py index 53091ec56..2da4ffe04 100644 --- a/B05_Profile/B05_Profile_Router_RouteInitial.py +++ b/B05_Profile/B05_Profile_Router_RouteInitial.py @@ -337,7 +337,7 @@ async def balanced_profile_of( async def refresh_initial_snapshot( project_id: UUID, project_root: Path, route_id: int | None ) -> None: - """노선 [확인] 결과로 초기값(initial_snapshot)을 다시 뜬다(PLAN 74-4) — 실패는 비치명적.""" + """추천 노선 [적용](as_initial) 결과로 초기값을 다시 뜬다(346-3) — 실패는 비치명적.""" from common_util.common_util_initial_snapshot import ( discard_initial_snapshot, save_initial_snapshot, diff --git a/B06_Section/B06_Section_Router_Confirm.py b/B06_Section/B06_Section_Router_Confirm.py index 3a23ceddf..e535d552f 100644 --- a/B06_Section/B06_Section_Router_Confirm.py +++ b/B06_Section/B06_Section_Router_Confirm.py @@ -186,6 +186,18 @@ async def _recompute_stored_designs(project_id: UUID, route_id: int) -> None: ) +def _prune_cross_files(project_root: Path, longitudinal_file_path: str) -> None: + """측점을 고친 뒤 옛 cross_*.json 을 지움 — GET 은 안 지우므로 쓰는 쪽(저장 · 확정)이 맡음.""" + from B05_Profile.B05_Profile_Engine_Sections import prune_stale_cross_files + + path = project_root / longitudinal_file_path + try: + stations = json.loads(path.read_text(encoding="utf-8")).get("stations") or [] + except (OSError, ValueError, AttributeError): + return # 측점을 못 읽으면 안 지움(오삭제 막음) + prune_stale_cross_files(path.parent.parent / "cross_sections", stations) + + async def _prebuild_corridor(project_id: UUID, route_id: int, project_root: Path) -> None: """3D 길 모형을 서버 Node 가 지어 남긴다(342-2) — 정본이 다시 선 [저장] · [확정] 끝. @@ -261,6 +273,10 @@ async def save_sections( missing = await get_cross_sections_missing_design_chainages(connection, route_id) project_root = Path(resolve_stored_project_path(stored_path)) + # 측점 맞춤이 지운 자리의 옛 횡단 파일 정리(쓰는 쪽 몫) + await asyncio.to_thread( + _prune_cross_files, project_root, str(existing["longitudinal_file_path"]) + ) rowless = missing + await asyncio.to_thread( _rowless_station_chainages, project_root, @@ -341,6 +357,10 @@ async def confirm_sections( known = await get_cross_section_chainages(connection, route_id) marks.append(("조회(종단·경로·미지정 측점)", time.perf_counter())) project_root = Path(resolve_stored_project_path(stored_path)) + # 측점 맞춤이 지운 자리의 옛 횡단 파일 정리(쓰는 쪽 몫) + await asyncio.to_thread( + _prune_cross_files, project_root, str(existing["longitudinal_file_path"]) + ) # 행 자체가 없는 측점(구조물 등 비정규)도 확정 대상에 넣는다 — 정본이 없으면 # 조회 때마다 프리뷰가 다시 계산돼 3D·수량이 확정 결과가 아니게 된다(2026-08-24). missing = missing + await asyncio.to_thread( diff --git a/resources/tester/run_346_3_route_save_timing.py b/resources/tester/run_346_3_route_save_timing.py new file mode 100644 index 000000000..67e7081cf --- /dev/null +++ b/resources/tester/run_346_3_route_save_timing.py @@ -0,0 +1,57 @@ +# -*- coding: utf-8 -*- +"""346-3 노선 저장 실경로 시간 — 검증 프로젝트의 지금 노드 · 곡선 그대로 저장 한 번(수동 실행). + +같은 노선이라 값은 그대로(새 노선 행 번호만 바뀜) · 초기값은 안 건드림(`as_initial` 끔). +쓰는 법: `./venv/Scripts/python.exe resources/tester/run_346_3_route_save_timing.py ` +""" + +import asyncio +import json +import logging +import sys +import time +from pathlib import Path +from uuid import UUID + +ROOT = Path(__file__).resolve().parents[2] +sys.path.insert(0, str(ROOT)) + +from config.config_db import close_db_pool, init_db_pool # noqa: E402 + + +async def main(project_id: UUID) -> None: + from B05_Profile import B05_Profile_Router_Replan as replan + from B05_Profile.B05_Profile_Route_Save import RouteSaveRequest + + logging.basicConfig(level=logging.INFO, format="%(message)s") + await init_db_pool() + try: + plan = await replan.read_route_plan(project_id) + nodes = plan["nodes"] + radii: dict[int, float] = {} + for curve in plan["curves"]: + if curve.get("node_first") == curve.get("node_last"): + radii[int(curve["node_first"])] = float(curve["radius_m"]) + body = RouteSaveRequest( + vertices=[ + { + "x": node["x"], + "y": node["y"], + "curve": index in radii, + "radius_m": radii.get(index), + } + for index, node in enumerate(nodes) + ], + station_snap=bool(plan.get("station_snap")), + ) + started = time.perf_counter() + result = await replan.replan_route(project_id, body) + seconds = time.perf_counter() - started + body_text = getattr(result, "body", None) + print(json.dumps(json.loads(body_text) if body_text else result, ensure_ascii=False)) + print(f"[시간] 노선 저장 전체 {seconds:.1f}s") + finally: + await close_db_pool() + + +asyncio.run(main(UUID(sys.argv[1]))) diff --git a/resources/tester/test_133_2_design_terrain.py b/resources/tester/test_133_2_design_terrain.py index 6abb6a75c..92fddd064 100644 --- a/resources/tester/test_133_2_design_terrain.py +++ b/resources/tester/test_133_2_design_terrain.py @@ -18,7 +18,7 @@ import numpy as np import B03_FileInput.B03_FileInput_Service_Chain as chain import B04_PreProcess.B04_PreProcess_Service as service import B05_Profile.B05_Profile_Repository as route_repo -import common_util.common_util_drainage_pipes as drainage_pipes +import B05_Profile.B05_Profile_Route_Save as route_save import common_util.common_util_workflow_state as workflow_state from B05_Profile import B05_Profile_Router_Replan as replan from B06_Section import B06_Section_Router_Stations as stations @@ -212,16 +212,15 @@ def _patch_recompute(monkeypatch, tmp_path: Path, db: _Db, failure: str | None): async def start(*_args): return None - async def run_chain(_project_id, model_id, selection, _target=None): + async def run_chain(_project_id, _root, _stored, model_id, selection, *_rest): seen["model_id"], seen["selection"] = model_id, dict(selection) - return failure + return {"error": failure} if failure else {"route_id": 238} points = [{"x": 0.0, "y": 0.0}, {"x": 5.0, "y": 5.0}] monkeypatch.setattr(replan, "get_db_pool", lambda: _Pool()) - monkeypatch.setattr(drainage_pipes, "clear_pipe_points", lambda _s: None) monkeypatch.setattr(route_repo, "get_latest_route", latest) monkeypatch.setattr(chain, "_planned_route_points_in_project_crs", lambda *_a: points) - monkeypatch.setattr(chain, "run_redesign_chain", run_chain) + monkeypatch.setattr(route_save, "run_route_save_chain", run_chain) monkeypatch.setattr(workflow_state, "get_workflow_state", state) monkeypatch.setattr(workflow_state, "start_stage", start) return seen diff --git a/resources/tester/test_346_3_route_save.py b/resources/tester/test_346_3_route_save.py new file mode 100644 index 000000000..3a0618cb5 --- /dev/null +++ b/resources/tester/test_346_3_route_save.py @@ -0,0 +1,561 @@ +# -*- coding: utf-8 -*- +"""346-3 노선 저장 서버 길 — 차례 · 관 지킴 · 새 교차점 기본 관 · 오목 보충 · 이월 열쇠 · +초기값 규칙 · +배수유역 낡음 · 실패 때 옛 노선 되살림. + +순수 부품(`B05_Profile_Route_Save_Parts`)은 값으로 · 체인(`run_route_save_chain`)과 API +(`replan_route`)는 DB · 엔진을 가짜로 바꿔 차례 · 되살림 · 초기값만 잰다. +""" + +from __future__ import annotations + +import asyncio +import json +import sys +from pathlib import Path +from types import SimpleNamespace +from uuid import uuid4 + +import numpy as np +import pytest + +ROOT = Path(__file__).resolve().parents[2] +if str(ROOT) not in sys.path: + sys.path.insert(0, str(ROOT)) + +from fastapi.responses import JSONResponse # noqa: E402 + +import B03_FileInput.B03_FileInput_Service_Chain as chain # noqa: E402 +import B04_PreProcess.B04_PreProcess_Router_Watershed as watershed # noqa: E402 +import B05_Profile.B05_Profile_Route_Save as save # noqa: E402 +import B05_Profile.B05_Profile_Router as b05_router # noqa: E402 +import B05_Profile.B05_Profile_Router_Replan as replan # noqa: E402 +import B05_Profile.B05_Profile_Router_RouteInitial as route_initial # noqa: E402 +import B05_Profile.B05_Profile_Structures_Router as structures_router # noqa: E402 +import B06_Section.B06_Section_Repository as b06_repo # noqa: E402 +import B06_Section.B06_Section_Router as b06_router # noqa: E402 +import B06_Section.B06_Section_Router_Confirm as b06_confirm # noqa: E402 +import common_util.common_util_drainage_pipes as drainage_pipes # noqa: E402 +import common_util.common_util_initial_snapshot as snapshot # noqa: E402 +import common_util.common_util_workflow_state as workflow # noqa: E402 +from B05_Profile.B05_Profile_Route_Save_Parts import ( # noqa: E402 + add_crossing_pipes, + carry_cross_inputs, + concave_chainages, + fill_concave_pipes, + follow_pipes, + station_keys, + vertices_of, +) +from common_util.common_util_drainage_pipes import ( # noqa: E402 + PIPE_SOURCE_SPACING, + PIPE_SOURCE_STREAM, + PipePoint, + analysis_route_signature, + route_signature, +) + +OLD = [(0.0, 0.0), (100.0, 0.0), (200.0, 0.0), (300.0, 0.0), (400.0, 0.0)] +# 새 노선 = 100 ~ 300 m 를 북으로 휘게(앞 · 뒤는 옛 노선 위) +NEW = [(0.0, 0.0), (100.0, 0.0), (200.0, 60.0), (300.0, 0.0), (400.0, 0.0)] + + +# ── 관 따라감 · 교차점 · 오목 보충 ───────────────────────────────────────────── + + +def test_관은_지우지_않고_좌표를_따라감(): + old = [ + PipePoint(chainage_m=50.0, x=50.0, y=0.0, options={"pipe_kind": "흄관"}), + PipePoint(chainage_m=350.0, x=350.0, y=0.0, start_m=345.0, end_m=355.0), + ] + pairs = follow_pipes(old, vertices_of(NEW)) + assert [pipe_id for pipe_id, _ in pairs] == [0, 1] + assert pairs[0][1].options == {"pipe_kind": "흄관"} # 사용자 관 속성 그대로 + shift = pairs[1][1].chainage_m - 350.0 # 새 노선이 길어진 만큼 뒤 관이 밀림 + assert shift > 30 and pairs[1][1].start_m == pytest.approx(345.0 + shift) + assert old[1].chainage_m == 350.0 # 옛 목록은 안 고침 + + +def test_새_교차점에만_기본_관(): + pairs = follow_pipes([PipePoint(chainage_m=50.0, x=50.0, y=0.0)], vertices_of(NEW)) + crossings = [ + SimpleNamespace(chainage_m=20.0, x=20.0, y=0.0), # 옛 노선 위 — 사용자가 지운 관 안 되살림 + SimpleNamespace(chainage_m=150.4, x=150.0, y=30.0), # 새 자리 · 관 없음 → 더함 + SimpleNamespace(chainage_m=52.0, x=52.0, y=0.0), # 옛 자리 + ] + added = add_crossing_pipes(pairs, crossings, OLD, 456.0) + assert added == 1 + new = [pipe for pipe_id, pipe in pairs if pipe_id < 0] + assert new[0].chainage_m == 150.0 and new[0].source == PIPE_SOURCE_STREAM + # 같은 자리 다시 = 이미 관이 있음 + assert add_crossing_pipes(pairs, crossings, OLD, 456.0) == 0 + + +def test_오목_지점_보충은_새_자리의_오목만(): + vertices = vertices_of(NEW) + total = vertices[-1].chainage_m + chainage = np.arange(0.0, total, 2.0) + ground = 100.0 + np.abs(chainage - 220.0) * 0.05 # 220 m 에 골(새 자리) + ground[chainage < 60] = ( + 100.0 + np.abs(chainage[chainage < 60] - 30.0) * 0.05 + ) # 30 m 골(옛 자리) + concave = concave_chainages(chainage, ground) + assert any(abs(at - 220.0) < 3 for at, _ in concave) + pairs: list = [] + added = fill_concave_pipes(pairs, concave, vertices, OLD) + places = [pipe.chainage_m for _, pipe in pairs] + assert added >= 1 and all(pipe.source == PIPE_SOURCE_SPACING for _, pipe in pairs) + assert any(abs(at - 220.0) <= 3 for at in places) + assert not any(at < 60 for at in places) # 옛 노선 위 오목엔 안 둠 + + +def test_간격이_좁으면_보충_없음(): + vertices = vertices_of(NEW) + pairs = [(0, PipePoint(chainage_m=at)) for at in (0.0, 90.0, 180.0, 270.0, 360.0, 450.0)] + concave = [(220.0, 99.0)] + assert fill_concave_pipes(pairs, concave, vertices, OLD) == 0 + + +# ── 측점 열쇠 · 횡단 이월 ────────────────────────────────────────────────────── + + +def _structure(sid, type_id, anchor, start=None, end=None): + return SimpleNamespace( + structure_id=sid, + type_id=type_id, + chainage_m=anchor, + start_m=start, + end_m=end, + options={}, + ) + + +def test_측점_열쇠_규칙(): + stations = [{"chainage_m": s} for s in (0.0, 20.0, 40.0, 50.0, 60.0, 77.0, 80.0, 95.5)] + pipes = [(3, PipePoint(chainage_m=50.0))] + structures = [_structure("w1", "retaining_wall", 77.0, 60.0, 80.0)] + keys = station_keys(stations, pipes, structures, 20.0) + assert keys == { + "end:BP": 0.0, + "grid:1": 20.0, + "grid:2": 40.0, + "pipe:3": 50.0, + "structure:w1:start": 60.0, + "structure:w1": 77.0, + "structure:w1:end": 80.0, + "end:EP": 95.5, + } + + +def test_이월은_열쇠로_누가거리가_밀려도(): + old_keys = { + "end:BP": 0.0, + "grid:1": 20.0, + "pipe:0": 50.0, + "structure:w1": 77.0, + "grid:5": 100.0, + } + new_keys = {"end:BP": 0.0, "grid:1": 20.0, "pipe:0": 57.0, "structure:w1": 84.0} + designs = { + 0.0: {"ground_type": "soil"}, + 20.0: {"ground_type": "rock", "berm": {"on": True}, "cut_area_m2": 9.0}, + 50.0: {"ditch_type": "u", "inlet_structure": "basin", "basin_adjust": {"d": 1}}, + 77.0: {"rock_boundary_offset_m": 0.8, "revet_adjust": {"h": 2}}, + 100.0: {"ground_type": "soil"}, + } + inputs, adjusts, dropped = carry_cross_inputs( + old_keys, + new_keys, + designs, + ("ground_type", "ditch_type", "berm", "rock_boundary_offset_m"), + ("inlet_structure", "basin_adjust", "revet_adjust"), + ) + assert inputs == { + 0.0: {"ground_type": "soil"}, + 20.0: {"ground_type": "rock", "berm": {"on": True}}, # 결과 칸(면적)은 안 넘김 + 57.0: {"ditch_type": "u"}, + 84.0: {"rock_boundary_offset_m": 0.8}, + } + # 구조물 조정은 관 · 구조물 열쇠만 따라감 + assert adjusts == { + 57.0: {"inlet_structure": "basin", "basin_adjust": {"d": 1}}, + 84.0: {"revet_adjust": {"h": 2}}, + } + assert dropped == 1 # grid:5 — 새 노선에 없음(노선 밖 고침) + + +# ── 노선 초안 요청 ──────────────────────────────────────────────────────────── + + +def test_초안_모양을_옛_칸으로(): + body = save.RouteSaveRequest( + draft={ + "rev": 3, + "nodes": [[0, 0], [10, 5], [20, 0]], + "curveOn": [True, False, True], + "curveRadius": [None, 30.0, None], + "curveLock": [None, "radius", None], + "curveArc": [None, None, None], + "stationSnap": True, + "terrain": "sheet", + } + ) + request = body.resolved() + assert [(v.x, v.y, v.curve, v.radius_m) for v in request.vertices] == [ + (0, 0, True, None), + (10, 5, False, 30.0), + (20, 0, True, None), + ] + assert request.station_snap is True and request.terrain == "sheet" + with pytest.raises(ValueError): + save.RouteSaveRequest(draft={"nodes": [[0, 0], [1, 1]], "curveOn": [True]}).resolved() + + +# ── 체인 차례 · 되살림 · 배수유역 낡음 ───────────────────────────────────────── + + +class _Cursor: + def __init__(self, log): + self.log = log + + async def __aenter__(self): + return self + + async def __aexit__(self, *exc): + return False + + async def execute(self, sql, params=None): + self.log.append(" ".join(sql.split())) + + +class _Connection: + def __init__(self, log): + self.log = log + + def cursor(self, *args): + return _Cursor(self.log) + + async def begin(self): + return None + + async def commit(self): + return None + + async def rollback(self): + return None + + +class _Pool: + def __init__(self, log): + self.log = log + + def acquire(self): + connection = _Connection(self.log) + + class _Acquire: + async def __aenter__(self): + return connection + + async def __aexit__(self, *exc): + return False + + return _Acquire() + + +def _async(value, order=None, name=None): + async def fake(*args, **kwargs): + if order is not None: + order.append(name) + return value + + return fake + + +@pytest.fixture +def chain_world(monkeypatch, tmp_path): + order: list[str] = [] + sql: list[str] = [] + root = tmp_path + route_file = root / "B05_Profile" / "route" / "route_main.geojson" + route_file.parent.mkdir(parents=True) + route_file.write_text( + json.dumps({"geometry": {"coordinates": [[*p, 0] for p in OLD]}}), "utf-8" + ) + long_dir = root / "B06_Section" / "longitudinal" + long_dir.mkdir(parents=True) + (long_dir / "longitudinal.json").write_text(json.dumps({"stations": [], "old": True}), "utf-8") + drain = root / "drain" + pipes_file = drain / "edits" / "pipe_points.json" + pipes_file.parent.mkdir(parents=True) + pipes_file.write_text( + json.dumps( + {"route_signature": "old", "points": [{"chainage_m": 50.0, "x": 50.0, "y": 0.0}]} + ), + "utf-8", + ) + (drain / "00_watershed_response.json").write_text( + json.dumps({"route_signature": "old"}), "utf-8" + ) + state = {"stages": [{"stage_no": 2, "params": {"points": {}, "options": {}}}]} + + def solve(project_id, request, *, generate_sections=True): + order.append("solve") + assert generate_sections is False + route_file.write_text( + json.dumps({"geometry": {"coordinates": [[*p, 0] for p in NEW]}}), "utf-8" + ) + return _async( + SimpleNamespace( + route_id=8, + total_length_m=456.0, + route_data_path="B05_Profile/route/route_main.geojson", + ) + )() + + async def regenerate(*args, **kwargs): + order.append("sections") + (long_dir / "longitudinal.json").write_text( + json.dumps({"stations": [], "new": True}), "utf-8" + ) + return SimpleNamespace() + + async def watershed_must_not_run(*args, **kwargs): + raise AssertionError("배수유역은 체인에서 뗌") + + async def prepare(project_id): + order.append("pipes") + return {"vertices": vertices_of(NEW), "streams": []} + + monkeypatch.setattr(save, "get_db_pool", lambda: _Pool(sql)) + monkeypatch.setattr(workflow, "get_workflow_state", _async(state)) + monkeypatch.setattr(workflow, "start_stage", _async(None, order, "restore_params")) + monkeypatch.setattr( + b06_repo, + "get_longitudinal_section", + _async( + {"data": {}, "longitudinal_file_path": "B06_Section/longitudinal/longitudinal.json"} + ), + ) + monkeypatch.setattr(b06_repo, "get_cross_section_designs", _async([])) + monkeypatch.setattr(b06_repo, "get_latest_section_options", _async({})) + monkeypatch.setattr( + structures_router, + "follow_structures_to_route", + lambda *a: order.append("structures") or {"moved": 0, "far": 0}, + ) + monkeypatch.setattr( + chain, + "redesign_solve_request", + lambda *a: b05_router.RouteSolveRequest( + filter_key="csf", + method="dtm", + smooth=False, + algorithm="as_planned", + bp={"x": 0, "y": 0}, + ep={"x": 400, "y": 0}, + ), + ) + monkeypatch.setattr(b05_router, "solve_route_record", solve) + monkeypatch.setattr(b05_router, "confirm_latest_route", _async({})) + monkeypatch.setattr(watershed, "get_primary_region", watershed_must_not_run) + monkeypatch.setattr(watershed, "_prepare", prepare) + monkeypatch.setattr(drainage_pipes, "clear_pipe_points", watershed_must_not_run) + monkeypatch.setattr(drainage_pipes, "drainage_dir", lambda stored: drain) + monkeypatch.setattr(save, "_ground_along", lambda *a: (np.zeros(0), np.zeros(0))) + monkeypatch.setattr(b06_router, "regenerate_route_sections", regenerate) + monkeypatch.setattr(save, "_carry_profile", _async(None, order, "profile")) + monkeypatch.setattr(save, "_carry_cross", _async(2, order, "cross")) + monkeypatch.setattr( + b06_confirm, "confirm_sections", _async(SimpleNamespace(), order, "confirm") + ) + monkeypatch.setattr(snapshot, "mark_designing", lambda root: None) + monkeypatch.setattr(snapshot, "clear_designing", lambda root: None) + return SimpleNamespace( + order=order, + sql=sql, + root=root, + route_file=route_file, + pipes_file=pipes_file, + long_dir=long_dir, + drain=drain, + ) + + +def _run(world, **kwargs): + old_route = {"id": 7, "route_data_path": "B05_Profile/route/route_main.geojson"} + return asyncio.run( + save.run_route_save_chain( + uuid4(), + world.root, + "p", + 1, + {}, + old_route, + [{"x": x, "y": y} for x, y in NEW], + {"old": 1}, + **kwargs, + ) + ) + + +def test_차례와_배수유역_뗌(chain_world): + result = _run(chain_world) + assert "error" not in result, result + assert chain_world.order == [ + "structures", + "solve", + "pipes", + "sections", + "profile", + "cross", + "confirm", + ] + assert result["cross_dropped"] == 2 and result["drainage_stale"] is True + assert result["pipes"] == {"kept": 1, "crossing": 0, "concave": 0} + # 관 정본 = 새 노선 지문 · 배수유역 분석 지문(옛) 과 달라 B04 가 「낡음」 + saved = json.loads(chain_world.pipes_file.read_text("utf-8")) + assert saved["route_signature"] == route_signature(vertices_of(NEW)) + assert analysis_route_signature("p") != saved["route_signature"] + assert len(saved["points"]) == 1 # 사용자 관 그대로(지우지 않음) + + +def test_실패하면_옛_노선을_되살림(chain_world, monkeypatch): + monkeypatch.setattr( + b06_confirm, "confirm_sections", _async(JSONResponse(status_code=500, content={})) + ) + before_pipes = chain_world.pipes_file.read_text("utf-8") + result = _run(chain_world) + assert result["error"].startswith("종횡단 확정 실패") + assert json.loads(chain_world.route_file.read_text("utf-8"))["geometry"]["coordinates"] == [ + [*p, 0] for p in OLD + ] + assert chain_world.pipes_file.read_text("utf-8") == before_pipes + assert json.loads((chain_world.long_dir / "longitudinal.json").read_text("utf-8")) == { + "stations": [], + "old": True, + } + assert any( + sql.startswith("DELETE FROM routes WHERE project_id = %s AND id > %s") + for sql in chain_world.sql + ) + assert chain_world.order[-1] == "restore_params" + assert not list(chain_world.root.glob("route_save_*")) # 손에 쥔 사본 치움 + + +def test_추천_종단_반영이면_균형_종단(chain_world, monkeypatch): + monkeypatch.setattr( + route_initial, "apply_profile_target", _async(None, chain_world.order, "balanced") + ) + _run(chain_world, profile_target=([0.0, 456.0], [100.0, 110.0])) + assert "balanced" in chain_world.order and "profile" not in chain_world.order + + +# ── 초기값 규칙 ─────────────────────────────────────────────────────────────── + + +@pytest.fixture +def api_world(monkeypatch, tmp_path): + calls: list = [] + monkeypatch.setattr(replan, "_project_paths", _async((tmp_path, "p"))) + monkeypatch.setattr(replan, "_ensure_expected_route", lambda root: None) + monkeypatch.setattr(replan, "_plan_criteria", _async((20.0, 0.0, 0.0))) + monkeypatch.setattr(replan, "_ensure_planned_initial", lambda *a: None) + monkeypatch.setattr( + replan, + "_write_planned_polyline", + lambda *a, **k: {"vertices": 3, "curves": 0, "violations": 0}, + ) + monkeypatch.setattr(replan, "_recompute", _async({"route_id": 9, "total_length_m": 10.0})) + monkeypatch.setattr(replan, "read_initial_choice", lambda root: None) + monkeypatch.setattr( + route_initial, "refresh_initial_snapshot", lambda *a: calls.append(a) or _async(None)() + ) + return calls + + +def _body(**kwargs): + return save.RouteSaveRequest(vertices=[{"x": 0, "y": 0}, {"x": 5, "y": 5}], **kwargs) + + +def test_손으로_고친_노선_저장은_초기값_그대로(api_world): + result = asyncio.run(replan.replan_route(uuid4(), _body())) + assert result["initial_refreshed"] is False and api_world == [] + + +def test_추천_적용은_초기값_다시(api_world): + result = asyncio.run(replan.replan_route(uuid4(), _body(as_initial=True))) + assert result["initial_refreshed"] is True and len(api_world) == 1 + + +def test_초안_칸이_어긋나면_400(api_world): + body = save.RouteSaveRequest(draft={"nodes": [[0, 0], [1, 1]], "curveOn": [True]}) + response = asyncio.run(replan.replan_route(uuid4(), body)) + assert response.status_code == 400 + + +def _longitudinal(total: float) -> dict: + from B05_Profile.B05_Profile_Engine_Grade import resolve_grade_options + from B05_Profile.B05_Profile_Engine_Grade_Profile import design_ground_following_profile + + samples = [ + {"chainage_m": float(s), "elevation_m": 100.0 + 0.05 * s + 2.0 * np.sin(s / 17.0)} + for s in range(int(total) + 1) + ] + stations = [{"chainage_m": float(s)} for s in range(0, int(total) + 1, 20)] + longitudinal = {"samples": samples, "stations": stations} + alignment, profile = design_ground_following_profile( + longitudinal, resolve_grade_options("trunk") + ) + return {**longitudinal, "profile_alignment": alignment, "design_profiles": [profile]} + + +def test_계획고_옮김은_앞_그대로_뒤_밂(tmp_path, monkeypatch): + new_xy = [list(p) for p in NEW] + total = float(vertices_of(NEW)[-1].chainage_m) + old = _longitudinal(400.0) + old["profile_alignment"]["edits"] = { + "station_offsets": {"40.000": 0.7, "200.000": -0.4, "380.000": 0.3}, + "curve_radii": {}, + } + path = tmp_path / "B06_Section" / "longitudinal" / "longitudinal.json" + path.parent.mkdir(parents=True) + path.write_text(json.dumps(_longitudinal(total)), "utf-8") + import B05_Profile.B05_Profile_Repository as b05_repo + + monkeypatch.setattr(save, "get_db_pool", lambda: _Pool([])) + monkeypatch.setattr(b05_repo, "update_longitudinal_grade_summary", _async(None)) + asyncio.run( + save._carry_profile( + tmp_path, + "B06_Section/longitudinal/longitudinal.json", + 8, + old["profile_alignment"], + [list(p) for p in OLD], + new_xy, + ) + ) + edits = json.loads(path.read_text("utf-8"))["profile_alignment"]["edits"]["station_offsets"] + delta = total - 400.0 + assert edits["40.000"] == 0.7 # 바뀐 구간 앞 — 그대로 + assert "200.000" not in edits # 바뀐 구간 안 — 새 지반 + assert edits[f"{380.0 + delta:.3f}"] == 0.3 # 뒤 — 길이 차만큼 밂 + + +def test_저장_확정이_옛_횡단_파일을_지움(tmp_path): + from B05_Profile.B05_Profile_Engine_Sections import cross_filename + + long_path = tmp_path / "B06_Section" / "longitudinal" / "longitudinal.json" + cross_dir = tmp_path / "B06_Section" / "cross_sections" + long_path.parent.mkdir(parents=True) + cross_dir.mkdir(parents=True) + long_path.write_text(json.dumps({"stations": [{"chainage_m": 0.0}, {"chainage_m": 20.0}]})) + for at in (0.0, 20.0, 37.0): + (cross_dir / cross_filename(at)).write_text("{}") + b06_confirm._prune_cross_files(tmp_path, "B06_Section/longitudinal/longitudinal.json") + assert sorted(p.name for p in cross_dir.glob("cross_*.json")) == sorted( + cross_filename(at) for at in (0.0, 20.0) + ) + source = (ROOT / "B06_Section" / "B06_Section_Router_Confirm.py").read_text("utf-8") + for name in ("async def save_sections(", "async def confirm_sections("): + body = source[source.index(name) :] + assert body.index("_stand_patch_stations(") < body.index("_prune_cross_files,")