From 858ef3b40f44c7cca402391bed62b0f483f3102f Mon Sep 17 00:00:00 2001 From: umsangdon Date: Mon, 19 Oct 2026 16:31:16 +0900 Subject: [PATCH 1/3] =?UTF-8?q?fix(476-1=20=E2=91=A0):=20=EC=B6=94?= =?UTF-8?q?=EC=B2=9C=20[=EC=A0=81=EC=9A=A9]=20=3D=20=EB=8A=98=20=EC=83=88?= =?UTF-8?q?=20=EA=B7=9C=EC=B9=99=20=EC=B4=88=EA=B8=B0=20=EC=A2=85=EB=8B=A8?= =?UTF-8?q?=20=E2=80=94=20=EC=98=9B=20=EA=B3=84=ED=9A=8D=EA=B3=A0=20=C2=B7?= =?UTF-8?q?=20=ED=8E=B8=EC=A7=91=20=C2=B7=20L=20=EC=95=88=20=EC=98=AE?= =?UTF-8?q?=EA=B9=80?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - run_route_save_chain `initial`(= as_initial) — 끔 · 균형 셈 실패면 `_carry_profile` 안 부름 = 막 쓴 1차 계획선(438 · L 20) 그대로 - 손 노선 [저장]은 그대로 계획고 옮김 - 횡단 손 입력 넘김 판정 한 자리 `carries_cross_inputs`(사용자 답 대기 · 지금 = 넘김) - 시험 test_476_1_apply_initial · 346-3 가짜 _recompute 칸 더함 Co-Authored-By: Claude Opus 5.5 --- B05_Profile/B05_Profile_Route_Save.py | 13 ++- B05_Profile/B05_Profile_Router_Replan.py | 3 + .../concepts/design_page/operations_b05.md | 2 +- docs/wiki/concepts/design_page/rules.md | 1 + resources/tester/test_346_3_route_save.py | 4 +- resources/tester/test_476_1_apply_initial.py | 106 ++++++++++++++++++ 6 files changed, 124 insertions(+), 5 deletions(-) create mode 100644 resources/tester/test_476_1_apply_initial.py diff --git a/B05_Profile/B05_Profile_Route_Save.py b/B05_Profile/B05_Profile_Route_Save.py index ebf385d78..d309e903f 100644 --- a/B05_Profile/B05_Profile_Route_Save.py +++ b/B05_Profile/B05_Profile_Route_Save.py @@ -22,7 +22,7 @@ API — `POST /api/projects/{id}/route/replan`(`B05_Profile_Router_Replan.replan (`B05_Profile_Route_Save_Parts`) · 배수유역 저장분은 노선 지문이 달라 「낡음」(B04 `_resolve`). - 계획고 = 바뀐 자리만 새 지반 첫 계획선 · 같은 자리(옆 1 m 안) 사용자 계획고 지킴 (`B05_Profile_Engine_Grade_Carry` · 옛 · 새 **계획 폴리라인** — 화면 planline 과 같은 선) · - 균형 종단(`apply_profile_target`)은 추천 [적용](`as_initial`)만. + 추천 [적용](`as_initial`) = 옮기지 않음 — 균형 종단(`apply_profile_target`) 또는 1차 계획선(438). - 초기값은 여기서 안 뜸 — 부르는 쪽이 `as_initial` 일 때만(자동설계 체인 직후 + 추천 [적용]). """ @@ -258,6 +258,11 @@ async def _carry_profile( await connection.commit() +def carries_cross_inputs(initial: bool) -> bool: + """횡단 손 입력을 새 측점(같은 자리)에 넘기나 — 추천 [적용]도 넘김(475-3 · 사용자 답 대기 476).""" + return True + + async def _carry_cross( project_id: UUID, project_root: Path, @@ -358,11 +363,13 @@ async def run_route_save_chain( previous_params: dict[str, Any], profile_target: tuple[list[float], list[float]] | None = None, planned_lines: tuple[list[list[float]], list[list[float]]] | None = None, + initial: bool = False, ) -> dict[str, Any]: """노선 저장 체인 — 성공이면 요약 dict · 실패면 `{"error": 사유}`(파일 · 행은 되살린 뒤). `planned_lines` = (옛 · 새 계획 폴리라인) — 바뀐 구간을 화면 planline 단계와 같은 선으로 가름 (없으면 풀린 노선 geojson · 점 간격이 달라 앞머리가 일찍 갈림). + `initial` = 추천 [적용](`as_initial`) — 옛 계획고 · 편집 · L 안 옮김 · 새 규칙 초기 종단(476). """ import aiomysql @@ -465,7 +472,7 @@ async def run_route_save_chain( from B05_Profile.B05_Profile_Router_RouteInitial import apply_profile_target await apply_profile_target(project_id, new_route_id, profile_target) - else: + elif not initial: # [적용] = 막 쓴 1차 계획선(438 · L 20) 그대로(476 · 요구 276) await _carry_profile( project_root, relative, @@ -482,7 +489,7 @@ async def run_route_save_chain( new_route_id, relative, old_long, - old_designs, + old_designs if carries_cross_inputs(initial) else [], # 옛 관 = 옛 노선 누가거리(B05 가 읽는 자리 · 정본은 다른 선 누가거리일 수 있음 · 459) [pipe for _, pipe in follow_pipes(old_pipes, vertices_of(old_xy))], old_structures, diff --git a/B05_Profile/B05_Profile_Router_Replan.py b/B05_Profile/B05_Profile_Router_Replan.py index 6111af307..2b008b81d 100644 --- a/B05_Profile/B05_Profile_Router_Replan.py +++ b/B05_Profile/B05_Profile_Router_Replan.py @@ -333,6 +333,7 @@ async def _recompute( balanced: RouteInitialRequest | None = None, terrain: Literal["las", "sheet"] | None = None, planned_lines: tuple[list[list[float]], list[list[float]]] | None = None, + initial: bool = False, ) -> dict[str, Any]: """노선을 갈아 끼운 뒤의 재계산 — 초기 업로드 체인과 같은 로직을 그대로 탄다. @@ -441,6 +442,7 @@ async def _recompute( previous_params, target, planned_lines, + initial, ) failure = saved.get("error") marks.append(("재확정 체인", time.perf_counter())) @@ -621,6 +623,7 @@ async def replan_route(project_id: UUID, body: RouteSaveRequest) -> dict[str, An ), request.terrain, (old_line, await asyncio.to_thread(_vertices_of, working_path)), + as_initial, ) if "error" in result: # 노선을 되돌린다 — 실패했는데 새 노선만 남으면 화면·정본이 어긋난 채 굳는다. diff --git a/docs/wiki/concepts/design_page/operations_b05.md b/docs/wiki/concepts/design_page/operations_b05.md index 352272882..af516d1c1 100644 --- a/docs/wiki/concepts/design_page/operations_b05.md +++ b/docs/wiki/concepts/design_page/operations_b05.md @@ -21,7 +21,7 @@ source: ["docs/raw/조사/2026-10-19_470_B05B06_데이터흐름.md", "docs/raw/ | 교점 추가 · 삭제 | 편집기 메모리 → `route-draft` | 묶음 전 단계 · 측점 수 바뀜 → 갈아끼움 | 같음 | 같음 | 11 | `B05_Profile_UI_RouteEdit_Input.ts:149-158` · `logic/B05_Profile_UI_RouteEdit_Edits.ts:410-426` | | 곡선 R · L | 편집기 `curveRadius` · `curveArc` → `route-draft` | 묶음 전 단계 · 확폭 · MC 이동 → 곡선 열쇠 옮김(`curveMoves`) | 같음 | `/design/save` `route` 칸 — 넣은 R · L 로 서버가 같은 규칙 | 11 | `B05_Profile_UI_RouteEdit_CurveBar.ts:76-162` · `logic/B05_Profile_Sections_Stations.ts:332-337` · `B05_Profile_Router_Replan.py:582 · 599-600` | | 스냅 켬 · 간격 | 간격 = pref `routeedit-snap-step` · 켬 = `route-draft.stationSnap` | 켬 = 곡선 IP 를 옮김(한 걸음) → 묶음 전 단계 | — | 곡선표 파일 · 서버는 측점만 맞춤(IP 맞춤은 브라우저만) | 3 | `B05_Profile_UI_RouteEdit_SnapStep.ts:74-78` · `logic/design_pipeline_stations.ts:52-59` · `B05_Profile_Engine_Sections_Core.py:457-464` | -| 추천 노선 [적용] | result `routeedit-initial` → `route-draft` | 묶음 한 바퀴 | 미리 셈 `/route/initial` → `route_recommend/*` | **저장 밖** — 바로 `/route/replan`(as_initial) · 초기값 다시 뜸 · 초안 전부 지움 · 페이지 다시 세움 | 11 | `B05_Profile_UI_RouteEdit_Initial.ts:139 · 161-181` · `B05_Profile_UI_RouteEdit_Apply.ts:211 · 220-245` · `B05_Profile_Router_Replan.py:659-662` | +| 추천 노선 [적용] | result `routeedit-initial` → `route-draft` | 묶음 한 바퀴 | 미리 셈 `/route/initial` → `route_recommend/*` | **저장 밖** — 바로 `/route/replan`(as_initial) · 종단 = 새 규칙 초기 종단(옛 계획고 · 편집 · L 안 옮김 · 476) · 초기값 다시 뜸 · 초안 전부 지움 · 페이지 다시 세움 | 11 | `B05_Profile_UI_RouteEdit_Initial.ts:139 · 161-181` · `B05_Profile_UI_RouteEdit_Apply.ts:211 · 220-245` · `B05_Profile_Router_Replan.py:659-662` | ## 시설(관 · 구조물 · 추가 측점) diff --git a/docs/wiki/concepts/design_page/rules.md b/docs/wiki/concepts/design_page/rules.md index 271b5546c..4854979b6 100644 --- a/docs/wiki/concepts/design_page/rules.md +++ b/docs/wiki/concepts/design_page/rules.md @@ -30,6 +30,7 @@ source: ["CLAUDE.md §5", "docs/raw/조사/2026-10-19_465_요구272_값안변함 | 측점 열쇠 = `common_util_station_key` 한 벌 · 노선 바뀜 = 관 · 구조물 좌표(x · y) 따라감 · 누가거리 다시 매김 | `common_util_station_key.ts:32` · `common_util_route_follow.ts:291` | `test_344_3_station_key_mirror` · `test_410_follow_xy` | | 노선 바뀜 넘김 = 열쇠마다 지반 · 측구 종류 · 측구 선택 · 사면 쉬프트 · 소단 · 소단 깊이 · 자리 칸(`SEAT_KEYS` = 암 경계 · 암 절토 경사 · 2단 경사 · 손 상단측 = 단면 유형 · 측구 방향)은 같은 자리(`sameSeat` · 중심 0.01 m · 축 1e-3) 측점만 · 새 자리 = 기본값 · 세션 값은 쓴 때 자리(`@seat`)로 판정(↶ ↷ 유지) · 끌기 갈아끼움 · 미리보기 · Node 넘김 · 서버 노선 저장 한 규칙 | `logic/B06_Section_Station_Swap.ts` SEAT_KEYS · unseated · `logic/B06_Section_Route_Carry.ts` carryKeyedStates · `B05_Profile/logic/B05_Profile_Pipeline_Detail.ts` crossAtSeat · `B05_Profile_Route_Save_Parts.py` carry_cross_inputs | `test_475_3_seat_carry` | | 노선 초안 ↶(버림) = 차례 하나 · 버림 동안 종단도 그리기 멈춤 · 낡음 다시 셈 미룸 → 되돌림 · 저장 기준 맞춤 · 뒤 차례 끝에 한 번 그림(사이 섞인 유토곡선 없음) | `design_pipeline_page.ts:642 · 403` · `B06_Section_Section_Store.ts:316` | `test_472_5c_drop_once` | +| 추천 [적용] 종단 = 새 규칙 초기 종단 — 끔 · 균형 셈 실패 = 1차 계획선(438 · L 20) · 켬 = 균형 종단을 꼭짓점에 얹음(438) · 옛 계획고 · 편집 · L 안 옮김 · 손 노선 저장만 옮김 | `B05_Profile_Route_Save.py` run_route_save_chain `initial` | `test_476_1_apply_initial` | | B05 · B06 캐시 = 한 벌 · [저장] 한 번에 함께 나감 | `design_shared.ts:395-448` · `design_save.ts:237` | — | | 관 · 구조물 주인 = 가게 칸 하나 — 화면 사본은 가게 알림으로 맞춤(제 쓰기는 같은 값 · 남 고침은 다시 실음) | `design_view_profile.ts` 가게 구독 · `design_view_cross.ts` onShop · `B06_Section_UI_Page_Structures_Panel.ts` seen | `test_472_5_shop_owner` | | 정본 자리 = 값을 낳은 주인 — 노선 시설 = `pipe_points.json` · `structures.json` · 횡단 값 = 횡단 패치 | `B05_Profile_Api_Pipes_Draft.ts:30-37` · `B05_Profile_Structures_Router.py:386-405` · `B06_Section_Section_Store.ts:449` | — | diff --git a/resources/tester/test_346_3_route_save.py b/resources/tester/test_346_3_route_save.py index 4ed12bff6..d5b5fc733 100644 --- a/resources/tester/test_346_3_route_save.py +++ b/resources/tester/test_346_3_route_save.py @@ -487,7 +487,9 @@ def test_추천_적용은_초기값_다시(api_world): def test_균형_종단은_추천_적용만_손_저장은_계획고_옮김(api_world, monkeypatch): calls: list = [] - async def recompute(project_id, root, stored, balanced=None, terrain=None, lines=None): + async def recompute( + project_id, root, stored, balanced=None, terrain=None, lines=None, initial=False + ): # 476 — [적용] = initial 칸 더함 calls.append((balanced, lines)) return {"route_id": 9, "total_length_m": 10.0} diff --git a/resources/tester/test_476_1_apply_initial.py b/resources/tester/test_476_1_apply_initial.py new file mode 100644 index 000000000..16c814f9d --- /dev/null +++ b/resources/tester/test_476_1_apply_initial.py @@ -0,0 +1,106 @@ +# -*- coding: utf-8 -*- +# ruff: noqa: F811 — 346-3 fixture 를 불러 씀 +"""476-1 ① 추천 [적용] = 늘 새 규칙 초기 종단(요구 276) — 옛 계획고 · 편집 · L 안 옮김. + +체인은 346-3 시험의 가짜 세계(`chain_world`)로 · 종단 파일은 실제 1차 계획선. +""" + +from __future__ import annotations + +import asyncio +import json +import sys +from pathlib import Path +from uuid import uuid4 + +ROOT = Path(__file__).resolve().parents[2] +TESTER = Path(__file__).resolve().parent +for path in (ROOT, TESTER): + if str(path) not in sys.path: + sys.path.insert(0, str(path)) + +from test_346_3_route_save import ( # noqa: E402, F401 — 가짜 세계 fixture + NEW, + _async, + _longitudinal, + _run, + api_world, + chain_world, +) + +import B05_Profile.B05_Profile_Repository as b05_repo # 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_Router as b06_router # noqa: E402 +from B05_Profile.B05_Profile_Route_Save_Parts import vertices_of # noqa: E402 + +LONG = "B06_Section/longitudinal/longitudinal.json" + + +def _real_profile_world(world, monkeypatch): + """옛 종단 = 손 편집 · L 15 · 새 종횡단 = 실제 1차 계획선 — `_carry_profile` 은 진짜.""" + total = float(vertices_of(NEW)[-1].chainage_m) + old = _longitudinal(total) + old["profile_alignment"]["edits"] = {"station_offsets": {"40.000": 0.7}, "curve_radii": {}} + old["profile_alignment"]["policy"]["default_curve_length_m"] = 15.0 + path = world.root / LONG + path.write_text(json.dumps(old), "utf-8") + fresh = _longitudinal(total) + + async def regenerate(*args, **kwargs): + world.order.append("sections") + path.write_text(json.dumps(fresh), "utf-8") + + monkeypatch.setattr(b06_router, "regenerate_route_sections", regenerate) + monkeypatch.setattr(save, "_carry_profile", _CARRY) + monkeypatch.setattr(b05_repo, "update_longitudinal_grade_summary", _async(None)) + return path, fresh["profile_alignment"] + + +_CARRY = save._carry_profile + + +def test_적용_끔이면_계획고_안_옮김(chain_world): + result = _run(chain_world, initial=True) + assert "error" not in result, result + assert "profile" not in chain_world.order # 옛 코드 = 같은 노선 옛 꼭짓점 통째 옮김 + + +def test_적용_같은_노선_끔_새_1차_계획선(chain_world, monkeypatch): + path, fresh = _real_profile_world(chain_world, monkeypatch) + _run(chain_world, initial=True) + alignment = json.loads(path.read_text("utf-8"))["profile_alignment"] + assert alignment == fresh # 꼭짓점 · L(정책) · 편집 = 새 1차 계획선(438) 그대로 + assert not (alignment.get("edits") or {}).get("station_offsets") + + +def test_손_노선_저장은_계획고_옮김_그대로(chain_world, monkeypatch): + path, fresh = _real_profile_world(chain_world, monkeypatch) + _run(chain_world) + alignment = json.loads(path.read_text("utf-8"))["profile_alignment"] + assert alignment["edits"]["station_offsets"].get("40.000") == 0.7 # 앞 구간 손 편집 옮김 + + +def test_적용_켬_균형_셈_실패도_1차_계획선(chain_world, monkeypatch): + path, fresh = _real_profile_world(chain_world, monkeypatch) + _run(chain_world, profile_target=None, initial=True) # 균형 목표 None(셈 실패) + assert json.loads(path.read_text("utf-8"))["profile_alignment"] == fresh + + +def test_적용은_체인에_initial(api_world, monkeypatch): + seen: list = [] + + async def recompute(*args, **kwargs): + seen.append(args[6] if len(args) > 6 else kwargs.get("initial")) + return {"route_id": 9, "total_length_m": 10.0} + + monkeypatch.setattr(replan, "_recompute", recompute) + body = save.RouteSaveRequest(vertices=[{"x": 0, "y": 0}, {"x": 5, "y": 5}]) + asyncio.run(replan.replan_route(uuid4(), body)) + asyncio.run(replan.replan_route(uuid4(), body.model_copy(update={"as_initial": True}))) + assert seen == [False, True] + + +def test_횡단_손_입력_넘김_한_자리(): + # 사용자 답 전 = [적용]도 넘김(475-3) — 답 뒤 이 함수 한 줄 + assert save.carries_cross_inputs(True) is True and save.carries_cross_inputs(False) is True From b0e38de44c2df25e06b5403cc845633ef90df8e1 Mon Sep 17 00:00:00 2001 From: umsangdon Date: Mon, 19 Oct 2026 16:36:26 +0900 Subject: [PATCH 2/3] =?UTF-8?q?refactor(476-1=20=E2=91=A1):=20=EC=B8=A1?= =?UTF-8?q?=EC=A0=90=20=C2=B7=201=EC=B0=A8=20=EA=B3=84=ED=9A=8D=EC=84=A0?= =?UTF-8?q?=20=EC=85=88=EA=B3=BC=20=ED=8C=8C=EC=9D=BC=20=EC=93=B0=EA=B8=B0?= =?UTF-8?q?=20=EB=82=98=EB=88=94(=EB=8F=99=EC=9E=91=20=EA=B0=99=EC=9D=8C)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - Engine_Sections `build_sections`(측점 틀 · 지반 · 438 1차 계획선 · 포장 제안 · 파일 안 씀) · run_section_generation = 읽기 → 셈 → 쓰기 - 횡단 이름 규칙 → 새 `B05_Profile_Engine_Sections_Names`(700줄 · 다시 내보냄) - resolve_extra_stations 사본 칸(structures · plus) · 따라감 셈 몫 `followed_structures` · `followed_plus_rows` · `followed_pipes` · `section_seed` · `seat_base_pvi` 뗌 - 체인 · 추천 초기 종단이 같은 조각을 부를 자리 Co-Authored-By: Claude Opus 5.5 --- B05_Profile/B05_Profile_Engine_Sections.py | 181 ++++++++---------- .../B05_Profile_Engine_Sections_Names.py | 82 ++++++++ B05_Profile/B05_Profile_Plus_Stations.py | 13 +- B05_Profile/B05_Profile_Route_Save.py | 51 +++-- .../B05_Profile_Router_RouteInitial.py | 14 +- B05_Profile/B05_Profile_Structures_Router.py | 31 ++- 6 files changed, 230 insertions(+), 142 deletions(-) create mode 100644 B05_Profile/B05_Profile_Engine_Sections_Names.py diff --git a/B05_Profile/B05_Profile_Engine_Sections.py b/B05_Profile/B05_Profile_Engine_Sections.py index bbaf3707f..108484692 100644 --- a/B05_Profile/B05_Profile_Engine_Sections.py +++ b/B05_Profile/B05_Profile_Engine_Sections.py @@ -23,6 +23,17 @@ from B05_Profile.B05_Profile_Engine_Sections_Core import ( SectionGenerationOptions, generate_sections, ) +from B05_Profile.B05_Profile_Engine_Sections_Names import ( # noqa: F401 — 부르는 곳이 여기서 씀 + CROSS_KEY_PATTERN, + cross_filename, + cross_files_in_order, + cross_key, + cross_station, + find_cross_file, + parse_cross_key, + prune_stale_cross_files, + valid_cross_names, +) from B05_Profile.B05_Profile_Structures_Repository import load_structures from B05_Profile.B05_Profile_Structures_Schema import ( structure_type_map, @@ -118,82 +129,6 @@ def _read_design_curve_table(project_root: Path) -> dict[str, Any]: return data if isinstance(data, dict) else {} -# ── 횡단 이름 규칙(한 벌) — 파일 이름 · B07 도면 id · 찾기가 모두 이것을 씀 ──────────── -#: 이름 속 측점 글 — 정수 m `00120` · 소수 측점 `00119.80` -CROSS_KEY_PATTERN = r"(\d+(?:\.\d{2})?)" - - -def cross_station(chainage_m: float) -> int | float: - """측점 → 이름 열쇠 값 — 정수 m(0.005m 안)면 int(옛 이름 · 키 그대로) · 아니면 소수 2자리.""" - value = round(float(chainage_m), 2) - whole = round(value) - return int(whole) if abs(value - whole) < 0.005 else value - - -def cross_key(chainage_m: float) -> str: - """이름 속 측점 글 — `00120`(정수 · 옛 이름 그대로) · `00119.80`(소수 측점).""" - station = cross_station(chainage_m) - return f"{station:05d}" if isinstance(station, int) else f"{station:08.2f}" - - -def parse_cross_key(text: str) -> int | float: - """`cross_key` 되읽기.""" - return cross_station(float(text)) - - -def cross_filename(chainage_m: float) -> str: - """측점 chainage에 대응하는 횡단면 파일명(단일 규칙).""" - return f"cross_{cross_key(chainage_m)}m.json" - - -def find_cross_file(cross_dir: Path, chainage_m: float) -> Path: - """읽기용 — 지금 이름이 없으면 옛 정수 m 이름(소수 측점이 정수로 붙던 옛 프로젝트).""" - path = cross_dir / cross_filename(chainage_m) - legacy = cross_dir / f"cross_{int(round(float(chainage_m))):05d}m.json" - return path if path.exists() or not legacy.exists() else legacy - - -def valid_cross_names(cross_dir: Path, stations: list[Any]) -> set[str]: - """stations 에 맞는 cross_*.json 파일명 집합 — 읽기 전용(지우지 않음 · GET 이 씀). - - 측점 정보가 비면 빈 집합(호출부는 빈 집합이면 필터를 생략한다). - """ - # 옛 프로젝트는 소수 측점 파일이 정수 m 이름이라 찾기 규칙(find_cross_file)으로 가림 - return { - find_cross_file(cross_dir, station["chainage_m"]).name - for station in stations - if isinstance(station, dict) and isinstance(station.get("chainage_m"), (int, float)) - } - - -def cross_files_in_order(cross_dir: Path, stations: list[Any]) -> list[Path]: - """잔재를 걸러 읽을 cross_*.json — **측점 차례**(이름 차례면 `01080.16` 이 `01080` 앞).""" - valid = valid_cross_names(cross_dir, stations) - - def chainage(path: Path) -> tuple[float, str]: - match = re.fullmatch(rf"cross_{CROSS_KEY_PATTERN}m", path.stem) - return (float(match.group(1)) if match else float("inf"), path.name) - - files = [path for path in cross_dir.glob("cross_*.json") if not valid or path.name in valid] - return sorted(files, key=chainage) - - -def prune_stale_cross_files(cross_dir: Path, stations: list[Any]) -> set[str]: - """stations에 없는 잔재 cross_*.json을 삭제하고 유효 파일명 집합을 반환한다. - - 쓰는 쪽(측점 만들기 · 측점 목록 고침)만 부른다 — 조회(GET)는 `valid_cross_names`. - 측점 정보가 비어 있으면 오삭제를 피하기 위해 아무것도 지우지 않고 - 빈 집합을 반환한다(호출부는 빈 집합이면 필터를 생략한다). - """ - valid = valid_cross_names(cross_dir, stations) - if not valid: - return valid - for path in cross_dir.glob("cross_*.json"): - if path.name not in valid: - path.unlink(missing_ok=True) - return valid - - def _cross_summary(cross_section: dict[str, Any]) -> dict[str, Any]: """횡단면 상세에서 DB data 컬럼에 저장할 요약을 만든다.""" samples = cross_section.get("samples", []) @@ -297,7 +232,12 @@ def relabel_pipe_stations( def resolve_extra_stations( - project_root: Path, pipes: list[PipePoint], xy: Sequence[Sequence[float]] | None = None + project_root: Path, + pipes: list[PipePoint], + xy: Sequence[Sequence[float]] | None = None, + *, + structures: list[Any] | None = None, + plus: list[float] | None = None, ) -> tuple[tuple[float, str], ...]: """구조물(비정규) 측점의 **단일 공급자** — 관 정본 + 구조물 정본 A군에서 파생한다. @@ -314,12 +254,14 @@ def resolve_extra_stations( 거울 시험 = `resources/tester/test_322_3_plan_mirror.py` · `test_272_1_station_list.py`. 추가 측점(368-2) = `plus_stations.json` · 이름 "". `xy` = 지금 노선 — 노선 밖(시점 앞 · 종점 뒤 · `route_outside`) 관 · 구조물은 뺌(459). + `structures` · `plus` = 파일 대신 따라간 사본(476 추천 초기 종단 · 파일 안 씀). """ from B05_Profile.B05_Profile_Plus_Stations import plus_chainages from common_util.common_util_route_follow import route_outside try: - structures = load_structures(str(project_root))[1] + if structures is None: + structures = load_structures(str(project_root))[1] except Exception: logger.exception("B05: 구조물 정본을 읽지 못했습니다 (관 유래 측점만 씁니다)") structures = [] @@ -332,7 +274,9 @@ def resolve_extra_stations( ] return tuple( (float(station["chainage_m"]), str(station["title"])) - for station in plan_stations(pipes, structures, plus_chainages(project_root)) + for station in plan_stations( + pipes, structures, plus_chainages(project_root) if plus is None else plus + ) ) @@ -450,6 +394,55 @@ def _annotate_pavement_suggestions( station["pavement_grade_limit_pct"] = round(unpaved_limit, 2) +def build_sections( + polyline: list[list[float]], + sampler: Any, + pipes: list[PipePoint], + extras: tuple[tuple[float, str], ...], + *, + options: SectionGenerationOptions | None, + grade_options: GradeDesignOptions | None, + source_snapshot: dict[str, Any], + crs: str | None, + design_curves: list[dict[str, Any]], + station_snap: bool, + station_snap_step: float | None, +) -> tuple[dict[str, Any], dict[str, Any] | None]: + """측점 틀 · 지반 · 1차 계획선(438) · 포장 제안 — 파일 안 씀 · (결과, 계획선 요약). + + 종횡단 생성(`run_section_generation`)과 추천 후보 초기 종단(476 · `..._RouteInitial_Initial`) + 이 같은 셈 한 벌.""" + result = generate_sections( + polyline, + sampler, + replace(options or SectionGenerationOptions(), extra_stations=extras), + source_snapshot=source_snapshot, + crs=crs, + design_curves=design_curves, + station_snap=station_snap, + station_snap_step=station_snap_step, + ) + grade_summary = _append_design_profiles( + result["longitudinal"], + grade_options, + (result.get("options") or {}).get("station_interval_m"), + # 1차 계획선의 변화점 = 배수유역도가 확정한 배관 배치 측점(측점 생성과 같은 목록). + pipe_chainages=[pipe.chainage_m for pipe in pipes], + # 시설 제원 최소 여유(관경+토피 등)로 계획선 들어올림 — 사용자 스위치(기본 해제 · 09-01). + pipe_clearances=( + dict(pipe_anchor_clearances(pipes)) + if grade_options and grade_options.enforce_pipe_clearance + else None + ), + ) + # 계획선 경사 기반 포장 제안(법정 상한 초과 측점) — 비치명적. + try: + _annotate_pavement_suggestions(result["longitudinal"], grade_options) + except Exception: + logger.exception("B05 포장 제안 판정 실패 (종횡단은 유지)") + return result, grade_summary + + def run_section_generation( project_root: Path, route_data_path: str, @@ -487,10 +480,13 @@ def run_section_generation( extras = () logger.info("B05 종횡단: 구조물 측점 %d건 파생(관 %d건)", len(extras), len(pipes)) - result = generate_sections( + result, grade_summary = build_sections( polyline, sampler, - replace(options or SectionGenerationOptions(), extra_stations=extras), + pipes, + extras, + options=options, + grade_options=grade_options, source_snapshot={"filter": filter_key, "method": method, "smooth": smooth}, crs=crs, # 곡선부 확폭의 정본 — 설계 곡선표(시·종점·반지름)를 그대로 쓴다. 없으면 빈 목록이라 @@ -505,29 +501,6 @@ def run_section_generation( cross_dir = stage_root / "cross_sections" long_dir.mkdir(parents=True, exist_ok=True) cross_dir.mkdir(parents=True, exist_ok=True) - - # 종단면 저장 (계획선은 저장 직전에 종단 데이터에 붙인다) - grade_summary = _append_design_profiles( - result["longitudinal"], - grade_options, - (result.get("options") or {}).get("station_interval_m"), - # 1차 계획선의 변화점 = 배수유역도가 확정한 배관 배치 측점(측점 생성과 같은 목록). - pipe_chainages=[pipe.chainage_m for pipe in pipes], - # 시설 제원이 요구하는 최소 여유(관경+토피 등) — 계획선이 그만큼 들려야 관·구체가 - # 들어갈 자리가 생긴다(2026-08-23 사용자 지시). 다만 이 들어올림이 배수 자리마다 - # 계획고를 잡아당겨 제어가 어려워, 적용 여부를 사용자 스위치로 뺐다(2026-09-01 - # 사용자 지시 — 기본 해제). 산식은 그대로이고 켜고 끄는 것만 바뀐다. - pipe_clearances=( - dict(pipe_anchor_clearances(pipes)) - if grade_options and grade_options.enforce_pipe_clearance - else None - ), - ) - # 계획선 경사 기반 포장 제안(법정 상한 초과 측점) — 비치명적. - try: - _annotate_pavement_suggestions(result["longitudinal"], grade_options) - except Exception: - logger.exception("B05 포장 제안 판정 실패 (종횡단은 유지)") long_file = long_dir / "longitudinal.json" atomic_write_json(long_file, result["longitudinal"]) long_summary = { diff --git a/B05_Profile/B05_Profile_Engine_Sections_Names.py b/B05_Profile/B05_Profile_Engine_Sections_Names.py new file mode 100644 index 000000000..60605bb82 --- /dev/null +++ b/B05_Profile/B05_Profile_Engine_Sections_Names.py @@ -0,0 +1,82 @@ +"""횡단 이름 규칙 한 벌 — 파일 이름 · B07 도면 id · 찾기 · 잔재 거름. + +`B05_Profile_Engine_Sections` 에서 나눔(700줄) · 그 모듈이 다시 내보냄.""" + +import re +from pathlib import Path +from typing import Any + +# ── 횡단 이름 규칙(한 벌) — 파일 이름 · B07 도면 id · 찾기가 모두 이것을 씀 ──────────── +#: 이름 속 측점 글 — 정수 m `00120` · 소수 측점 `00119.80` +CROSS_KEY_PATTERN = r"(\d+(?:\.\d{2})?)" + + +def cross_station(chainage_m: float) -> int | float: + """측점 → 이름 열쇠 값 — 정수 m(0.005m 안)면 int(옛 이름 · 키 그대로) · 아니면 소수 2자리.""" + value = round(float(chainage_m), 2) + whole = round(value) + return int(whole) if abs(value - whole) < 0.005 else value + + +def cross_key(chainage_m: float) -> str: + """이름 속 측점 글 — `00120`(정수 · 옛 이름 그대로) · `00119.80`(소수 측점).""" + station = cross_station(chainage_m) + return f"{station:05d}" if isinstance(station, int) else f"{station:08.2f}" + + +def parse_cross_key(text: str) -> int | float: + """`cross_key` 되읽기.""" + return cross_station(float(text)) + + +def cross_filename(chainage_m: float) -> str: + """측점 chainage에 대응하는 횡단면 파일명(단일 규칙).""" + return f"cross_{cross_key(chainage_m)}m.json" + + +def find_cross_file(cross_dir: Path, chainage_m: float) -> Path: + """읽기용 — 지금 이름이 없으면 옛 정수 m 이름(소수 측점이 정수로 붙던 옛 프로젝트).""" + path = cross_dir / cross_filename(chainage_m) + legacy = cross_dir / f"cross_{int(round(float(chainage_m))):05d}m.json" + return path if path.exists() or not legacy.exists() else legacy + + +def valid_cross_names(cross_dir: Path, stations: list[Any]) -> set[str]: + """stations 에 맞는 cross_*.json 파일명 집합 — 읽기 전용(지우지 않음 · GET 이 씀). + + 측점 정보가 비면 빈 집합(호출부는 빈 집합이면 필터를 생략한다). + """ + # 옛 프로젝트는 소수 측점 파일이 정수 m 이름이라 찾기 규칙(find_cross_file)으로 가림 + return { + find_cross_file(cross_dir, station["chainage_m"]).name + for station in stations + if isinstance(station, dict) and isinstance(station.get("chainage_m"), (int, float)) + } + + +def cross_files_in_order(cross_dir: Path, stations: list[Any]) -> list[Path]: + """잔재를 걸러 읽을 cross_*.json — **측점 차례**(이름 차례면 `01080.16` 이 `01080` 앞).""" + valid = valid_cross_names(cross_dir, stations) + + def chainage(path: Path) -> tuple[float, str]: + match = re.fullmatch(rf"cross_{CROSS_KEY_PATTERN}m", path.stem) + return (float(match.group(1)) if match else float("inf"), path.name) + + files = [path for path in cross_dir.glob("cross_*.json") if not valid or path.name in valid] + return sorted(files, key=chainage) + + +def prune_stale_cross_files(cross_dir: Path, stations: list[Any]) -> set[str]: + """stations에 없는 잔재 cross_*.json을 삭제하고 유효 파일명 집합을 반환한다. + + 쓰는 쪽(측점 만들기 · 측점 목록 고침)만 부른다 — 조회(GET)는 `valid_cross_names`. + 측점 정보가 비어 있으면 오삭제를 피하기 위해 아무것도 지우지 않고 + 빈 집합을 반환한다(호출부는 빈 집합이면 필터를 생략한다). + """ + valid = valid_cross_names(cross_dir, stations) + if not valid: + return valid + for path in cross_dir.glob("cross_*.json"): + if path.name not in valid: + path.unlink(missing_ok=True) + return valid diff --git a/B05_Profile/B05_Profile_Plus_Stations.py b/B05_Profile/B05_Profile_Plus_Stations.py index 7f21321a2..911c93462 100644 --- a/B05_Profile/B05_Profile_Plus_Stations.py +++ b/B05_Profile/B05_Profile_Plus_Stations.py @@ -75,6 +75,14 @@ def follow_plus_stations(project_root: str | Path, new_xy: Sequence[Sequence[flo rows = load_plus_stations(project_root) if not rows: return 0 + followed = followed_plus_rows(rows, new_xy) + if followed["moved"]: + atomic_write_json(str(plus_stations_path(project_root)), {"stations": followed["rows"]}) + return int(followed["moved"]) + + +def followed_plus_rows(rows: list[dict[str, Any]], new_xy: Sequence[Sequence[float]]) -> dict: + """추가 측점을 새 노선에 투영(저장 안 함) → `{rows, moved}` — 체인 · 추천 초기 종단(476).""" results = run_calc( ["facility.plus_follow"], { @@ -89,7 +97,4 @@ def follow_plus_stations(project_root: str | Path, new_xy: Sequence[Sequence[flo ) if results is None: raise RuntimeError("추가 측점 따라감 실패 — 계산 Node 입구(facility.plus_follow)") - followed = results["facility.plus_follow"] - if followed["moved"]: - atomic_write_json(str(plus_stations_path(project_root)), {"stations": followed["rows"]}) - return int(followed["moved"]) + return results["facility.plus_follow"] diff --git a/B05_Profile/B05_Profile_Route_Save.py b/B05_Profile/B05_Profile_Route_Save.py index d309e903f..fda51025b 100644 --- a/B05_Profile/B05_Profile_Route_Save.py +++ b/B05_Profile/B05_Profile_Route_Save.py @@ -22,7 +22,7 @@ API — `POST /api/projects/{id}/route/replan`(`B05_Profile_Router_Replan.replan (`B05_Profile_Route_Save_Parts`) · 배수유역 저장분은 노선 지문이 달라 「낡음」(B04 `_resolve`). - 계획고 = 바뀐 자리만 새 지반 첫 계획선 · 같은 자리(옆 1 m 안) 사용자 계획고 지킴 (`B05_Profile_Engine_Grade_Carry` · 옛 · 새 **계획 폴리라인** — 화면 planline 과 같은 선) · - 추천 [적용](`as_initial`) = 옮기지 않음 — 균형 종단(`apply_profile_target`) 또는 1차 계획선(438). + 추천 [적용](`as_initial`) = 옮기지 않음 — 균형 종단(`apply_profile_target`) · 1차 계획선(438). - 초기값은 여기서 안 뜸 — 부르는 쪽이 `as_initial` 일 때만(자동설계 체인 직후 + 추천 [적용]). """ @@ -165,6 +165,31 @@ def _xy(project_root: Path, route_data_path: str | None) -> list[list[float]]: return [] +def followed_pipes( + old_pipes: list, old_xy: list[list[float]], points: list[dict[str, float]], streams: list +) -> tuple[list, int, list]: + """관 따라감 + 새 세류선 교차점 기본 관(저장 안 함) → (짝, 교차 수, 새 노선 정점) — 체인 · + 추천 초기 종단(476) 한 벌.""" + from B04_PreProcess.B04_PreProcess_Engine_Watershed_Analyze import _base_pipes + from common_util.common_util_drainage_pipes import fill_pipe_coordinates + + vertices = vertices_of([(p["x"], p["y"]) for p in points]) + fill_pipe_coordinates(old_pipes, vertices_of(old_xy)) + pairs = follow_pipes(old_pipes, vertices) + total = float(vertices[-1].chainage_m) + crossing = add_crossing_pipes(pairs, _base_pipes(vertices, streams), old_xy, total) + return pairs, crossing, vertices + + +def section_seed(request: Any, old_section_options: dict[str, Any] | None) -> dict[str, Any]: + """체인 종횡단 생성 옵션(건너뛴 첫 생성이 남겼을 값) — 체인 · 추천 초기 종단(476) 한 벌.""" + from B05_Profile.B05_Profile_Router import _section_options + + seed = asdict(_section_options(request, old_section_options)) + seed.pop("extra_stations", None) + return seed + + async def _place_pipes( project_id: UUID, stored_path: str, @@ -178,22 +203,13 @@ async def _place_pipes( 노선은 범위를 안 잘라 누가거리가 갈림 · 세류선만 거기서. 좌표 없는 옛 관은 옛 노선 자리로 먼저 채움(459). """ - from B04_PreProcess.B04_PreProcess_Engine_Watershed_Analyze import _base_pipes from B04_PreProcess.B04_PreProcess_Router_Watershed import _prepare - from common_util.common_util_drainage_pipes import ( - fill_pipe_coordinates, - route_signature, - save_pipe_points, - ) + from common_util.common_util_drainage_pipes import route_signature, save_pipe_points from common_util.common_util_route_follow import route_outside prepared = await _prepare(project_id) streams = [] if isinstance(prepared, JSONResponse) else prepared["streams"] # 도엽 없음 - vertices = vertices_of([(p["x"], p["y"]) for p in points]) - fill_pipe_coordinates(old_pipes, vertices_of(old_xy)) - pairs = follow_pipes(old_pipes, vertices) - total = float(vertices[-1].chainage_m) - crossing = add_crossing_pipes(pairs, _base_pipes(vertices, streams), old_xy, total) + pairs, crossing, vertices = followed_pipes(old_pipes, old_xy, points, streams) await asyncio.to_thread( save_pipe_points, stored_path, @@ -259,7 +275,7 @@ async def _carry_profile( def carries_cross_inputs(initial: bool) -> bool: - """횡단 손 입력을 새 측점(같은 자리)에 넘기나 — 추천 [적용]도 넘김(475-3 · 사용자 답 대기 476).""" + """횡단 손 입력을 새 측점(같은 자리)에 넘기나 — [적용]도 넘김(475-3 · 사용자 답 대기 476).""" return True @@ -376,11 +392,7 @@ async def run_route_save_chain( from B03_FileInput.B03_FileInput_Service_Chain import _log_steps, redesign_solve_request from B05_Profile.B05_Profile_Engine_Sections import _STAGE_SUBDIR from B05_Profile.B05_Profile_Plus_Stations import follow_plus_stations, plus_stations_path - from B05_Profile.B05_Profile_Router import ( - _section_options, - confirm_latest_route, - solve_route_record, - ) + from B05_Profile.B05_Profile_Router import confirm_latest_route, solve_route_record from B05_Profile.B05_Profile_Structures_Repository import load_structures, structures_file_path from B05_Profile.B05_Profile_Structures_Router import follow_structures_to_route from B06_Section.B06_Section_Repository import ( @@ -456,8 +468,7 @@ async def run_route_save_chain( pairs, pipes = await _place_pipes(project_id, stored_path, old_pipes, old_xy, points) marks.append(("관 따라감 · 교차점", time.perf_counter())) - seed = asdict(_section_options(request, old_section_options)) - seed.pop("extra_stations", None) + seed = section_seed(request, old_section_options) made = await regenerate_route_sections( project_id, new_route_id, float(seed["cross_half_width_m"]), stored_options=seed ) diff --git a/B05_Profile/B05_Profile_Router_RouteInitial.py b/B05_Profile/B05_Profile_Router_RouteInitial.py index c2936fbd2..19579feec 100644 --- a/B05_Profile/B05_Profile_Router_RouteInitial.py +++ b/B05_Profile/B05_Profile_Router_RouteInitial.py @@ -355,15 +355,22 @@ async def refresh_expected_route(project_id: UUID) -> None: def _set_base_pvi(path: Path, stations: list[float], plan: list[float]) -> int: """종단 정본의 1차 변화점(`base_pvi`) 계획고를 `stations` · `plan` 직선 보간으로 바꿔 쓴다.""" - import json + from common_util.common_util_json import atomic_write_json + longitudinal = json.loads(path.read_text(encoding="utf-8")) + count = seat_base_pvi(longitudinal, stations, plan) + atomic_write_json(path, longitudinal) + return count + + +def seat_base_pvi(longitudinal: dict[str, Any], stations: list[float], plan: list[float]) -> int: + """1차 변화점 계획고를 균형 종단에 얹음(제자리 · 파일 안 씀) — [적용] 체인 · 추천 초기 종단(476) + 한 벌.""" import numpy as np from B05_Profile.B05_Profile_Engine_Grade_Alignment import AlignmentPolicy from B05_Profile.B05_Profile_Engine_RouteInitial_Profile import held_pass_through - from common_util.common_util_json import atomic_write_json - longitudinal = json.loads(path.read_text(encoding="utf-8")) alignment = longitudinal["profile_alignment"] base = alignment["base_pvi"] base_s = np.array([float(item["chainage_m"]) for item in base]) @@ -374,7 +381,6 @@ def _set_base_pvi(path: Path, stations: list[float], plan: list[float]) -> int: ) for item, value in zip(base, base_z, strict=True): item["elevation_m"] = round(float(value), 6) # 직선 판정(426) 1e-5 m 안 - atomic_write_json(path, longitudinal) return len(base) diff --git a/B05_Profile/B05_Profile_Structures_Router.py b/B05_Profile/B05_Profile_Structures_Router.py index 9b1059aa3..f639d071b 100644 --- a/B05_Profile/B05_Profile_Structures_Router.py +++ b/B05_Profile/B05_Profile_Structures_Router.py @@ -121,6 +121,26 @@ async def _route_xy(project_id: UUID, root: str) -> list[list[float]] | None: return None +def followed_structures( + project_root: Path, + items: list[Any], + old_route_data_path: str | None, + new_xy: list[dict[str, float]], +) -> list[Any]: + """구조물을 새 노선 땅 위 자리로 옮김(제자리 · 저장 안 함) — 노선 저장 체인 · 추천 초기 + 종단(476) 한 벌. 좌표 없는 옛 자료는 옛 노선으로 먼저 채움.""" + from B05_Profile.B05_Profile_Engine_Sections import _load_route_polyline + + if old_route_data_path: + try: + fill_structure_coordinates( + items, _load_route_polyline(project_root, old_route_data_path), only_missing=True + ) + except (OSError, ValueError): + logger.warning("B05 구조물 따라감: 옛 노선을 못 읽어 좌표 없는 구조물은 그대로") + return follow_structures(items, [(point["x"], point["y"]) for point in new_xy]) + + def follow_structures_to_route( project_root: Path, old_route_data_path: str | None, new_xy: list[dict[str, float]] ) -> dict[str, int]: @@ -129,21 +149,12 @@ def follow_structures_to_route( 좌표가 없는 옛 자료는 옛 노선(아직 안 바뀐 경로 파일)으로 먼저 채운다. 실패는 비치명 — 구조물은 옛 누가거리 그대로 남는다(지금까지와 같음). """ - from B05_Profile.B05_Profile_Engine_Sections import _load_route_polyline - root = str(project_root) revision, items = load_structures(root) if not items: return {"moved": 0, "far": 0} - if old_route_data_path: - try: - fill_structure_coordinates( - items, _load_route_polyline(project_root, old_route_data_path), only_missing=True - ) - except (OSError, ValueError): - logger.warning("B05 구조물 따라감: 옛 노선을 못 읽어 좌표 없는 구조물은 그대로") before = [(item.chainage_m, item.start_m, item.end_m) for item in items] - result = follow_structures(items, [(point["x"], point["y"]) for point in new_xy]) + result = followed_structures(project_root, items, old_route_data_path, new_xy) moved = sum( before[index] != (item.chainage_m, item.start_m, item.end_m) for index, item in enumerate(items) From aa87dcfd689e6b500a9d3e442e4fc7b0daa9e283 Mon Sep 17 00:00:00 2001 From: umsangdon Date: Mon, 19 Oct 2026 16:49:59 +0900 Subject: [PATCH 3/3] =?UTF-8?q?feat(476-1=20=E2=91=A2):=20=EC=B6=94?= =?UTF-8?q?=EC=B2=9C=20=ED=9B=84=EB=B3=B4=20=EC=B4=88=EA=B8=B0=20=EC=A2=85?= =?UTF-8?q?=EB=8B=A8=20=EB=AF=B8=EB=A6=AC=20=EC=85=88=20=E2=80=94=20?= =?UTF-8?q?=EC=9D=91=EB=8B=B5=20=EC=B9=B8=20initial=5Fprofile=20=C2=B7=20?= =?UTF-8?q?=EC=A0=80=EC=9E=A5=EB=B3=B8?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - 새 `B05_Profile_Engine_InitialProfile` — 후보 노드 → 화면 펴기 짝(`draft_nodes` ↔ TS flattenServerPlan) → 노선 짜기(csv 4자리) → 관 · 구조물 · 추가 측점 따라감(새 세류선 교차 관 포함) → build_sections(438 1차 계획선) → 켬 = 1 m 균형 종단 얹기(seat_base_pvi) - 새 `B05_Profile_Router_RouteInitial_Initial.initial_profile_of` — [적용] 체인과 같은 입력 모으기 · 지문(후보 · 지형 · 페이지 설정 · 시설 파일 · 측점 옵션 · 균형 인자 · 코드) · 저장본 `route_recommend/initial/<키>.json` · 실패 = {error} - compute_initial 응답에 `initial_profile`(stations_m · ground_m · plan_m · line · pvi_count · curve_count · straightened · max_grade_pct · curve_length_m · warnings · with_profile · seconds) - 시험 test_476_1_recommend_initial — 미리 셈 ↔ 체인 파일 길 꼭짓점 차 0 mm(끔 · 노선 바뀜 + 옹벽 + 추가 측점 · 켬) · 지문 · 응답 칸 - 위키 calculations N2 · rules 줄 Co-Authored-By: Claude Opus 5.5 --- .../B05_Profile_Engine_InitialProfile.py | 263 +++++++++++ .../B05_Profile_Router_RouteInitial.py | 8 + ...B05_Profile_Router_RouteInitial_Initial.py | 185 ++++++++ .../logic/B05_Profile_UI_RouteEdit_Edits.ts | 3 + .../wiki/concepts/design_page/calculations.md | 1 + docs/wiki/concepts/design_page/rules.md | 2 +- resources/tester/helper_476_1_flatten.cjs | 39 ++ .../tester/test_476_1_recommend_initial.py | 409 ++++++++++++++++++ 8 files changed, 909 insertions(+), 1 deletion(-) create mode 100644 B05_Profile/B05_Profile_Engine_InitialProfile.py create mode 100644 B05_Profile/B05_Profile_Router_RouteInitial_Initial.py create mode 100644 resources/tester/helper_476_1_flatten.cjs create mode 100644 resources/tester/test_476_1_recommend_initial.py diff --git a/B05_Profile/B05_Profile_Engine_InitialProfile.py b/B05_Profile/B05_Profile_Engine_InitialProfile.py new file mode 100644 index 000000000..ae629a60f --- /dev/null +++ b/B05_Profile/B05_Profile_Engine_InitialProfile.py @@ -0,0 +1,263 @@ +"""추천 후보 초기 종단(476 · 요구 276) — [적용]하면 작업본 · 초기값에 설 종단을 미리 셈. + +[적용] 체인(`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`) → + 「종단 반영」 켬이면 1 m 균형 종단을 꼭짓점에 얹음(`seat_base_pvi` → `rebuild_alignment_profile`). +[적용]은 이 값을 받아 적지 않음 — 서버가 체인에서 같은 조각으로 다시 셈(CLAUDE.md 5). + +저장본 = `B05_Profile/route_recommend/initial/<결과 키>.json` `{fingerprint, profile}` — 결과 정리 +(`*.json` glob)가 안 보는 하위 폴더 · 결과 짝 없는 파일은 쓸 때 지움 · 지문 다르면 다시 셈. +응답 칸 `initial_profile`(화면 476-2) = `{stations_m, ground_m, plan_m, line: [[s, z]…], pvi_count, +curve_count, straightened, max_grade_pct, curve_length_m, warnings, with_profile, seconds}` · 실패 = +`{error}`. +""" + +from __future__ import annotations + +import hashlib +import json +import math +from pathlib import Path +from typing import Any + +from common_util.common_util_json import atomic_write_json + +INITIAL_DIR = "B05_Profile/route_recommend/initial" +# 초기 종단을 낳는 코드(저장소 뿌리 기준) — 바뀌면 저장본 다시 셈 +SOURCE_GLOBS = ( + "B05_Profile/B05_Profile_Engine_InitialProfile.py", + "B05_Profile/B05_Profile_Engine_Grade*.py", + "B05_Profile/B05_Profile_Engine_Sections*.py", + "B05_Profile/B05_Profile_Engine_RouteInitial_Profile.py", + "B05_Profile/B05_Profile_Route_Save_Parts.py", + "B05_Profile/B05_Profile_Plus_Stations.py", + "common_util/common_util_route_polyline.py", + "common_util/common_util_route_follow.py", + "common_util/common_util_structure_station.py", + "common_util/common_util_drainage_pipes.py", + "config/config_system_design.py", +) +# 직선 위 꼭짓점(438 직선화) 판정 — 앞뒤 기울기 차(비율) +_STRAIGHT_EPS = 1e-5 + + +def draft_nodes( + nodes: list[dict[str, Any]], curves: list[dict[str, Any]] +) -> tuple[list[tuple[float, float]], list[bool], list[float | None]]: + """후보 노드 · 곡선 → [적용] 초안(노드 · 곡선 켬 · R) — 화면 `flattenServerPlan`(편집 없음) · + `snapshotOf` · `routeDraftOf` 와 같은 꼴(곡선 = 교각점 하나 · R 0.01 m 반올림 = Math.round).""" + if not curves: + return ( + [(float(n["x"]), float(n["y"])) for n in nodes], + [n.get("curve") is not False for n in nodes], + [n.get("radius_m") for n in nodes], + ) + starts = {int(c["node_first"]): c for c in curves} + inside = {i for c in curves for i in range(int(c["node_first"]) + 1, int(c["node_last"]) + 1)} + points: list[tuple[float, float]] = [] + flags: list[bool] = [] + radii: list[float | None] = [] + for index, node in enumerate(nodes): + if index in inside: + continue + curve = starts.get(index) + points.append( + (float(curve["apex"][0]), float(curve["apex"][1])) + if curve + else (float(node["x"]), float(node["y"])) + ) + flags.append(curve is not None) + radii.append(math.floor(float(curve["radius_m"]) * 100 + 0.5) / 100 if curve else None) + return points, flags, radii + + +def candidate_line( + nodes: list[dict[str, Any]], curves: list[dict[str, Any]], radius_m: float +) -> tuple[list[tuple[float, float]], list[dict[str, Any]]]: + """[적용]이 쓸 노선(csv 4자리 = 풀린 노선 geojson) · 곡선표 — `_write_planned_polyline` 짜기.""" + from common_util.common_util_route_polyline import build_planned_polyline + + points, flags, radii = draft_nodes(nodes, curves) + outline = build_planned_polyline( + points, min_radius_m=radius_m, simplify=False, curve_flags=flags, radii=radii + ) + line = [(round(x, 4), round(y, 4)) for x, y in outline.vertices] + 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 + 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, + parse_pipe_points, + ) + + 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) + 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 + + +def initial_longitudinal( + project_root: Path, + line: list[tuple[float, float]], + curves: list[dict[str, Any]], + sampler: Any, + facilities: tuple[list, list, list[float] | None], + *, + options: Any, + grade_options: Any, + source_snapshot: dict[str, Any], + crs: str | None, + target: tuple[list[float], list[float]] | None, +) -> dict[str, Any]: + """[적용] 체인이 남길 종단(측점 · 계획선) — 측점 생성 · 균형 얹기와 같은 조각.""" + from B05_Profile.B05_Profile_Engine_Grade_Profile import rebuild_alignment_profile + from B05_Profile.B05_Profile_Engine_Sections import build_sections, resolve_extra_stations + from B05_Profile.B05_Profile_Router_RouteInitial import seat_base_pvi + + polyline = [[x, y, 0.0] for x, y in line] + pipes, structures, plus = facilities + extras = resolve_extra_stations(project_root, pipes, polyline, structures=structures, plus=plus) + result, _summary = build_sections( + polyline, + sampler, + pipes, + extras, + options=options, + grade_options=grade_options, + source_snapshot=source_snapshot, + crs=crs, + design_curves=curves, + station_snap=True, # [적용] = 스냅 켬 · 측점 No.(`saveRecommendedRoute`) + station_snap_step=None, + ) + longitudinal = result["longitudinal"] + if target is not None and longitudinal.get("profile_alignment"): + seat_base_pvi(longitudinal, *target) + alignment, profile = rebuild_alignment_profile(longitudinal, {}) + longitudinal["profile_alignment"] = alignment + longitudinal["design_profiles"][0] = profile + return longitudinal + + +def straightened_count(base: list[dict[str, Any]]) -> int: + """앞뒤 기울기가 같은 안쪽 꼭짓점 수 = 직선 위(438 직선화 · 종단곡선 없음).""" + s = [float(p["chainage_m"]) for p in base] + z = [float(p["elevation_m"]) for p in base] + grades = [(z[i + 1] - z[i]) / (s[i + 1] - s[i]) for i in range(len(s) - 1)] + return sum(abs(grades[i] - grades[i - 1]) < _STRAIGHT_EPS for i in range(1, len(grades))) + + +def profile_summary(longitudinal: dict[str, Any]) -> dict[str, Any]: + """응답 칸 `initial_profile` — 측점 · 지반 · 계획고 · 계획선(종단곡선 포함) · 수.""" + alignment = longitudinal.get("profile_alignment") + if not isinstance(alignment, dict) or not alignment.get("base_pvi"): + return {"error": "계획선을 세우지 못했습니다(지반고 종단)."} + summary = longitudinal["design_profiles"][0]["summary"] + rows = alignment.get("stations") or [] + return { + "stations_m": [round(float(r["chainage_m"]), 3) for r in rows], + "ground_m": [round(float(r["ground_elevation_m"]), 3) for r in rows], + "plan_m": [round(float(r["plan_elevation_m"]), 4) for r in rows], + "line": [ + [round(float(p["chainage_m"]), 3), round(float(p["elevation_m"]), 4)] + for p in alignment.get("samples") or [] + ], + "pvi_count": len(alignment["base_pvi"]), + "curve_count": int(summary.get("vertical_curve_count") or 0), + "straightened": straightened_count(alignment["base_pvi"]), + "max_grade_pct": round(float(summary.get("max_grade_pct") or 0.0), 2), + "curve_length_m": (alignment.get("policy") or {}).get("default_curve_length_m"), + "warnings": list(alignment.get("warnings") or []), + } + + +def source_fingerprint() -> str: + """초기 종단 코드 지문 — `SOURCE_GLOBS` 내용(줄끝 LF).""" + root = Path(__file__).resolve().parents[1] + digest = hashlib.blake2b(digest_size=8) + for path in sorted({p for pattern in SOURCE_GLOBS for p in root.glob(pattern)}): + digest.update(path.name.encode()) + digest.update(path.read_bytes().replace(b"\r\n", b"\n")) + return digest.hexdigest() + + +def file_marks(paths: list[Path]) -> list[str]: + """지문에 드는 파일 자국 — 내용 해시(작은 정본) · 없음 「-」.""" + marks = [] + for path in paths: + try: + marks.append(hashlib.blake2b(path.read_bytes(), digest_size=8).hexdigest()) + except OSError: + marks.append("-") + return marks + + +def stat_marks(paths: list[Path]) -> list[str]: + """큰 자료(지형 npz · 도엽) 자국 — 이름 · 크기 · 수정시각.""" + marks = [] + for path in sorted(paths): + try: + info = path.stat() + marks.append(f"{path.name}:{info.st_size}:{info.st_mtime_ns}") + except OSError: + marks.append(f"{path.name}:-") + return marks + + +def fingerprint(parts: dict[str, Any]) -> str: + text = json.dumps({**parts, "source": source_fingerprint()}, sort_keys=True, default=str) + return hashlib.blake2b(text.encode(), digest_size=16).hexdigest() + + +def _path(project_root: Path, key: str) -> Path: + return Path(project_root) / INITIAL_DIR / f"{key}.json" + + +def read_initial(project_root: Path, key: str, mark: str) -> dict[str, Any] | None: + """저장본 — 없거나 지문 다르면 None.""" + try: + data = json.loads(_path(project_root, key).read_text(encoding="utf-8")) + except (OSError, ValueError): + return None + return ( + data.get("profile") if isinstance(data, dict) and data.get("fingerprint") == mark else None + ) + + +def write_initial(project_root: Path, key: str, mark: str, profile: dict[str, Any]) -> None: + """한 결과에 하나(덮어씀) · 결과 저장본 짝 없는 파일 지움.""" + path = _path(project_root, key) + atomic_write_json(path, {"fingerprint": mark, "profile": profile}) + results = path.parent.parent + for other in path.parent.glob("*.json"): + if not (results / other.name).exists(): + other.unlink(missing_ok=True) diff --git a/B05_Profile/B05_Profile_Router_RouteInitial.py b/B05_Profile/B05_Profile_Router_RouteInitial.py index 19579feec..bc894ebe2 100644 --- a/B05_Profile/B05_Profile_Router_RouteInitial.py +++ b/B05_Profile/B05_Profile_Router_RouteInitial.py @@ -66,6 +66,7 @@ async def compute_initial( request_fingerprint, write_stored, ) + from B05_Profile.B05_Profile_Router_RouteInitial_Initial import initial_profile_of from common_util.common_util_surface_confirmation import ( analysis_terrain_params, get_surface_confirmation_params, @@ -175,6 +176,13 @@ async def compute_initial( profile.update(cover_shortfall(profile, line, pipes)) except (FileNotFoundError, ValueError) as error: return JSONResponse(status_code=409, content={"status": "error", "message": str(error)}) + # [적용]하면 설 초기 종단(476) — 같은 조각으로 미리 셈 · 저장본 · 실패 = {error}(추천은 그대로) + result = { + **result, + "initial_profile": await initial_profile_of( # 476 + project_id, project_root, paths[1], result, request, selection, key + ), + } logger.info( "초기 계획노선 %s(%s): project_id=%s %.1fs 연장 %.0fm 이격 평균 %.1fm", request.mode, diff --git a/B05_Profile/B05_Profile_Router_RouteInitial_Initial.py b/B05_Profile/B05_Profile_Router_RouteInitial_Initial.py new file mode 100644 index 000000000..a18f11767 --- /dev/null +++ b/B05_Profile/B05_Profile_Router_RouteInitial_Initial.py @@ -0,0 +1,185 @@ +"""추천 후보 초기 종단 모으기(476) — `POST /route/initial` 결과에 `initial_profile` 을 붙임. + +입력 모으기(DB · 파일) · 지문 · 저장본 = 여기 · 셈 조각 = `B05_Profile_Engine_InitialProfile`(머리에 +흐름 · 응답 칸). `B05_Profile_Router_RouteInitial`(700줄)에서 나눔 — `compute_initial` 이 부름. +""" + +from __future__ import annotations + +import asyncio +import functools +import logging +import time +from pathlib import Path +from typing import Any +from uuid import UUID + +from fastapi.responses import JSONResponse + +from config.config_db import get_db_pool + +logger = logging.getLogger(__name__) + + +async def initial_profile_of( + project_id: UUID, + project_root: Path, + stored_path: str, + result: dict[str, Any], + request: Any, + selection: dict[str, Any], + key: str, +) -> dict[str, Any]: + """추천 후보 초기 종단(476 · `B05_Profile_Engine_InitialProfile` 머리) — [적용] 체인과 같은 + 입력 · 같은 조각 · 지문 같으면 저장본 꺼냄. 실패 = `{error}`.""" + from dataclasses import asdict + + import aiomysql + + from B03_FileInput.B03_FileInput_Service_Chain import redesign_solve_request + from B04_PreProcess.B04_PreProcess_Router_Watershed import STREAM_FILE, _prepare, _sheet_dir + 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_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, + ) + from B06_Section.B06_Section_Router_Regenerate_Options import ( + _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 + + started = time.perf_counter() + try: + radius_m, _, _ = await replan._plan_criteria(project_id) # [적용] 노선 짜기와 같은 R + line, curves = ip.candidate_line(result["nodes"], result.get("curves") or [], radius_m) + pool = get_db_pool() + async with pool.acquire() as connection: + latest = await get_latest_route(connection, project_id) + source = ( + await get_route_generation_source(connection, project_id, int(latest["id"])) + if latest + else None + ) + old_options = await get_latest_section_options(connection, project_id) + async with connection.cursor(aiomysql.DictCursor) as cursor: + state = await get_workflow_state(cursor, str(project_id)) + if not latest: + return {"error": "저장된 노선이 없어 초기 종단을 셀 수 없습니다(자동설계 먼저)."} + stage2 = next((s for s in state.get("stages", []) if int(s.get("stage_no", -1)) == 2), None) + params = dict((stage2 or {}).get("params") or {}) + first, last = line[0], line[-1] + # 체인 = 새 노선 점으로 stage 2 를 세움 — 측점 간격 · 계획선 기준엔 무관 + params["points"] = { + "bp": {"x": first[0], "y": first[1]}, + "ep": {"x": last[0], "y": last[1]}, + } + model_id = int(latest.get("surface_model_id") or 0) + seed = section_seed(redesign_solve_request(params, selection, model_id), old_options) + options = _regeneration_options(seed, params, float(seed["cross_half_width_m"])) + grade_options = _regeneration_grade_options(params) + epsg = (source or {}).get("crs_epsg") + crs = f"EPSG:{epsg}" if epsg is not None else None + surface = {k: selection.get(k) for k in ("source_filter", "method", "smooth")} + models = project_root / _MODELS_SUBDIR + route_path = str(latest.get("route_data_path") or "") + mark = ip.fingerprint( + { + "key": key, + "nodes": result["nodes"], + "curves": result.get("curves"), + "radius": radius_m, + "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, + ] + ), + "options": asdict(options), + "grade": grade_options.as_dict() if grade_options else None, + "crs": crs, + "with_profile": request.with_profile, + "criteria": request.criteria.model_dump() if request.criteria else None, + "standard": await project_standard(project_id), + "road": await replan._road_settings(project_id), + } + ) + stored = await asyncio.to_thread(ip.read_initial, project_root, key, mark) + if stored is not None: + 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, + ) + target = None + if request.with_profile: # [적용] `balanced_target` 과 같은 셈(노선 정점 · 새 곡선표) + points = [{"x": x, "y": y} for x, y in line] + balanced = await balanced_profile_of( + project_id, project_root, points, request.criteria, curves + ) + target = None if balanced is None else (balanced["stations_m"], balanced["plan_m"]) + sampler = await asyncio.to_thread( + build_surface_sampler, + models, + str(surface["source_filter"]), + str(surface["method"] or "dtm"), + bool(surface["smooth"]), + ) + longitudinal = await asyncio.to_thread( + functools.partial( + ip.initial_longitudinal, + project_root, + line, + curves, + sampler, + facilities, + options=options, + grade_options=grade_options, + source_snapshot={ + "filter": surface["source_filter"], + "method": surface["method"], + "smooth": bool(surface["smooth"]), + }, + crs=crs, + target=target, + ) + ) + summary = { + **ip.profile_summary(longitudinal), + "with_profile": bool(request.with_profile), + "seconds": round(time.perf_counter() - started, 2), + } + await asyncio.to_thread(ip.write_initial, project_root, key, mark, summary) + logger.info( + "추천 초기 종단 셈: project_id=%s %s 꼭짓점 %s 직선화 %s %.2fs", + project_id, + request.mode, + summary.get("pvi_count"), + summary.get("straightened"), + summary["seconds"], + ) + return summary + except Exception as error: # noqa: BLE001 — 추천은 그대로 · 화면이 이유 한 줄 + logger.exception("추천 초기 종단 셈 실패: project_id=%s", project_id) + return {"error": f"초기 종단을 셈하지 못했습니다: {error}"} diff --git a/B05_Profile/logic/B05_Profile_UI_RouteEdit_Edits.ts b/B05_Profile/logic/B05_Profile_UI_RouteEdit_Edits.ts index abafe855d..359bd14f1 100644 --- a/B05_Profile/logic/B05_Profile_UI_RouteEdit_Edits.ts +++ b/B05_Profile/logic/B05_Profile_UI_RouteEdit_Edits.ts @@ -345,6 +345,9 @@ export interface RouteNodeEdit { * * `edits`(430 · 노드 수와 같을 때만 서버가 줌)가 있으면 곡선 켬끔 · R · 고정 · 고정 길이는 **넣은 값**으로 — * 그린 R 로 열면 접선 자리에 눌린 R 이 넣은 값으로 굳고 R · 길이 고정이 [저장] 뒤 사라짐. + * + * 짝(edits 없음) = 파이썬 `B05_Profile_Engine_InitialProfile.draft_nodes`(추천 초기 종단 476) · + * 거울 `test_476_1_recommend_initial`. */ export function flattenServerPlan( nodes: ServerNode[], diff --git a/docs/wiki/concepts/design_page/calculations.md b/docs/wiki/concepts/design_page/calculations.md index f69e093a8..c24929aaf 100644 --- a/docs/wiki/concepts/design_page/calculations.md +++ b/docs/wiki/concepts/design_page/calculations.md @@ -57,6 +57,7 @@ source: ["docs/raw/조사/2026-10-19_472_구조다시세움_목록.md", "PLAN.md | D3 | 3D 길 모형 | `B05_Profile_UI_Corridor_Build.ts` · Worker | — | co | ②(`B05_Profile_Corridor_Prebuild.py`) | | D4 | 평면 발자국 | `logic/B05_Profile_PlanFootprint.ts:160` · `logic/B05_Profile_Footprint_Stage.ts` | `B07_DesignDetail_Engine_Cad_PlanSlope.py`(B07 따로) | — | 브라우저 · `test_346_2_plan_footprint_mirror` | | N1 | 노선 추천 | `B05_Profile_UI_RouteEdit_Initial_Model.ts`(보기만) | `B05_Profile_Engine_RouteInitial*` | — | 서버 · `test_419_1_recommend_engine` | +| N2 | 추천 후보 초기 종단(476) — [적용] 체인과 같은 조각을 메모리에서(노선 짜기 · 시설 따라감 · `build_sections` 438 · 켬 = 균형 얹기) · 저장본 `route_recommend/initial` · 화면 펴기 짝 = `flattenServerPlan` ↔ `draft_nodes` | 그리기만(476-2) · `logic/B05_Profile_UI_RouteEdit_Edits.ts` flattenServerPlan | `B05_Profile_Engine_InitialProfile.py` · `B05_Profile_Router_RouteInitial_Initial.py` | — | 서버 · [적용]은 체인이 다시 셈 · `test_476_1_recommend_initial`(체인 대조 0 mm) | ## 짝인데 값이 갈리는 자리 diff --git a/docs/wiki/concepts/design_page/rules.md b/docs/wiki/concepts/design_page/rules.md index 4854979b6..113a2e6cf 100644 --- a/docs/wiki/concepts/design_page/rules.md +++ b/docs/wiki/concepts/design_page/rules.md @@ -30,7 +30,7 @@ source: ["CLAUDE.md §5", "docs/raw/조사/2026-10-19_465_요구272_값안변함 | 측점 열쇠 = `common_util_station_key` 한 벌 · 노선 바뀜 = 관 · 구조물 좌표(x · y) 따라감 · 누가거리 다시 매김 | `common_util_station_key.ts:32` · `common_util_route_follow.ts:291` | `test_344_3_station_key_mirror` · `test_410_follow_xy` | | 노선 바뀜 넘김 = 열쇠마다 지반 · 측구 종류 · 측구 선택 · 사면 쉬프트 · 소단 · 소단 깊이 · 자리 칸(`SEAT_KEYS` = 암 경계 · 암 절토 경사 · 2단 경사 · 손 상단측 = 단면 유형 · 측구 방향)은 같은 자리(`sameSeat` · 중심 0.01 m · 축 1e-3) 측점만 · 새 자리 = 기본값 · 세션 값은 쓴 때 자리(`@seat`)로 판정(↶ ↷ 유지) · 끌기 갈아끼움 · 미리보기 · Node 넘김 · 서버 노선 저장 한 규칙 | `logic/B06_Section_Station_Swap.ts` SEAT_KEYS · unseated · `logic/B06_Section_Route_Carry.ts` carryKeyedStates · `B05_Profile/logic/B05_Profile_Pipeline_Detail.ts` crossAtSeat · `B05_Profile_Route_Save_Parts.py` carry_cross_inputs | `test_475_3_seat_carry` | | 노선 초안 ↶(버림) = 차례 하나 · 버림 동안 종단도 그리기 멈춤 · 낡음 다시 셈 미룸 → 되돌림 · 저장 기준 맞춤 · 뒤 차례 끝에 한 번 그림(사이 섞인 유토곡선 없음) | `design_pipeline_page.ts:642 · 403` · `B06_Section_Section_Store.ts:316` | `test_472_5c_drop_once` | -| 추천 [적용] 종단 = 새 규칙 초기 종단 — 끔 · 균형 셈 실패 = 1차 계획선(438 · L 20) · 켬 = 균형 종단을 꼭짓점에 얹음(438) · 옛 계획고 · 편집 · L 안 옮김 · 손 노선 저장만 옮김 | `B05_Profile_Route_Save.py` run_route_save_chain `initial` | `test_476_1_apply_initial` | +| 추천 [적용] 종단 = 새 규칙 초기 종단 — 끔 · 균형 셈 실패 = 1차 계획선(438 · L 20) · 켬 = 균형 종단을 꼭짓점에 얹음(438) · 옛 계획고 · 편집 · L 안 옮김 · 손 노선 저장만 옮김 | `B05_Profile_Route_Save.py` run_route_save_chain `initial` · 미리 보기 = `B05_Profile_Engine_InitialProfile`(같은 조각) | `test_476_1_apply_initial` · `test_476_1_recommend_initial` | | B05 · B06 캐시 = 한 벌 · [저장] 한 번에 함께 나감 | `design_shared.ts:395-448` · `design_save.ts:237` | — | | 관 · 구조물 주인 = 가게 칸 하나 — 화면 사본은 가게 알림으로 맞춤(제 쓰기는 같은 값 · 남 고침은 다시 실음) | `design_view_profile.ts` 가게 구독 · `design_view_cross.ts` onShop · `B06_Section_UI_Page_Structures_Panel.ts` seen | `test_472_5_shop_owner` | | 정본 자리 = 값을 낳은 주인 — 노선 시설 = `pipe_points.json` · `structures.json` · 횡단 값 = 횡단 패치 | `B05_Profile_Api_Pipes_Draft.ts:30-37` · `B05_Profile_Structures_Router.py:386-405` · `B06_Section_Section_Store.ts:449` | — | diff --git a/resources/tester/helper_476_1_flatten.cjs b/resources/tester/helper_476_1_flatten.cjs new file mode 100644 index 000000000..55402bede --- /dev/null +++ b/resources/tester/helper_476_1_flatten.cjs @@ -0,0 +1,39 @@ +/* 476-1 추천 [적용] 초안 — 화면 `flattenServerPlan`(logic/B05_Profile_UI_RouteEdit_Edits.ts) + + * `routeDraftOf` 칸 꼴(곡선 켬 = !== false · R ?? null)을 Node 로 돌려 JSON 한 줄 + * (`test_476_1_recommend_initial.py` 가 파이썬 `draft_nodes` 와 견줌). + * node helper_476_1_flatten.cjs <후보 {nodes, curves} JSON 파일> + */ +const fs = require("fs"); +const path = require("path"); +const ts = require(path.join(__dirname, "..", "..", "config", "node_modules", "typescript")); + +const ROOT = path.join(__dirname, "..", ".."); +const cache = new Map(); + +function loadTs(file) { + const full = path.resolve(file); + if (cache.has(full)) return cache.get(full); + const js = ts.transpileModule(fs.readFileSync(full, "utf8"), { + compilerOptions: { module: ts.ModuleKind.CommonJS, target: ts.ScriptTarget.ES2020 }, + }).outputText; + const box = { exports: {} }; + cache.set(full, box.exports); + const localRequire = (request) => + request.startsWith(".") + ? loadTs(path.join(path.dirname(full), `${request}.ts`)) + : require(request); + new Function("exports", "module", "require", js)(box.exports, box, localRequire); + cache.set(full, box.exports); + return box.exports; +} + +const edits = loadTs(path.join(ROOT, "B05_Profile", "logic", "B05_Profile_UI_RouteEdit_Edits.ts")); +const body = JSON.parse(fs.readFileSync(process.argv[2], "utf8")); +const flat = edits.flattenServerPlan(body.nodes, body.curves); +process.stdout.write( + JSON.stringify({ + nodes: flat.planned.map(([x, y]) => [x, y]), + curveOn: flat.planned.map((_, i) => flat.curveOn[i] !== false), + curveRadius: flat.planned.map((_, i) => flat.curveRadius[i] ?? null), + }), +); diff --git a/resources/tester/test_476_1_recommend_initial.py b/resources/tester/test_476_1_recommend_initial.py new file mode 100644 index 000000000..a8d923301 --- /dev/null +++ b/resources/tester/test_476_1_recommend_initial.py @@ -0,0 +1,409 @@ +# -*- coding: utf-8 -*- +"""476-1 추천 후보 초기 종단 — 미리 셈 = [적용] 체인 종단 · 화면 펴기 짝 · 응답 칸 · 저장본 지문. + +고정 자료 = `fixtures/golden_472`(사본) · 체인 쪽 = `run_route_save_chain` 파일 길 차례(노선 csv → +구조물 · 추가 측점 따라감 → 노선 geojson → 관 따라감 · 정본 → `run_section_generation` → +켬이면 `_set_base_pvi` + `_apply_alignment_edits`) · 미리 셈 쪽 = 메모리 조각(파일 안 씀). +화면 펴기 = TS `flattenServerPlan`(Node `helper_476_1_flatten.cjs`) ↔ 파이썬 `draft_nodes`. +""" + +from __future__ import annotations + +import asyncio +import copy +import json +import subprocess +import sys +from pathlib import Path +from uuid import uuid4 + +import pytest + +ROOT = Path(__file__).resolve().parents[2] +TESTER = Path(__file__).resolve().parent +for path in (ROOT, TESTER): + if str(path) not in sys.path: + sys.path.insert(0, str(path)) + +import helper_472_1_paths as P # noqa: E402 +from test_224_3_recommend_store import _ask, wired # noqa: E402, F401 — 가짜 세계 fixture + +from B05_Profile import B05_Profile_Engine_InitialProfile as ip # noqa: E402 + +RADIUS = 12.0 +HELPER = TESTER / "helper_476_1_flatten.cjs" +NODES_CSV = "B05_Profile/route/planned_route_initial_nodes.csv" +WALL = { + "structure_id": "wall600", + "type_id": "retaining_wall", + "placement": "interval", + "chainage_m": 600.0, + "start_m": 595.0, + "end_m": 605.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": 208785.614, + "y": 367845.1028, +} + + +def _candidate(root: Path, shift: float = 0.0) -> dict: + """엔진 결과 꼴 후보(`nodes` · `curves` as_dict) — 교점 9 를 동으로 `shift` m.""" + from B05_Profile.B05_Profile_Router_Replan import _vertices_of + from common_util.common_util_route_polyline import build_planned_polyline + + nodes = [tuple(p) for p in _vertices_of(root / NODES_CSV)] + nodes[9] = (nodes[9][0] + shift, nodes[9][1]) + outline = build_planned_polyline(nodes, min_radius_m=RADIUS, simplify=False) + return { + "nodes": [n.as_dict() for n in outline.nodes], + "curves": [c.as_dict() for c in outline.curves], + } + + +def _browser_draft(candidate: dict, work: Path) -> dict: + if not (ROOT / "config" / "node_modules" / "typescript").is_dir(): + pytest.skip("프론트엔드 의존성 미설치") + body = work / "candidate.json" + body.write_text(json.dumps(candidate), encoding="utf-8") + out = subprocess.run( + ["node", str(HELPER), str(body)], capture_output=True, text=True, check=True + ) + return json.loads(out.stdout) + + +def _settings() -> dict: + from B05_Profile.B05_Profile_Engine_Grade import GradeDesignOptions + from B05_Profile.B05_Profile_Engine_Sections_Core import SectionGenerationOptions + + summary = P.read("summary.json")["longitudinal"]["data"] + options = {k: v for k, v in summary["options"].items() if k != "extra_stations"} + return { + "options": SectionGenerationOptions(**options), + "grade_options": GradeDesignOptions(**summary["grade_options"]), + "crs": f"EPSG:{P.read('context.json')['crs_epsg']}", + } + + +def _old_xy(root: Path) -> list[list[float]]: + coords = json.loads((root / P.ROUTE).read_text("utf-8"))["geometry"]["coordinates"] + return [[c[0], c[1]] for c in coords] + + +def _seed(root: Path, extras: bool) -> None: + from B05_Profile.B05_Profile_Plus_Stations import save_plus_stations + from common_util.common_util_json import atomic_write_json + + if extras: + atomic_write_json( + root / "B05_Profile" / "route" / "structures.json", + {"revision": 1, "structures": [copy.deepcopy(WALL)], "definition_version": None}, + ) + save_plus_stations(root, [510], _old_xy(root)) + + +def _chain(root: Path, draft: dict, target) -> dict: + """`run_route_save_chain` 파일 길 차례(DB 없이) — [적용] 뒤 종단 정본.""" + 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_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 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( + working, + [tuple(n) for n in draft["nodes"]], + RADIUS, + simplify=False, + curve_flags=draft["curveOn"], + radii=draft["curveRadius"], + station_snap=True, + station_snap_step=None, + locks=[None] * count, + 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]) + 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, []) + pipes = [pipe for _, pipe in pairs] + fill_pipe_coordinates(pipes, vertices) + ensure_pipe_ids(pipes) + atomic_write_json( + pipe_points_path_in(root), + {"route_signature": route_signature(vertices), "points": [p.as_dict() for p in pipes]}, + ) + settings = _settings() + run_section_generation(root, P.ROUTE, *P.SURFACE, **settings) + if target is not None: + _set_base_pvi(root / P.LONG, *target) + _apply_alignment_edits(root, P.LONG, {}) + return json.loads((root / P.LONG).read_text("utf-8")) + + +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)), + [], + ) + sampler = build_surface_sampler(root / "B04_PreProcess" / "models", *P.SURFACE) + longitudinal = ip.initial_longitudinal( + root, + line, + curves, + sampler, + facilities, + source_snapshot={"filter": "classification", "method": "dtm", "smooth": True}, + target=target, + **_settings(), + ) + return longitudinal, line, curves + + +def _target(root: Path, line, curves): + from B05_Profile.B05_Profile_Engine_RouteInitial_Profile import route_profile + from common_util.common_util_surface_sampler import build_surface_sampler + + sampler = build_surface_sampler(root / "B04_PreProcess" / "models", *P.SURFACE) + profile = route_profile(line, sampler, max_grade=0.14, design_speed=20, design_curves=curves) + return profile["stations_m"], profile["plan_m"] + + +CASES = { + "same_off": {"shift": 0.0, "extras": False, "balanced": False}, + "moved_off": {"shift": 5.0, "extras": True, "balanced": False}, + "moved_on": {"shift": 5.0, "extras": True, "balanced": True}, +} + + +@pytest.fixture(scope="module", params=list(CASES)) +def pair(request, tmp_path_factory): + case = CASES[request.param] + work = tmp_path_factory.mktemp(request.param) + chain_root = P.stage_project(work / "chain") + preview_root = P.stage_project(work / "preview") + for root in (chain_root, preview_root): + _seed(root, case["extras"]) + candidate = _candidate(preview_root, case["shift"]) + draft = _browser_draft(candidate, work) + target = None + if case["balanced"]: + line, curves = ip.candidate_line(candidate["nodes"], candidate["curves"], RADIUS) + target = _target(preview_root, line, curves) + preview, line, curves = _preview(preview_root, candidate, target) + chain = _chain(chain_root, draft, target) + return {"preview": preview, "chain": chain, "line": line, "curves": curves, "root": chain_root} + + +def test_화면_펴기_짝(tmp_path): + root = P.stage_project(tmp_path / "p") + candidate = _candidate(root, 5.0) + draft = _browser_draft(candidate, tmp_path) + points, flags, radii = ip.draft_nodes(candidate["nodes"], candidate["curves"]) + assert [list(p) for p in points] == draft["nodes"] + assert flags == draft["curveOn"] and radii == draft["curveRadius"] + + +def test_노선_곡선표_같음(pair): + from B05_Profile.B05_Profile_Engine_Sections import _load_design_curves + from B05_Profile.B05_Profile_Router_Replan import _vertices_of + from common_util.common_util_route_geometry import planned_route_working_path + + root = pair["root"] + assert [tuple(p) for p in _vertices_of(planned_route_working_path(root))] == pair["line"] + assert _load_design_curves(root) == pair["curves"] + + +def test_미리_셈_종단_체인과_1mm(pair): + preview, chain = pair["preview"], pair["chain"] + got = [float(s["chainage_m"]) for s in preview["stations"]] + want = [float(s["chainage_m"]) for s in chain["stations"]] + assert got == pytest.approx(want, abs=1e-6) + a = preview["profile_alignment"]["base_pvi"] + b = chain["profile_alignment"]["base_pvi"] + assert len(a) == len(b) + for p, q in zip(a, b): + assert abs(p["chainage_m"] - q["chainage_m"]) <= 1e-3 + assert abs(p["elevation_m"] - q["elevation_m"]) <= 1e-3 + assert ip.profile_summary(preview)["plan_m"] == pytest.approx( + ip.profile_summary(chain)["plan_m"], abs=1e-3 + ) + + +def test_초기_종단_438_규칙(pair): + summary = ip.profile_summary(pair["preview"]) + assert summary["curve_length_m"] == 20.0 # 432 L 20 + assert summary["straightened"] > 0 # R > 500 꼭짓점 묶음 = 직선 위(438) + curves = pair["preview"]["profile_alignment"]["curves"] + assert not [c for c in curves if not c["omitted"] and c["r_m"] > 500.0 + 1e-6] + assert len(summary["line"]) > len(summary["stations_m"]) # 종단곡선 포함 선 + + +# ── 응답 칸 · 저장본 지문 ────────────────────────────────────────────────────── + + +class _Pool: + def acquire(self): + pool = self + + class _Acquire: + async def __aenter__(self): + return pool + + async def __aexit__(self, *exc): + return False + + return _Acquire() + + def cursor(self, *args): + return self.acquire() + + +def _async(value): + async def fake(*args, **kwargs): + return value() if callable(value) else value + + return fake + + +@pytest.fixture +def gather(tmp_path, monkeypatch): + import B04_PreProcess.B04_PreProcess_Router_Watershed as watershed + import B05_Profile.B05_Profile_Repository as b05_repo + import B05_Profile.B05_Profile_Router_Replan as replan + import B05_Profile.B05_Profile_Router_RouteInitial as route_initial + import B05_Profile.B05_Profile_Router_RouteInitial_Initial as gatherer + import B06_Section.B06_Section_Repository as b06_repo + import common_util.common_util_drainage_pipes as drainage + import common_util.common_util_workflow_state as workflow + + root = P.stage_project(tmp_path / "project") + summary = P.read("summary.json")["longitudinal"]["data"] + grade = summary["grade_options"] + params = { + "options": { + "grade_class": grade.get("grade_class") or "trunk", + "terrain_type": grade.get("terrain_type") or "special", + "design_speed_kph": grade.get("design_speed_kph"), + }, + "max_grade_pct": grade.get("max_grade_pct"), + } + state = {"stages": [{"stage_no": 2, "params": params}]} + pipes_file = root / "B04_PreProcess" / "drainage" / "edits" / "pipe_points.json" + monkeypatch.setattr(gatherer, "get_db_pool", lambda: _Pool()) + monkeypatch.setattr(replan, "_plan_criteria", _async((RADIUS, 0.0, 0.0))) + monkeypatch.setattr(replan, "_road_settings", _async(("trunk", 20, "special"))) + monkeypatch.setattr(b05_repo, "get_latest_route", _async({"id": 1, "route_data_path": P.ROUTE})) + monkeypatch.setattr(b06_repo, "get_route_generation_source", _async({"crs_epsg": 5186})) + monkeypatch.setattr(b06_repo, "get_latest_section_options", _async(summary["options"])) + monkeypatch.setattr(workflow, "get_workflow_state", _async(lambda: copy.deepcopy(state))) + monkeypatch.setattr(route_initial, "project_standard", _async(None)) + monkeypatch.setattr(watershed, "_prepare", _async({"streams": []})) + monkeypatch.setattr(watershed, "_sheet_dir", lambda stored: root / "none") + monkeypatch.setattr(drainage, "pipe_points_path", lambda stored: pipes_file) + request = replan.RouteInitialRequest(mode="follow", with_profile=False) + results = root / "B05_Profile" / "route_recommend" + results.mkdir(parents=True, exist_ok=True) + (results / "k1.json").write_text("{}", "utf-8") # 결과 저장본 짝(없으면 초기 종단 파일 지움) + candidate = _candidate(root) + selection = {"source_filter": "classification", "method": "dtm", "smooth": True} + + def run(key: str = "k1") -> dict: + return asyncio.run( + gatherer.initial_profile_of(uuid4(), root, "p", candidate, request, selection, key) + ) + + return {"run": run, "root": root, "state": state, "pipes": pipes_file} + + +def test_응답_칸(gather): + got = gather["run"]() + assert "error" not in got, got + for name in ( + "stations_m", + "ground_m", + "plan_m", + "line", + "pvi_count", + "curve_count", + "straightened", + "max_grade_pct", + "curve_length_m", + "warnings", + ): + assert name in got, name + assert got["with_profile"] is False and got["curve_length_m"] == 20.0 + assert len(got["stations_m"]) == len(got["plan_m"]) == len(got["ground_m"]) + + +def test_지문_같으면_꺼냄_다르면_다시(gather, monkeypatch): + first = gather["run"]() + assert "error" not in first, first + + def broken(*args, **kwargs): + raise RuntimeError("다시 셈") + + monkeypatch.setattr(ip, "initial_longitudinal", broken) + assert gather["run"]() == first # 같은 입력 = 저장본 + gather["state"]["stages"][0]["params"]["max_grade_pct"] = 9.0 # 페이지 설정 최대 기울기 + assert "다시 셈" in gather["run"]()["error"] + gather["state"]["stages"][0]["params"].pop("max_grade_pct") + 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"] + + +def test_코드_바뀌면_다시(gather, monkeypatch): + assert "error" not in gather["run"]() + monkeypatch.setattr(ip, "source_fingerprint", lambda: "other") + monkeypatch.setattr(ip, "initial_longitudinal", lambda *a, **k: {}) + assert "error" in gather["run"]() # 438 등 소스 바뀜 = 다시 셈(빈 종단 = 실패 글) + + +def test_추천_응답에_initial_profile(wired, monkeypatch): + import B05_Profile.B05_Profile_Router_RouteInitial_Initial as gatherer + + seen: list = [] + + async def fake(project_id, root, stored, result, request, selection, key): + seen.append(key) + return {"pvi_count": 3} + + monkeypatch.setattr(gatherer, "initial_profile_of", fake) + body = _ask(mode="earthwork") # 224-3 가짜 세계 + assert body["initial_profile"] == {"pvi_count": 3} and seen # 옛 코드 = 칸 없음