fix(492-1 노선 바뀜 초기화): 노선 모양이 바뀐 [저장] · 추천 [적용] = 서버 초기화 + 초기값 다시 뜸

- 판정 = 저장 · 새 계획 폴리라인 1 mm 견줌(`route_shape_changed` · TS 짝 `routeShapeChanged` = sub1) · 스냅 · 간격만 = 따라감 그대로
- 체인 초기화 = 사용자 구조물 · 추가 측점 · 관(옵션 포함) 버림 · 관 = 새 노선 세류선 교차만 · 1차 계획선 · 횡단 기본 설계 · 응답 route_reset {reset, dropped}
- 한 본문 초기화 = 보낸 관 · 구조물 · 추가 측점 · 계획고 · 관 옵션 · 넘김 칸 안 씀(페이지 설정만) · 되돌림 범위에 initial_snapshot
- 추천 초기 종단 미리 셈 = 같은 초기화 시설 · 지문에서 시설 파일 뺌
- 시험: 새 test_283_route_change_reset(옛 코드 8 실패) · 검증_sub8 자료 고정 · 실DB 시험 둘 = storage 임시 사본(helper_temp_storage) · 따라감 기대 시험 문구 고침

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
This commit is contained in:
2026-10-20 07:12:35 +09:00
co-authored by Claude Opus 5.5
parent f2a11dbac7
commit 7a857d626a
15 changed files with 691 additions and 132 deletions
+13 -1
View File
@@ -5,7 +5,8 @@
▲▼)은 이어받음.
- `changed_span` = 옛 · 새 노선 정점열의 같은 앞머리 · 같은 꼬리를 뺀 가운데 —
「노선 그대로?」 · 범위.
「노선 그대로?」 · 범위 · `route_shape_changed` = 노선 모양 바뀜 판정(492 · TS
`routeShapeChanged` 짝).
- 같은 자리(`line_matcher`) = 새 선 점을 옛 선에 투영 · 옆 `SAME_LINE_TOL_M` 안 · 방향 30° 안
(측점 스냅이 뒤 교점을 밀어도 이어받음).
- 자동 꼭짓점(`carry_base`) · 편집(`carry_edits`)은 **새 측점 위에** 다시 놓음 — 같은 자리면
@@ -80,6 +81,17 @@ def changed_span(
}
def route_shape_changed(
old_line: list[tuple[float, float]], new_line: list[tuple[float, float]]
) -> bool:
"""노선 모양이 바뀌었나(492 · 요구 284 · 285) — 저장 · 새 **계획 폴리라인** 정점 1 mm 견줌.
교점 · R · 곡선 켬끔은 선에 담김 · 측점 스냅 · 간격은 선을 안 바꿈(= 초기화 아님).
짝 = TS `logic/B05_Profile_Grade_Carry.ts` `routeShapeChanged`(sub1 · 거울 시험 = 그 시험).
"""
return changed_span(old_line, new_line) is not None
def _segments(xy: list[tuple[float, float]]) -> np.ndarray:
"""토막 표 [x, y, ux, uy, 길이, 시작 누가거리] — 길이 0 토막은 뺌(누가거리는 셈)."""
rows: list[tuple[float, ...]] = []
@@ -3,8 +3,8 @@
[적용] 체인(`B05_Profile_Route_Save.run_route_save_chain`)과 같은 조각을 메모리에서 부름(파일 안
씀):
후보 노드 → 화면 `flattenServerPlan` 꼴(`draft_nodes`) → `build_planned_polyline`(편집 갈래 · csv
4자리) → 관 · 구조물 · 추가 측점 따라감(`followed_pipes` · `followed_structures` ·
`followed_plus_rows` · 새 세류선 교차 관 포함) → 측점 틀 · 1차 계획선 438(`build_sections`) →
4자리) → 초기화 시설(492 · `reset_facilities` = 새 세류선 교차 관만 · 사용자 구조물 · 추가 측점 ·
관 버림) → 측점 틀 · 1차 계획선 438(`build_sections`) →
「종단 반영」 켬이면 1 m 균형 종단을 꼭짓점에 얹음(`seat_base_pvi` → `rebuild_alignment_profile`).
[적용]은 이 값을 받아 적지 않음 — 서버가 체인에서 같은 조각으로 다시 셈(CLAUDE.md 5).
@@ -88,19 +88,12 @@ def candidate_line(
return line, [curve.as_dict() for curve in outline.curves]
def followed_facilities(
project_root: Path,
line: list[tuple[float, float]],
old_route_data_path: str | None,
old_xy: list[list[float]],
old_pipes: list,
streams: list,
) -> tuple[list, list, list[float] | None]:
"""체인과 같은 따라감(파일 안 씀) → (관 · 구조물 · 추가 측점 누가거리 | 없음 None)."""
from B05_Profile.B05_Profile_Plus_Stations import followed_plus_rows, load_plus_stations
def reset_facilities(
line: list[tuple[float, float]], streams: list
) -> tuple[list, list, list[float]]:
"""체인 초기화(492)와 같은 시설(파일 안 씀) → (관 = 새 노선 세류선 교차 기본 관만 ·
구조물 없음 · 추가 측점 없음)."""
from B05_Profile.B05_Profile_Route_Save import followed_pipes
from B05_Profile.B05_Profile_Structures_Repository import load_structures
from B05_Profile.B05_Profile_Structures_Router import followed_structures
from common_util.common_util_drainage_pipes import (
ensure_pipe_ids,
fill_pipe_coordinates,
@@ -108,21 +101,13 @@ def followed_facilities(
)
points = [{"x": x, "y": y} for x, y in line]
items = load_structures(str(project_root))[1]
if items:
followed_structures(project_root, items, old_route_data_path, points)
rows = load_plus_stations(project_root)
plus = (
sorted({float(r["chainage_m"]) for r in followed_plus_rows(rows, line)["rows"]})
if rows
else None
)
pairs, _crossing, vertices = followed_pipes(old_pipes, old_xy, points, streams)
pairs, _crossing, vertices = followed_pipes([], [], points, streams)
pipes = [pipe for _, pipe in pairs]
fill_pipe_coordinates(pipes, vertices)
ensure_pipe_ids(pipes)
# 체인은 관 정본을 쓰고 다시 읽음(지문 같음 = 저장분 그대로) — 같은 왕복
return parse_pipe_points([pipe.as_dict() for pipe in pipes]), items, plus
# 추가 측점 = 빈 목록(None 이면 측점 틀이 저장 파일을 읽음)
return parse_pipe_points([pipe.as_dict() for pipe in pipes]), [], []
def initial_longitudinal(
+73 -24
View File
@@ -11,20 +11,23 @@ API — `POST /api/projects/{id}/route/replan`(`B05_Profile_Router_Replan.replan
`as_initial` — 추천 노선 [적용](초기값 다시 뜸) · 손으로 고친 노선 저장은 false(초기값 그대로).
응답 200 `{status: "success", project_id, route_id, total_length_m, vertex_count, current_initial,
initial_refreshed, pipes: {kept, crossing, outside}, structures: {moved, far, outside},
follow_failed, cross_dropped,
follow_failed, cross_dropped, route_reset: {reset, dropped: {structures, pipes, plus_stations}},
drainage_stale: true, seconds}`.
오류 400(노드 둘 미만 · 초안 칸 길이 어긋남) · 404(프로젝트 없음) ·
409 `{status: "error", message}`(체인 실패 — 노선 · 관 · 구조물 · 종횡단 파일 · 새 노선 행을
옛것으로 되살린 뒤).
규칙:
- 관 정본을 지우지 않음 — 좌표를 따라감 · 새 세류선 교차점 기본 관만 더함
- 노선 초기화(`initial` = 추천 [적용] · 모양 바뀐 노선 [저장] · 492 · 요구 284 · 285) = 사용자
구조물 · 추가 측점 · 관(옵션 포함) 버림 · 관 = 새 노선 세류선 교차 기본 관만 · 종단 1차 계획선 ·
횡단 기본 설계 · 초기값 다시 뜸(부르는 쪽).
- 그 밖(스냅 · 간격만) = 관 정본을 지우지 않음 — 좌표를 따라감 · 새 세류선 교차점 기본 관만 더함
(`B05_Profile_Route_Save_Parts`) · 배수유역 저장분은 노선 지문이 달라 「낡음」(B04 `_resolve`).
- 계획고 = 바뀐 자리만 새 지반 첫 계획선 · 같은 자리(옆 1 m 안) 사용자 계획고 지킴
(`B05_Profile_Engine_Grade_Carry` · 옛 · 새 **계획 폴리라인** — 화면 planline 과 같은 선) ·
추천 [적용](`as_initial`) = 옮기지 않음 — 균형 종단(`apply_profile_target`) · 1차 계획선(438) ·
횡단 손 입력도 안 넘김(요구 277 · 측점마다 기본 설계).
- 초기값은 여기서 안 뜸 — 부르는 쪽이 `as_initial` 일 때만(자동설계 체인 직후 + 추천 [적용]).
- 초기값은 여기서 안 뜸 — 부르는 쪽이 초기화일 때만(자동설계 체인 직후 + 노선 초기화).
"""
from __future__ import annotations
@@ -199,6 +202,7 @@ async def _place_pipes(
points: list[dict[str, float]],
) -> tuple[list, dict[str, int]]:
"""관 따라감 → 새 교차점 기본 관 → 정본 저장(새 노선 지문) · 간격 보충 없음(369).
초기화(492)면 부르는 쪽이 옛 관 · 옛 선을 빈 목록으로 줌 = 새 노선 교차 관만.
따라갈 선 = 구조물 · 추가 측점과 같은 `points`(사용 범위 자름 · 459) — B04 `_prepare`
노선은 범위를 안 잘라 누가거리가 갈림 · 세류선만 거기서. 좌표 없는 옛 관은 옛 노선
@@ -370,6 +374,36 @@ async def _rollback(project_id: UUID, old_route_id: int, params: dict[str, Any])
logger.exception("노선 저장 되살림(DB) 실패: project_id=%s", project_id)
async def _follow_facilities(
project_id: UUID,
project_root: Path,
old_route: dict[str, Any],
points: list[dict[str, float]],
follow_failed: list[str],
) -> dict[str, int]:
"""구조물 · 추가 측점 땅 위 자리 따라감(초기화 아닌 저장) → 구조물 요약 · 실패 = 옛 누가거리."""
from B05_Profile.B05_Profile_Plus_Stations import follow_plus_stations
from B05_Profile.B05_Profile_Structures_Router import follow_structures_to_route
# 구조물 = 옛 노선 파일이 아직 옛 것일 때 — 좌표 없는 옛 자료 채움
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}
follow_failed.append("구조물")
try: # 추가 측점(368-2)
await asyncio.to_thread(
follow_plus_stations, project_root, [(p["x"], p["y"]) for p in points]
)
except Exception: # noqa: BLE001
logger.exception("추가 측점 따라감 실패(옛 누가거리 유지): project_id=%s", project_id)
follow_failed.append("추가 측점")
return structures
async def run_route_save_chain(
project_id: UUID,
project_root: Path,
@@ -387,16 +421,24 @@ async def run_route_save_chain(
`planned_lines` = (옛 · 새 계획 폴리라인) — 바뀐 구간을 화면 planline 단계와 같은 선으로 가름
(없으면 풀린 노선 geojson · 점 간격이 달라 앞머리가 일찍 갈림).
`initial` = 추천 [적용](`as_initial`) — 옛 계획고 · 편집 · L 안 옮김 · 새 규칙 초기 종단(476).
`initial` = 노선 초기화(추천 [적용] · 모양 바뀐 [저장] · 492) — 사용자 시설 버림 · 옛 계획고 ·
편집 · L 안 옮김 · 새 규칙 초기 종단(476).
"""
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_Plus_Stations import follow_plus_stations, plus_stations_path
from B05_Profile.B05_Profile_Plus_Stations import (
load_plus_stations,
plus_stations_path,
save_plus_stations,
)
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 B05_Profile.B05_Profile_Structures_Repository import (
load_structures,
save_structures,
structures_file_path,
)
from B06_Section.B06_Section_Repository import (
get_cross_section_designs,
get_latest_section_options,
@@ -425,7 +467,8 @@ async def run_route_save_chain(
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))
structures_revision, old_structures = load_structures(str(project_root))
old_plus = load_plus_stations(project_root)
stage_dir = project_root / _STAGE_SUBDIR
backup = _Backup(
project_root,
@@ -440,22 +483,16 @@ async def run_route_save_chain(
mark_designing(project_root)
try:
follow_failed: list[str] = [] # 따라감 실패 = 옛 누가거리 그대로 · 화면이 알림(459)
# 구조물 = 땅 위 자리 따라감(옛 노선 파일이 아직 옛 것일 때 — 좌표 없는 옛 자료 채움)
try:
structures = await asyncio.to_thread(
follow_structures_to_route, project_root, old_route.get("route_data_path"), points
if initial: # 초기화(492) — 사용자 구조물 · 추가 측점 버림(실패면 `backup` 이 되살림)
if old_structures:
save_structures(str(project_root), [], base_revision=structures_revision)
if old_plus:
save_plus_stations(project_root, [], None)
structures = {"moved": 0, "far": 0, "outside": 0}
else:
structures = await _follow_facilities(
project_id, project_root, old_route, points, follow_failed
)
except Exception: # noqa: BLE001 — 따라감 실패 = 옛 누가거리 그대로(비치명)
logger.exception("구조물 따라감 실패(옛 누가거리 유지): project_id=%s", project_id)
structures = {"moved": 0, "far": 0}
follow_failed.append("구조물")
try: # 추가 측점도 땅 위 자리 따라감(368-2 · 실패 = 옛 누가거리 그대로)
await asyncio.to_thread(
follow_plus_stations, project_root, [(p["x"], p["y"]) for p in points]
)
except Exception: # noqa: BLE001
logger.exception("추가 측점 따라감 실패(옛 누가거리 유지): project_id=%s", project_id)
follow_failed.append("추가 측점")
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):
@@ -467,7 +504,10 @@ async def run_route_save_chain(
new_xy = _xy(project_root, solved.route_data_path)
marks.append(("노선 풀기 · 구조물 따라감", time.perf_counter()))
pairs, pipes = await _place_pipes(project_id, stored_path, old_pipes, old_xy, points)
# 초기화 = 옛 관(옵션 · 사용자 관) 버림 · 새 노선 세류선 교차 관만(492)
pairs, pipes = await _place_pipes(
project_id, stored_path, [] if initial else old_pipes, [] if initial else old_xy, points
)
marks.append(("관 따라감 · 교차점", time.perf_counter()))
seed = section_seed(request, old_section_options)
@@ -538,6 +578,15 @@ async def run_route_save_chain(
"structures": structures,
"follow_failed": follow_failed,
"cross_dropped": dropped,
# 노선 초기화(492 · sub1 약속) — 버린 사용자 시설 수
"route_reset": {
"reset": initial,
"dropped": {
"structures": len(old_structures) if initial else 0,
"pipes": len(old_pipes) if initial else 0,
"plus_stations": len(old_plus) if initial else 0,
},
},
"drainage_stale": True,
"seconds": round(marks[-1][1] - marks[0][1], 2),
}
+20 -11
View File
@@ -13,7 +13,8 @@
노선 저장(346-3) — 여기서 노선을 갈아 끼운 뒤 `B05_Profile_Route_Save.run_route_save_chain` 이
풀기 → 관 · 구조물 좌표 따라감 → 종횡단 한 번 → 계획고 옮김 → 횡단 이월(측점 열쇠) → B06 확정까지
돈다. 배수유역은 다시 안 셈(낡음) · 관 정본은 안 지움 · 초기값은 추천 [적용](`as_initial`)일 때만.
돈다. 배수유역은 다시 안 셈(낡음) · 노선 모양이 바뀌었거나 추천 [적용](`as_initial`)이면 초기화
(사용자 시설 · 손 입력 버림 · 초기값 다시 뜸 · 492) · 스냅 · 간격만 바꾼 저장은 시설 따라감.
요청 · 응답 · 오류 모양 = 그 모듈 머리.
GET /api/projects/{id}/route/plan → 예상노선·계획노선 정점(사업지 좌표계)
@@ -34,6 +35,7 @@ from fastapi import APIRouter
from fastapi.responses import JSONResponse
from B03_FileInput.B03_FileInput_Repository import get_project_storage_relative_path
from B05_Profile.B05_Profile_Engine_Grade_Carry import route_shape_changed
from B05_Profile.B05_Profile_Engine_RouteInitial import read_initial_choice, write_initial_choice
from B05_Profile.B05_Profile_Engine_RouteInitial_Station import snapped_initial
from B05_Profile.B05_Profile_Engine_Sections import load_station_snap, load_station_snap_step
@@ -331,7 +333,7 @@ async def _recompute(
marks = [("시작", time.perf_counter())]
# 관 정본은 지우지 않음(346-3) — 사용자 관 편집을 지키고 좌표로 새 노선을 따라감(체인 안).
# 관 정본 = 체인이 씀 — 초기화면 새 교차 관만 · 아니면 좌표로 새 노선을 따라감(346-3 · 492).
pool = get_db_pool()
async with pool.acquire() as connection:
await connection.begin() # 고른 지형 = 설계 지형 — 뒤 단계가 모두 이 확정값을 읽음
@@ -525,7 +527,8 @@ async def read_route_plan(project_id: UUID) -> 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`)일 때만 다시 뜬다.
배수유역은 다시 안 셈(낡음 표시) · 초기값은 노선 초기화(추천 [적용] · 모양 바뀜 · 492) 때
다시 뜬다.
추천 [적용] = 즉시 저장(요구 275 ②) · [저장]과 같은 프로젝트 잠금 · 떠 두고 실패면 되돌림(476 ·
472-6 `common_util_save_guard`) · 응답 `waited` · ⚠ `/design/save` 잠금 안에서 as_initial 금지.
"""
@@ -544,7 +547,9 @@ async def replan_route(project_id: UUID, body: RouteSaveRequest) -> dict[str, An
async with pool.acquire() as connection: # 설계 지형 바꿈(지표면 확정)도 떠 둠
rows = await guard.snapshot_rows(connection, str(project_id), surface=True)
files = await asyncio.to_thread(
guard.snapshot_storage, paths[0], guard.STORAGE_SCOPE + guard.PIPE_SCOPE
guard.snapshot_storage,
paths[0],
guard.STORAGE_SCOPE + guard.PIPE_SCOPE + guard.INITIAL_SCOPE,
)
try:
result = await _save_route(project_id, body)
@@ -620,7 +625,11 @@ async def _save_route(project_id: UUID, body: RouteSaveRequest) -> dict[str, Any
summary["violations"],
radius_m,
)
# 균형 종단 새로 풂 = 추천 [적용](`as_initial`)만 — 손으로 고친 노선은 계획고 옮김(344-2).
new_line = await asyncio.to_thread(_vertices_of, working_path)
# 노선 모양이 바뀜(교점 · R · 곡선 켬끔) · 추천 [적용] = 초기화(492 · 요구 284 · 285) — 사용자
# 시설 · 손 입력 버림 · 자동 다시 셈 · 초기값 다시 뜸 · 스냅 · 간격만 = 측점만 다시(따라감).
reset = as_initial or route_shape_changed(old_line, new_line)
# 균형 종단 새로 풂 = 추천 [적용](`as_initial`)만 — 손 노선 초기화 = 1차 계획선(476 길).
result = await _recompute(
project_id,
project_root,
@@ -631,8 +640,8 @@ async def _save_route(project_id: UUID, body: RouteSaveRequest) -> dict[str, Any
else None
),
request.terrain,
(old_line, await asyncio.to_thread(_vertices_of, working_path)),
as_initial,
(old_line, new_line),
reset,
)
if "error" in result:
# 노선을 되돌린다 — 실패했는데 새 노선만 남으면 화면·정본이 어긋난 채 굳는다.
@@ -662,9 +671,9 @@ async def _save_route(project_id: UUID, body: RouteSaveRequest) -> dict[str, Any
if choice
else await asyncio.to_thread(read_initial_choice, project_root)
)
# 초기값 = 자동설계 체인 직후 + 추천 노선 [적용] 때만(346-3 · 사용자 확정 10-16) — 손으로 고친
# 노선 저장은 [초기화]가 돌아갈 자리를 안 바꿈.
if as_initial:
# 초기값 = 자동설계 체인 직후 + 노선 초기화(추천 [적용] · 모양 바뀐 [저장] · 492) 때만 —
# 스냅 · 간격만 바꾼 저장은 [초기화]가 돌아갈 자리를 안 바꿈.
if reset:
from B05_Profile.B05_Profile_Router_RouteInitial import refresh_initial_snapshot
await refresh_initial_snapshot(project_id, project_root, result.get("route_id"))
@@ -673,7 +682,7 @@ async def _save_route(project_id: UUID, body: RouteSaveRequest) -> dict[str, Any
"project_id": str(project_id),
**result,
"current_initial": current,
"initial_refreshed": as_initial,
"initial_refreshed": reset,
}
@@ -41,11 +41,9 @@ async def initial_profile_of(
from B05_Profile import B05_Profile_Engine_InitialProfile as ip
from B05_Profile import B05_Profile_Router_Replan as replan
from B05_Profile.B05_Profile_Engine_Sections import _MODELS_SUBDIR
from B05_Profile.B05_Profile_Plus_Stations import plus_stations_path
from B05_Profile.B05_Profile_Repository import get_latest_route
from B05_Profile.B05_Profile_Route_Save import _xy, section_seed
from B05_Profile.B05_Profile_Route_Save import section_seed
from B05_Profile.B05_Profile_Router_RouteInitial import balanced_profile_of, project_standard
from B05_Profile.B05_Profile_Structures_Repository import structures_file_path
from B06_Section.B06_Section_Repository import (
get_latest_section_options,
get_route_generation_source,
@@ -54,7 +52,6 @@ async def initial_profile_of(
_regeneration_grade_options,
_regeneration_options,
)
from common_util.common_util_drainage_pipes import pipe_points_path, read_pipe_points_file
from common_util.common_util_surface_sampler import build_surface_sampler
from common_util.common_util_workflow_state import get_workflow_state
@@ -101,14 +98,8 @@ async def initial_profile_of(
"surface": surface,
"models": ip.stat_marks(list(models.glob(f"*_{surface['source_filter']}*.npz"))),
"streams": ip.stat_marks([_sheet_dir(stored_path) / STREAM_FILE]),
"files": ip.file_marks(
[
pipe_points_path(stored_path),
Path(structures_file_path(str(project_root))),
plus_stations_path(project_root),
project_root / route_path,
]
),
# 초기화(492) = 저장 관 · 구조물 · 추가 측점을 안 씀 — 노선 파일만
"files": ip.file_marks([project_root / route_path]),
"options": asdict(options),
"grade": grade_options.as_dict() if grade_options else None,
"crs": crs,
@@ -123,15 +114,7 @@ async def initial_profile_of(
return stored
prepared = await _prepare(project_id)
streams = [] if isinstance(prepared, JSONResponse) else prepared["streams"]
facilities = await asyncio.to_thread(
ip.followed_facilities,
project_root,
line,
route_path or None,
_xy(project_root, route_path),
read_pipe_points_file(pipe_points_path(stored_path)),
streams,
)
facilities = await asyncio.to_thread(ip.reset_facilities, line, streams)
target = None
if request.with_profile: # [적용] `balanced_target` 과 같은 셈(노선 정점 · 새 곡선표)
points = [{"x": x, "y": y} for x, y in line]
+13 -4
View File
@@ -8,7 +8,10 @@
(오류 응답 · 측점 맞춤 실패 · 재계산 실패 · 하위 차례 실패) DB 행 · storage 를 시작 전으로 되돌리고
그 차례 이름 · 까닭을 돌려줌(`common_util_save_guard` · 브레인 A). 되돌림마저 실패하면 「복구 확인
필요」. 같은 프로젝트 저장이 겹치면 뒤 요청이 기다림(`waited`).
노선 바뀐 저장(`route` · `carry` 칸 · 6-B) = 맨 앞 노선 저장 → 관 · 구조물 초안 다시 얹기 →
노선 모양 바뀐 저장(492 · 요구 284 · 285) = 맨 앞 노선 저장이 초기화(초기값 다시 뜸) · 뒤 차례는
페이지 설정만(보낸 시설 · 계획고 · 넘김 칸 안 씀) · 응답 `steps.route.route_reset`.
스냅 · 간격만 바뀐 노선 저장(`route` · `carry` 칸 · 6-B) = 맨 앞 노선 저장 → 관 · 구조물 초안
다시 얹기 →
위 차례(새 노선) · 횡단 = 옛 노선 세션 칸을 새 측점에 넘긴 patch · 상단측 안 보냄(요구 274) ·
늘 [저장](새 노선을 본 뒤 [확정]) · 되돌림 범위에 배수 폴더 · 지표면 확정(설계 지형 바꿈)까지
(`B06_Section_Router_DesignSave_Route`).
@@ -132,6 +135,12 @@ async def _run_steps(
before = await moving.saved_facilities(project_id)
routed = await step("route", replan_route(project_id, request.route))
route_id = int(routed["route_id"])
if (routed.get("route_reset") or {}).get("reset"):
# 노선 초기화(492) — 보낸 관 · 구조물 · 추가 측점 · 계획고 · 관 옵션 · 넘김 칸 안 씀
# (정본 = 서버 초기화 · 추천 [적용]과 한 벌) · 페이지 설정만
if request.settings is not None:
await step("settings", save_page_settings(project_id, request.settings))
return out
# 관 · 구조물 초안 = 서버가 따라 옮긴 정본 위에(Node · 옛 브라우저 `rebasePipeDraft` 짝)
try:
await moving.rebase_drafts(project_id, request, before)
@@ -213,9 +222,9 @@ async def save_design(
root = Path(resolve_stored_project_path(stored_path))
pipes = request.pipes is not None or bool(request.culvert_options) or request.confirm
pipes = pipes or moved # 노선 저장 = 관 따라감 · 새 교차 관(배수 폴더)
files = guard.snapshot_storage(
root, guard.STORAGE_SCOPE + (guard.PIPE_SCOPE if pipes else ())
)
scope = guard.STORAGE_SCOPE + (guard.PIPE_SCOPE if pipes else ())
# 노선 저장 = 초기화면 초기값 사본도 다시 뜸(492)
files = guard.snapshot_storage(root, scope + (guard.INITIAL_SCOPE if moved else ()))
try:
steps = await _run_steps(project_id, request, session)
except StepFailed as failed:
+4 -1
View File
@@ -4,7 +4,8 @@
- DB 행 — `projects` 한 줄 · `project_workflow_stages` · `routes` 와 자식 넷(`route_points` ·
`route_statistics` · `longitudinal_sections` · `cross_sections`)
- 노선 바뀐 저장(6-B)이면 지표면 모델 확정 상태(`surface_models.status` · 설계 지형 바꿈)
- storage — 저장 차례가 쓰는 폴더(`STORAGE_SCOPE` · 관 차례 · 노선 저장이면 배수 폴더도)
- storage — 저장 차례가 쓰는 폴더(`STORAGE_SCOPE` · 관 차례 · 노선 저장이면 배수 폴더 · 초기값
사본도)
를 떠 두고, 하나라도 실패하면 둘 다 되돌린다. 되돌림 쓰기는 바뀐 것만(시놀로지 동기 — 안 바뀐
파일 · 표는 안 씀). 같은 프로젝트 저장이 겹치면 뒤 요청이 기다림(`project_lock`).
"""
@@ -34,6 +35,8 @@ STORAGE_SCOPE = (
)
#: 관 · 관 옵션 차례 · [확정](배수유역 다시 셈)이 쓰는 자리.
PIPE_SCOPE = ("B04_PreProcess/drainage",)
#: 노선 초기화(추천 [적용] · 모양 바뀐 노선 [저장] · 492)가 다시 뜨는 초기값 사본.
INITIAL_SCOPE = ("initial_snapshot",)
#: 노선 아래 표(노선 행이 지워지면 함께 지워짐 · CASCADE).
ROUTE_CHILDREN = ("route_points", "route_statistics", "longitudinal_sections", "cross_sections")
File diff suppressed because one or more lines are too long
+74
View File
@@ -0,0 +1,74 @@
"""공유 storage 를 쓰는 실DB 시험 → 프로젝트 폴더를 임시 storage 사본으로(492 · 요구 283).
storage 는 Synology 로 데스크탑까지 번지므로 시험이 쓰는 자리는 사본이어야 함. 프로젝트 폴더를
`<임시>/storage/<같은 상대 경로>` 로 베끼고 `common_util_storage.STORAGE_BASE_DIR` 를 그 뿌리로 바꿈
— DB 의 저장 경로(`storage/…`)가 그대로 사본을 가리킴. 큰 원본 입력(파일 입력 · 전처리 산출)은
안 베끼고 정션(읽기만 · 저장 차례가 안 씀).
"""
from __future__ import annotations
import os
import shutil
import subprocess
import tempfile
from contextlib import contextmanager
from pathlib import Path
import pytest
#: 저장 차례가 안 쓰는 큰 입력 — 정션으로 이음
_LINKED = ("B03_FileInput", "B04_PreProcess/processed")
def _junction(target: Path, source: Path) -> None:
subprocess.run(
["cmd", "/c", "mklink", "/J", str(target), str(source)],
check=True,
capture_output=True,
)
def _copy(source: Path, target: Path, relative: str = "") -> None:
target.mkdir(parents=True, exist_ok=True)
for entry in source.iterdir():
name = f"{relative}/{entry.name}" if relative else entry.name
if name in _LINKED and entry.is_dir():
_junction(target / entry.name, entry)
elif entry.is_dir():
if any(link.startswith(f"{name}/") for link in _LINKED):
_copy(entry, target / entry.name, name)
else:
shutil.copytree(entry, target / entry.name)
else:
shutil.copy2(entry, target / entry.name)
def _unlink_junctions(root: Path) -> None:
for link in _LINKED:
path = root / link
if path.exists() or os.path.islink(path):
os.rmdir(path) # 정션만 걷음(원본 안 지움)
@contextmanager
def temp_project_storage(shared_root: Path):
"""공유 프로젝트 폴더 → 사본 뿌리(yield) · 끝나면 STORAGE_BASE_DIR 되돌리고 사본 지움."""
from common_util import common_util_storage as storage
base = Path(os.path.realpath(storage.STORAGE_BASE_DIR))
relative = Path(os.path.realpath(shared_root)).relative_to(base)
work = Path(tempfile.mkdtemp(prefix="aislo_storage_"))
copy_root = work / "storage" / relative
_copy(Path(shared_root), copy_root)
patch = pytest.MonkeyPatch()
patch.setattr(storage, "STORAGE_BASE_DIR", str(work / "storage"))
try:
# DB 경로가 사본을 가리키는지 — 아니면 공유 storage 를 쓰게 되므로 시작 전 멈춤
resolved = storage.resolve_stored_project_path(f"storage/{relative.as_posix()}")
assert Path(resolved) == Path(os.path.realpath(copy_root)), resolved
yield copy_root
finally:
patch.undo()
_unlink_junctions(copy_root)
shutil.rmtree(work, ignore_errors=True)
@@ -0,0 +1,337 @@
# -*- coding: utf-8 -*-
"""492-1 · 요구 283 · 284 · 285 — 노선 모양이 바뀐 [저장] · 추천 [적용] = 초기화 + 초기값 다시 뜸.
① 교점 끌기 [저장] — 판정(저장 · 새 계획 폴리라인) = 초기화 · 초기값 다시 뜸 · 체인 = 구조물 0 ·
추가 측점 0 · 관 = 새 교차만(stream · 옵션 없음 · 옆 0 m) · 계획고 옮김 없음 · 횡단 손 칸 0
② 추천 [적용] — 같은 초기화(균형 종단 갈래도)
③ 스냅 · 측점 간격만(선 그대로) — 초기화 아님 · 시설 따라감 · 초기값 그대로
④ 검증_sub8 자료(노선 522 → 628 · 관 p35100) — 관 = 새 교차 388 하나(옛 = 388 + 394 · 34.44 m 옆)
⑤ 한 본문 노선 초기화 — 보낸 관 · 구조물 · 추가 측점 · 계획고 · 관 옵션 · 넘김 칸 안 씀
⑥ 한 본문 실패 — 초기값 사본(`initial_snapshot`)까지 되돌림
옛 코드(492 전) = ① ② ④ ⑤ ⑥ 실패.
체인 · API 는 346-3 가짜 세계(`chain_world` · `api_world`)를 빌려 씀 — 파일은 임시 폴더만.
./venv/Scripts/python.exe -m pytest resources/tester/test_283_route_change_reset.py -q
"""
from __future__ import annotations
import asyncio
import json
import sys
from pathlib import Path
from types import SimpleNamespace
from uuid import uuid4
import pytest
HERE = Path(__file__).resolve().parent
ROOT = HERE.parents[1]
for path in (ROOT, HERE):
if str(path) not in sys.path:
sys.path.insert(0, str(path))
import test_346_3_route_save as base # noqa: E402
from test_346_3_route_save import _async, api_world, chain_world # noqa: E402, F401
import B04_PreProcess.B04_PreProcess_Engine_Watershed_Analyze as analyze # noqa: E402
import B05_Profile.B05_Profile_Route_Save as save # noqa: E402
import B05_Profile.B05_Profile_Router_Replan as replan # noqa: E402
import B06_Section.B06_Section_Repository as b06_repo # noqa: E402
from B05_Profile.B05_Profile_Plus_Stations import load_plus_stations, save_plus_stations # noqa: E402
from B05_Profile.B05_Profile_Structures_Repository import load_structures # noqa: E402
from common_util.common_util_json import atomic_write_json # noqa: E402
FIXTURE = HERE / "fixtures" / "req283_sub8_route.json"
WALL = {
"structure_id": "wall",
"type_id": "retaining_wall",
"placement": "interval",
"chainage_m": 250.0,
"start_m": 245.0,
"end_m": 255.0,
"options": {"wall_form": "콘크리트", "length_m": 10, "height_m": 2},
"memo": "",
"placement_source": "manual",
"status": "draft",
"revision": 0,
"geometry": None,
"summary_item": None,
"group_items": None,
"x": 250.0,
"y": 0.0,
}
# 새 노선(NEW) 100 ~ 200 m 토막 위 세류선 교차 — 옛 노선 밖
CROSSING = SimpleNamespace(chainage_m=158.31, x=150.0, y=30.0)
def _side(xy, x: float, y: float) -> float:
from shapely.geometry import LineString, Point
return LineString(xy).distance(Point(x, y))
def _seed(world, pipe_options: dict | None = None) -> None:
"""사용자 시설 — 관(옵션) · 구조물 · 추가 측점 · 옛 횡단 손 칸."""
world.pipes_file.write_text(
json.dumps(
{
"route_signature": "old",
"points": [
{
"chainage_m": 50.0,
"x": 50.0,
"y": 0.0,
"source": "user",
"options": pipe_options or {"pipe_kind": "흄관"},
}
],
}
),
"utf-8",
)
atomic_write_json(
world.root / "B05_Profile" / "route" / "structures.json",
{"revision": 1, "structures": [WALL], "definition_version": None},
)
save_plus_stations(world.root, [350], base.OLD)
@pytest.fixture
def crossings(monkeypatch):
"""세류선 교차 = 고정 목록(도엽 없이)."""
found = [CROSSING]
monkeypatch.setattr(analyze, "_base_pipes", lambda vertices, streams: list(found))
return found
@pytest.fixture
def cross_seen(monkeypatch):
seen: list = []
async def carry(*args):
seen.append(args[5]) # 옛 횡단 설계 — 넘길 손 입력
return 0
monkeypatch.setattr(save, "_carry_cross", carry)
monkeypatch.setattr(
b06_repo,
"get_cross_section_designs",
_async([{"chainage_m": 50.0, "design": {"inlet_basin_length_m": 3}}]),
)
return seen
def _assert_reset(world, result: dict, cross_seen: list) -> None:
assert "error" not in result, result
assert result["route_reset"] == {
"reset": True,
"dropped": {"structures": 1, "pipes": 1, "plus_stations": 1},
}
assert load_structures(str(world.root))[1] == []
assert load_plus_stations(world.root) == []
points = json.loads(world.pipes_file.read_text("utf-8"))["points"]
assert len(points) == 1, points # 옛 사용자 관 버림 · 새 교차 관만
assert points[0]["source"] == "stream" and not points[0].get("options")
assert _side(base.NEW, points[0]["x"], points[0]["y"]) < 1e-6
assert "profile" not in world.order # 계획고 옮김 없음 = 1차 계획선
assert cross_seen == [[]] # 횡단 손 입력 안 넘김
# ── ① 교점 끌기 [저장] ─────────────────────────────────────────────────────────
def test_판정_선이_바뀌면_참_같으면_거짓():
from B05_Profile.B05_Profile_Engine_Grade_Carry import route_shape_changed
line = [(0.0, 0.0), (100.0, 0.0), (200.0, 0.0)]
assert route_shape_changed(line, list(line)) is False
assert (
route_shape_changed(line, [(0.0, 0.0), (100.0, 0.0005), (200.0, 0.0)]) is False
) # 1 mm 안
assert route_shape_changed(line, [(0.0, 0.0), (100.0, 20.0), (200.0, 0.0)]) is True
assert route_shape_changed(line, [(0.0, 0.0), (200.0, 0.0)]) is True # 교점 지움
assert route_shape_changed(line, [*line, (300.0, 0.0)]) is True # 교점 더함
def _replan_world(monkeypatch, old_line, new_line) -> list:
calls: list = []
async def recompute(
project_id, root, stored, balanced=None, terrain=None, lines=None, initial=False
):
calls.append(("recompute", initial, balanced is not None))
return {"route_id": 9, "total_length_m": 10.0}
monkeypatch.setattr(replan, "_recompute", recompute)
monkeypatch.setattr(replan, "_planned_line", lambda root: old_line)
monkeypatch.setattr(replan, "_vertices_of", lambda path: new_line)
return calls
def test_교점_끌기_저장은_초기화_초기값_다시(api_world, monkeypatch):
calls = _replan_world(monkeypatch, [[0.0, 0.0], [4.0, 4.0]], [[0.0, 0.0], [5.0, 5.0]])
result = asyncio.run(replan.replan_route(uuid4(), base._body()))
assert calls == [("recompute", True, False)] # 균형 종단은 추천 [적용]만
assert result["initial_refreshed"] is True and len(api_world) == 1
def test_교점_끌기_체인은_사용자_시설_버림(chain_world, crossings, cross_seen):
_seed(chain_world)
result = base._run(chain_world, initial=True)
_assert_reset(chain_world, result, cross_seen)
assert result["pipes"]["crossing"] == 1
# ── ② 추천 [적용] ─────────────────────────────────────────────────────────────
def test_추천_적용도_같은_초기화(api_world, chain_world, crossings, cross_seen, monkeypatch):
import B05_Profile.B05_Profile_Router_RouteInitial as route_initial
same = [[0.0, 0.0], [5.0, 5.0]]
calls = _replan_world(monkeypatch, same, same)
result = asyncio.run(
replan.replan_route(uuid4(), base._body(as_initial=True, balanced_profile=True))
)
assert calls == [("recompute", True, True)] and result["initial_refreshed"] is True
_seed(chain_world)
monkeypatch.setattr(route_initial, "apply_profile_target", _async(None))
chained = base._run(chain_world, initial=True, profile_target=([0.0, 456.0], [1.0, 2.0]))
_assert_reset(chain_world, chained, cross_seen)
# ── ③ 스냅 · 간격만 ───────────────────────────────────────────────────────────
def test_스냅만_바꾼_저장은_초기화_아님(api_world, monkeypatch):
same = [[0.0, 0.0], [5.0, 5.0]]
calls = _replan_world(monkeypatch, same, [list(p) for p in same])
body = base._body(station_snap=True, station_snap_step=10.0)
result = asyncio.run(replan.replan_route(uuid4(), body))
assert calls == [("recompute", False, False)]
assert result["initial_refreshed"] is False and api_world == []
def test_스냅만_체인은_시설_따라감(chain_world, crossings, cross_seen):
_seed(chain_world)
result = base._run(chain_world)
assert result["route_reset"]["reset"] is False
assert result["route_reset"]["dropped"] == {"structures": 0, "pipes": 0, "plus_stations": 0}
assert len(load_structures(str(chain_world.root))[1]) == 1
assert load_plus_stations(chain_world.root)
points = json.loads(chain_world.pipes_file.read_text("utf-8"))["points"]
assert any(p.get("options") == {"pipe_kind": "흄관"} for p in points) # 사용자 관 그대로
# ── ④ 검증_sub8 자료 ─────────────────────────────────────────────────────────
@pytest.fixture
def sub8(monkeypatch):
data = json.loads(FIXTURE.read_text("utf-8"))
monkeypatch.setattr(base, "OLD", [tuple(p) for p in data["old_xy"]])
monkeypatch.setattr(base, "NEW", [tuple(p) for p in data["new_xy"]])
found = [SimpleNamespace(**c) for c in data["crossings"]]
monkeypatch.setattr(analyze, "_base_pipes", lambda vertices, streams: list(found))
return data
def test_검증_sub8_노선_522_628_관은_교차_하나(sub8, chain_world, cross_seen):
chain_world.pipes_file.write_text(
json.dumps({"route_signature": "old", "points": [sub8["old_pipe"]]}), "utf-8"
)
result = base._run(chain_world, initial=True)
assert "error" not in result, result
points = json.loads(chain_world.pipes_file.read_text("utf-8"))["points"]
assert [round(p["chainage_m"]) for p in points] == [388] # 옛 = [388, 394]
assert _side(sub8["new_xy"], points[0]["x"], points[0]["y"]) < 0.01 # 옛 394 = 34.44 m
assert not points[0].get("options")
assert result["route_reset"]["dropped"]["pipes"] == 1
# ── ⑤ · ⑥ 한 본문 ──────────────────────────────────────────────────────────────
def test_한_본문_초기화는_보낸_칸을_안_씀(monkeypatch):
import B04_PreProcess.B04_PreProcess_Router_Basins as basins
import B05_Profile.B05_Profile_Router as b05_router
import B05_Profile.B05_Profile_Router_Lifecycle as lifecycle
import B05_Profile.B05_Profile_Structures_Router as structures_router
import B06_Section.B06_Section_Router_Confirm as confirm
import B06_Section.B06_Section_Router_DesignSave as design_save
import B06_Section.B06_Section_Router_DesignSave_Route as moving
import B06_Section.B06_Section_Router_Stations as stations
called: list[str] = []
def record(name, value=None):
async def fake(*args, **kwargs):
called.append(name)
return {} if value is None else value
return fake
routed = {
"route_id": 9,
"route_reset": {
"reset": True,
"dropped": {"structures": 1, "pipes": 2, "plus_stations": 1},
},
}
monkeypatch.setattr(replan, "replan_route", record("route", routed))
monkeypatch.setattr(moving, "saved_facilities", record("before"))
monkeypatch.setattr(moving, "rebase_drafts", record("rebase"))
monkeypatch.setattr(moving, "carry_cross", record("carry", {"patches": [], "outside": 0}))
monkeypatch.setattr(lifecycle, "save_page_settings", record("settings"))
monkeypatch.setattr(basins, "put_pipe_points", record("pipes"))
monkeypatch.setattr(basins, "get_pipe_points", record("culvert", {"pipe_points": []}))
monkeypatch.setattr(structures_router, "write_structures", record("structures"))
monkeypatch.setattr(stations, "put_plus_stations", record("plus"))
monkeypatch.setattr(b05_router, "save_profile_alignment", record("alignment"))
monkeypatch.setattr(confirm, "save_sections", record("cross"))
request = design_save.DesignSaveRequest.model_construct(
route_id=7,
confirm=False,
settings=SimpleNamespace(),
pipes={"points": [{"chainage_m": 50.0}]},
structures=SimpleNamespace(),
plus_stations=SimpleNamespace(),
alignment=SimpleNamespace(model_copy=lambda update: SimpleNamespace(**update)),
culvert_options={"50.00": {"pipe_kind": "흄관"}},
uphill=None,
cross=None,
route=SimpleNamespace(),
carry=SimpleNamespace(),
)
out = asyncio.run(design_save._run_steps(uuid4(), request, None))
assert called == ["before", "route", "settings"], called
assert out["route"]["route_reset"]["reset"] is True
def test_한_본문_실패면_초기값_사본까지_되돌림(tmp_path, monkeypatch):
import B06_Section.B06_Section_Router_DesignSave as design_save
import common_util.common_util_save_guard as guard
from B06_Section.B06_Section_Router_DesignSave_Route import StepFailed
copy = tmp_path / "initial_snapshot" / "B04_PreProcess__drainage" / "edits"
copy.mkdir(parents=True)
(copy / "pipe_points.json").write_text("old", encoding="utf-8")
async def steps(project_id, request, session):
(copy / "pipe_points.json").write_text("new", encoding="utf-8") # 초기값 다시 뜸
raise StepFailed("route", "뒤 차례 실패")
monkeypatch.setattr(design_save, "get_db_pool", lambda: base._Pool([]))
monkeypatch.setattr(design_save, "get_project_storage_relative_path", _async("storage/x"))
monkeypatch.setattr(design_save, "resolve_stored_project_path", lambda stored: str(tmp_path))
monkeypatch.setattr(guard, "snapshot_rows", _async({}))
monkeypatch.setattr(guard, "restore_rows", _async(None))
monkeypatch.setattr(design_save, "_run_steps", steps)
request = design_save.DesignSaveRequest.model_construct(
route_id=7, confirm=False, pipes=None, culvert_options=None, route=SimpleNamespace()
)
response = asyncio.run(design_save.save_design(uuid4(), request, None))
assert response.status_code == 409
assert (copy / "pipe_points.json").read_text(encoding="utf-8") == "old"
+8 -4
View File
@@ -2,6 +2,8 @@
"""346-3 노선 저장 서버 길 — 차례 · 관 지킴 · 새 교차점 기본 관 · 보충 없음(369) · 이월 열쇠 ·
초기값 규칙 ·
배수유역 낡음 · 실패 때 옛 노선 되살림.
관 지킴 · 따라감 = 초기화 아닌 저장(스냅 · 간격만) 몫 — 모양 바뀐 저장 · 추천 [적용] = 초기화(492 ·
`test_283_route_change_reset`).
순수 부품(`B05_Profile_Route_Save_Parts`)은 값으로 · 체인(`run_route_save_chain`)과 API
(`replan_route`)는 DB · 엔진을 가짜로 바꿔 차례 · 되살림 · 초기값만 잰다.
@@ -480,7 +482,8 @@ def _body(**kwargs):
return save.RouteSaveRequest(vertices=[{"x": 0, "y": 0}, {"x": 5, "y": 5}], **kwargs)
def test_손으로_고친_노선_저장은_초기값_그대로(api_world):
def test_모양_그대로_노선_저장은_초기값_그대로(api_world):
# 492 — 저장 선 = 새 선(스냅 · 간격만) · 모양 바뀐 저장 = test_283_route_change_reset
result = asyncio.run(replan.replan_route(uuid4(), _body()))
assert result["initial_refreshed"] is False and api_world == []
@@ -490,13 +493,13 @@ def test_추천_적용은_초기값_다시(api_world):
assert result["initial_refreshed"] is True and len(api_world) == 1
def test_균형_종단은_추천_적용만_손_저장은_계획고_옮김(api_world, monkeypatch):
def test_균형_종단은_추천_적용만_손_저장은_초기화(api_world, monkeypatch):
calls: list = []
async def recompute(
project_id, root, stored, balanced=None, terrain=None, lines=None, initial=False
): # 476 — [적용] = initial 칸 더함
calls.append((balanced, lines))
calls.append((balanced, lines, initial))
return {"route_id": 9, "total_length_m": 10.0}
monkeypatch.setattr(replan, "_recompute", recompute)
@@ -504,7 +507,8 @@ def test_균형_종단은_추천_적용만_손_저장은_계획고_옮김(api_wo
monkeypatch.setattr(replan, "_vertices_of", lambda path: [[0.0, 0.0], [5.0, 5.0]])
asyncio.run(replan.replan_route(uuid4(), _body(balanced_profile=True))) # 손으로 고친 노선
asyncio.run(replan.replan_route(uuid4(), _body(balanced_profile=True, as_initial=True)))
assert calls[0] == (None, ([[0.0, 0.0], [4.0, 4.0]], [[0.0, 0.0], [5.0, 5.0]]))
# 손으로 모양 바꾼 노선 = 초기화(492) · 균형 종단 없음(1차 계획선)
assert calls[0] == (None, ([[0.0, 0.0], [4.0, 4.0]], [[0.0, 0.0], [5.0, 5.0]]), True)
assert calls[1][0] is not None # 추천 [적용] = 균형 종단 그대로
+24 -3
View File
@@ -4,7 +4,8 @@
② 실제 창구(검증_sub8 · DB · storage) — 중간 차례 실패 → DB 7 표 행 · storage 해시 시작 그대로 ·
차례 이름 · 까닭 · restored. 같은 본문 정상 = 옛 차례(안쪽 창구 차례로 부름)와 행 · 파일 같음.
검증 프로젝트가 없는 자리(다른 PC · DB 없음)는 ② 건너뜀.
② 는 살아 있는 검증_sub8 사용 · 노선 바뀌어도 됨(최신 노선 id · 추가 측점 자리를 그때 읽음).
② 는 살아 있는 검증_sub8 DB · storage 는 임시 사본(492 · `helper_temp_storage`) · 노선 바뀌어도 됨
(최신 노선 id · 추가 측점 자리를 그때 읽음).
./venv/Scripts/python.exe -m pytest resources/tester/test_472_6_design_save.py -q
"""
@@ -23,6 +24,7 @@ import pytest
HERE = Path(__file__).resolve().parent
ROOT = HERE.parents[1]
sys.path.insert(0, str(ROOT))
sys.path.insert(0, str(HERE))
from common_util import common_util_save_guard as guard # noqa: E402
@@ -77,8 +79,27 @@ def _db_ready() -> Path | None:
return None
ROOT_DIR = _db_ready()
live = pytest.mark.skipif(ROOT_DIR is None, reason="검증_sub8 · DB 없음")
SHARED_ROOT = _db_ready()
#: 시험이 쓰는 뿌리 = 공유 storage 의 임시 사본(492 · `sandbox` 가 세움 · Synology 로 안 번지게)
ROOT_DIR: Path | None = None
@pytest.fixture(scope="module")
def sandbox():
global ROOT_DIR
from helper_temp_storage import temp_project_storage
with temp_project_storage(SHARED_ROOT) as root:
ROOT_DIR = root
yield root
ROOT_DIR = None
def live(test):
"""실DB · 사본 storage 시험 — 검증 프로젝트 없으면 건너뜀."""
return pytest.mark.skipif(SHARED_ROOT is None, reason="검증_sub8 · DB 없음")(
pytest.mark.usefixtures("sandbox")(test)
)
def _state(root: Path) -> tuple[dict, dict]:
+96 -17
View File
@@ -6,9 +6,10 @@
② 다시 얹기 — 관 · 구조물 초안(`rebasePipeDraft` · `rebaseStructureDraft`) ==
Node `facility.rebase`.
③ 옛 코드 실패 — 한 본문이 `route` · `carry` 칸을 받음 · 브라우저가 노선 저장 창구를 안 부름.
④ 실제 창구(검증_sub8 · DB · storage) — 노선 차례 뒤 실패(시험 갈래 · 넘김 실패) = 노선 행 ·
자식 표 · storage 그대로 · 정상 한 본문 = 옛 차례(노선 저장 → 다시 얹기 → 횡단 저장)와
행 · 파일 같음.
④ 실제 창구(검증_sub8 DB · storage 는 임시 사본 · 492) — 노선 차례 뒤 실패(시험 갈래 · 넘김 실패)
= 노선 행 · 자식 표 · storage(초기값 사본 포함) 그대로 · 정상 한 본문(스냅만 바꾼 노선 =
따라감) = 옛 차례(노선 저장 → 다시 얹기 → 횡단 저장)와 행 · 파일 같음 · 모양 바뀐 노선 =
초기화(노선 차례만 · `route_reset`).
검증 프로젝트가 없는 자리(다른 PC · DB 없음)는 ④ 건너뜀.
./venv/Scripts/python.exe -m pytest resources/tester/test_472_6_route_save.py -q
@@ -30,6 +31,7 @@ import pytest
HERE = Path(__file__).resolve().parent
ROOT = HERE.parents[1]
sys.path.insert(0, str(ROOT))
sys.path.insert(0, str(HERE))
from common_util import common_util_save_guard as guard # noqa: E402
from common_util.common_util_calc_runner import run_calc # noqa: E402
@@ -175,8 +177,27 @@ def _db_ready() -> Path | None:
return None
ROOT_DIR = _db_ready()
live = pytest.mark.skipif(ROOT_DIR is None, reason="검증_sub8 · DB 없음")
SHARED_ROOT = _db_ready()
#: 시험이 쓰는 뿌리 = 공유 storage 의 임시 사본(492 · `sandbox` 가 세움 · Synology 로 안 번지게)
ROOT_DIR: Path | None = None
@pytest.fixture(scope="module")
def sandbox():
global ROOT_DIR
from helper_temp_storage import temp_project_storage
with temp_project_storage(SHARED_ROOT) as root:
ROOT_DIR = root
yield root
ROOT_DIR = None
def live(test):
"""실DB · 사본 storage 시험 — 검증 프로젝트 없으면 건너뜀."""
return pytest.mark.skipif(SHARED_ROOT is None, reason="검증_sub8 · DB 없음")(
pytest.mark.usefixtures("sandbox")(test)
)
def _rows() -> dict:
@@ -203,14 +224,17 @@ def _state(root: Path) -> tuple[dict, dict]:
}
files = {
relative: hashlib.sha1(path.read_bytes()).hexdigest()
for relative, path in guard._files(root, guard.STORAGE_SCOPE + guard.PIPE_SCOPE).items()
for relative, path in guard._files(
root, guard.STORAGE_SCOPE + guard.PIPE_SCOPE + guard.INITIAL_SCOPE
).items()
if not relative.startswith("_derived/")
}
return clean, files
def _plan_draft() -> dict:
"""지금 계획노선 노드 · 곡선 그대로 + 가운데 노드 하나를 3 m 옆으로(노선 끌기)."""
def _plan_draft(shape: bool = True) -> dict:
"""지금 계획노선 노드 · 넣은 편집값 그대로 + `shape` 면 가운데 노드 하나를 3 m 옆으로(노선 끌기
= 초기화) · 아니면 측점 스냅만 뒤집음(모양 그대로 = 따라감 · 492)."""
from B05_Profile.B05_Profile_Router_Replan import read_route_plan
from config.config_db import close_db_pool, init_db_pool
@@ -222,16 +246,26 @@ def _plan_draft() -> dict:
await close_db_pool()
plan = asyncio.run(run())
nodes = plan["nodes"]
at = len(nodes) // 2
points = [[float(n["x"]), float(n["y"])] for n in nodes]
points[at][1] += 3.0
from B05_Profile.B05_Profile_Engine_InitialProfile import draft_nodes
edits = plan.get("node_edits")
if edits: # 넣은 값 그대로(화면 `flattenServerPlan` 이 읽는 것)
points = [[float(n["x"]), float(n["y"])] for n in plan["nodes"]]
flags, radii = [bool(e["curve"]) for e in edits], [e["radius_m"] for e in edits]
else: # 편집값 없는 노선 = 화면 펴기(그린 R) 짝
xy, flags, radii = draft_nodes(plan["nodes"], plan["curves"])
points = [[x, y] for x, y in xy]
if shape:
points[len(points) // 2][1] += 3.0
snap = bool(plan.get("station_snap"))
return {
"rev": 1,
"nodes": points,
"curveOn": [True] * len(nodes),
"curveRadius": [None] * len(nodes),
"stationSnap": bool(plan.get("station_snap")),
"curveOn": flags,
"curveRadius": radii,
"curveLock": [e.get("lock") for e in edits] if edits else [None] * len(points),
"curveArc": [e.get("arc_m") for e in edits] if edits else [None] * len(points),
"stationSnap": snap if shape else not snap,
"stationSnapStep": plan.get("station_snap_step"),
}
@@ -286,7 +320,7 @@ def test_넘김_실패면_노선까지_그대로(monkeypatch) -> None:
return run_calc(ids, inputs)
monkeypatch.setattr(moving, "run_calc", broken)
status, body = _save(_body(_plan_draft()))
status, body = _save(_body(_plan_draft(shape=False))) # 넘김 = 모양 그대로 노선만
assert status == 409 and body["step"] == "cross" and body["restored"] is True, body
assert "넘김 일부러 실패" in body["message"]
assert _state(ROOT_DIR) == before
@@ -319,7 +353,7 @@ def test_정상_한_본문은_옛_차례와_같음() -> None:
from config.config_db import close_db_pool, get_db_pool, init_db_pool
start = _state(ROOT_DIR)
draft = _plan_draft()
draft = _plan_draft(shape=False)
old_ids = {int(row["id"]) for row in _rows()["routes"]}
async def old_way():
@@ -382,6 +416,51 @@ def test_정상_한_본문은_옛_차례와_같음() -> None:
assert _state(ROOT_DIR) == start
@live
def test_모양_바뀐_노선은_초기화_노선_차례만() -> None:
"""492 — 교점 3 m 끌기 = 서버 초기화 · 보낸 넘김 칸 안 씀 · 초기값 = 작업본 관 · 끝에 되돌림."""
from config.config_db import close_db_pool, get_db_pool, init_db_pool
start = _state(ROOT_DIR)
async def hold():
await init_db_pool()
try:
async with get_db_pool().acquire() as connection:
return await guard.snapshot_rows(connection, PROJECT, surface=True)
finally:
await close_db_pool()
rows = asyncio.run(hold())
files = guard.snapshot_storage(
ROOT_DIR, guard.STORAGE_SCOPE + guard.PIPE_SCOPE + guard.INITIAL_SCOPE
)
try:
status, body = _save(_body(_plan_draft()))
assert status == 200 and set(body["steps"]) == {"route"}, body
reset = body["steps"]["route"]["route_reset"]
assert reset["reset"] is True and body["steps"]["route"]["initial_refreshed"] is True
pipes = ROOT_DIR / "B04_PreProcess" / "drainage" / "edits" / "pipe_points.json"
kept = json.loads(pipes.read_text("utf-8"))["points"]
assert all(not p.get("options") and p.get("source") == "stream" for p in kept)
copy = ROOT_DIR / "initial_snapshot" / "B04_PreProcess__drainage" / "edits"
assert (copy / "pipe_points.json").read_bytes() == pipes.read_bytes() # 초기값 = 작업본
finally:
async def undo():
await init_db_pool()
try:
files.restore()
async with get_db_pool().acquire() as connection:
await guard.restore_rows(connection, PROJECT, rows)
finally:
await close_db_pool()
asyncio.run(undo())
files.discard()
assert _state(ROOT_DIR) == start
def test_저장한_노선_초안은_버림으로_되돌리지_않음() -> None:
"""화면 흠(6-B 화면 검증) — 저장 뒤 노선 초안 칸을 비우면 묶음이 「초안 버림」으로 보고 초안 전 관 ·
구조물 초안을 세션에 되써 다음 [저장]이 옛 자리로 되돌렸음 → 저장 성공이면 초안 전 값을 버림(commit)."""
+3 -2
View File
@@ -76,7 +76,8 @@ def test_적용_같은_노선_끔_새_1차_계획선(chain_world, monkeypatch):
assert not (alignment.get("edits") or {}).get("station_offsets")
def test_손_노선_저장은_계획고_옮김_그대로(chain_world, monkeypatch):
def test_초기화_아닌_저장은_계획고_옮김_그대로(chain_world, monkeypatch):
# 492 — 체인 initial=False = 노선 모양 그대로(스냅 · 간격만) 저장 몫
path, fresh = _real_profile_world(chain_world, monkeypatch)
_run(chain_world)
alignment = json.loads(path.read_text("utf-8"))["profile_alignment"]
@@ -104,7 +105,7 @@ def test_적용은_체인에_initial(api_world, monkeypatch):
def test_적용은_횡단_손_입력_버림(chain_world, monkeypatch):
# 요구 277 — 추천 [적용] = 새 노선 기준 초기화(횡단 손 입력 전부 버림) · 손 노선 [저장]은 넘김
# 요구 277 — 추천 [적용] · 모양 바뀐 [저장](492) = 초기화(횡단 손 입력 전부 버림) · 스냅 · 간격만은 넘김
import B06_Section.B06_Section_Repository as b06_repo
seen: list = []
@@ -2,7 +2,8 @@
"""476-1 추천 후보 초기 종단 — 미리 셈 = [적용] 체인 종단 · 화면 펴기 짝 · 응답 칸 · 저장본 지문.
고정 자료 = `fixtures/golden_472`(사본) · 체인 쪽 = `run_route_save_chain` 파일 길 차례(노선 csv →
구조물 · 추가 측점 따라감 → 노선 geojson → 관 따라감 · 정본 → `run_section_generation` →
초기화 = 구조물 · 추가 측점 비움(492) → 노선 geojson → 관 = 새 교차만 · 정본 →
`run_section_generation` →
켬이면 `_set_base_pvi` + `_apply_alignment_edits`) · 미리 셈 쪽 = 메모리 조각(파일 안 씀).
화면 펴기 = TS `flattenServerPlan`(Node `helper_476_1_flatten.cjs`) ↔ 파이썬 `draft_nodes`.
"""
@@ -113,24 +114,21 @@ def _chain(root: Path, draft: dict, target) -> dict:
from B05_Profile.B05_Profile_Engine import _route_geojson
from B05_Profile.B05_Profile_Engine_AsPlanned import _vertex_z
from B05_Profile.B05_Profile_Engine_Sections import run_section_generation
from B05_Profile.B05_Profile_Plus_Stations import follow_plus_stations
from B05_Profile.B05_Profile_Plus_Stations import save_plus_stations
from B05_Profile.B05_Profile_Route_Save import followed_pipes
from B05_Profile.B05_Profile_Router import _apply_alignment_edits
from B05_Profile.B05_Profile_Router_Replan import _vertices_of, _write_planned_polyline
from B05_Profile.B05_Profile_Router_RouteInitial import _set_base_pvi
from B05_Profile.B05_Profile_Structures_Router import follow_structures_to_route
from B05_Profile.B05_Profile_Structures_Repository import load_structures, save_structures
from common_util.common_util_drainage_pipes import (
ensure_pipe_ids,
fill_pipe_coordinates,
pipe_points_path_in,
read_pipe_points_file,
route_signature,
)
from common_util.common_util_json import atomic_write_json
from common_util.common_util_route_geometry import planned_route_working_path
old_xy = _old_xy(root)
old_pipes = read_pipe_points_file(pipe_points_path_in(root))
working = planned_route_working_path(root)
count = len(draft["nodes"])
_write_planned_polyline(
@@ -146,12 +144,14 @@ def _chain(root: Path, draft: dict, target) -> dict:
arcs=[None] * count,
)
points = [{"x": x, "y": y} for x, y in _vertices_of(working)]
follow_structures_to_route(root, P.ROUTE, points)
follow_plus_stations(root, [(p["x"], p["y"]) for p in points])
revision, items = load_structures(str(root)) # [적용] = 초기화(492) — 사용자 시설 버림
if items:
save_structures(str(root), [], base_revision=revision)
save_plus_stations(root, [], None)
xy = [(p["x"], p["y"]) for p in points]
z = _vertex_z(root, *P.SURFACE, xy)
atomic_write_json(root / P.ROUTE, _route_geojson([[x, y, h] for (x, y), h in zip(xy, z)]))
pairs, _crossing, vertices = followed_pipes(old_pipes, old_xy, points, [])
pairs, _crossing, vertices = followed_pipes([], [], points, [])
pipes = [pipe for _, pipe in pairs]
fill_pipe_coordinates(pipes, vertices)
ensure_pipe_ids(pipes)
@@ -168,18 +168,10 @@ def _chain(root: Path, draft: dict, target) -> dict:
def _preview(root: Path, candidate: dict, target) -> tuple[dict, list, list]:
from common_util.common_util_drainage_pipes import pipe_points_path_in, read_pipe_points_file
from common_util.common_util_surface_sampler import build_surface_sampler
line, curves = ip.candidate_line(candidate["nodes"], candidate["curves"], RADIUS)
facilities = ip.followed_facilities(
root,
line,
P.ROUTE,
_old_xy(root),
read_pipe_points_file(pipe_points_path_in(root)),
[],
)
facilities = ip.reset_facilities(line, [])
sampler = build_surface_sampler(root / "B04_PreProcess" / "models", *P.SURFACE)
longitudinal = ip.initial_longitudinal(
root,
@@ -385,7 +377,7 @@ def test_지문_같으면_꺼냄_다르면_다시(gather, monkeypatch):
doc = json.loads(gather["pipes"].read_text("utf-8"))
doc["points"][0]["x"] = float(doc["points"][0]["x"]) + 1.0 # 관 하나 옮김
gather["pipes"].write_text(json.dumps(doc), "utf-8")
assert "다시 셈" in gather["run"]()["error"]
assert gather["run"]() == first # 초기화(492) = 저장 관을 안 씀 → 지문 그대로
def test_코드_바뀌면_다시(gather, monkeypatch):