From eee1da09fa158333a13c2a2a589dab11598e6145 Mon Sep 17 00:00:00 2001 From: umsangdon Date: Sun, 18 Oct 2026 12:43:20 +0900 Subject: [PATCH] =?UTF-8?q?feat(B05=20=EC=B6=94=EC=B2=9C=20=ED=9A=A1?= =?UTF-8?q?=EB=8B=A8=20=ED=86=A0=EA=B3=B5):=20CrossEarth=20=EC=83=88=20?= =?UTF-8?q?=ED=8C=8C=EC=9D=BC=20=C2=B7=20=ED=91=9C=EC=A4=80=20=ED=9A=A1?= =?UTF-8?q?=EB=8B=A8=EB=A9=B4=20=3D=20=EC=A0=80=EC=9E=A5=20=ED=8C=A8?= =?UTF-8?q?=EB=84=90=EA=B0=92=20=C2=B7=20=ED=8C=90=208=20+=20=EC=86=8C?= =?UTF-8?q?=EC=8A=A4=20=EC=A7=80=EB=AC=B8=20=C2=B7=20[=EC=9E=AC=EA=B3=84?= =?UTF-8?q?=EC=82=B0]=20=EC=84=9C=EB=B2=84=20=C2=B7=20with=5Fprofile=20?= =?UTF-8?q?=EC=A0=80=EC=9E=A5=20(420)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - 새 B05_Profile_Engine_RouteInitial_CrossEarth.py — cross_earthwork(line, chainages, design_z, terrain, standard) · cross_earthwork_delta(before, after, changed_chainages, …) · attach_cross_earth(결과 표 절 · 성 = 횡단) - 균형 종단 · 횡단 토공 표준 횡단면 = 프로젝트 저장 패널값(추천 · [적용] 둘 다) · generate_initial_route standard 인자 - 저장본 ALGO_VERSION 8 · 엔진 소스 지문 열쇠 · 표준 횡단면 열쇠 · 옛 판 · 옛 지문 미리 셈 때 지움 - 요청 recompute — 묶음 잠금 안에서 이 묶음 저장본 · _POOLS 버리고 새로 셈 · 저장본 request 엔 안 적힘 - initial_choice.json with_profile([적용] 「종단 반영」) · 미리 셈 다섯 갈래 종단 켬 - 시험 420_1 새로 · 224_3 · 364_4 · route_initial 고쳐 씀 Co-Authored-By: Claude Opus 5.5 Claude-Session: https://claude.ai/code/session_01QfHKQs7J3hQELKMWQMUJEy --- .../B05_Profile_Engine_RouteInitial.py | 13 +- ..._Profile_Engine_RouteInitial_CrossEarth.py | 177 +++++++++++++ ...B05_Profile_Engine_RouteInitial_Profile.py | 42 ++- .../B05_Profile_Engine_RouteInitial_Store.py | 94 ++++++- B05_Profile/B05_Profile_Router_Replan.py | 1 + .../B05_Profile_Router_RouteInitial.py | 49 +++- B05_Profile/B05_Profile_Schema.py | 2 + .../tester/test_224_3_recommend_store.py | 27 +- .../tester/test_364_4_precompute_on_open.py | 1 + resources/tester/test_420_1_cross_earth.py | 243 ++++++++++++++++++ resources/tester/test_route_initial.py | 3 + 11 files changed, 626 insertions(+), 26 deletions(-) create mode 100644 B05_Profile/B05_Profile_Engine_RouteInitial_CrossEarth.py create mode 100644 resources/tester/test_420_1_cross_earth.py diff --git a/B05_Profile/B05_Profile_Engine_RouteInitial.py b/B05_Profile/B05_Profile_Engine_RouteInitial.py index ba623c52b..f90567fd1 100644 --- a/B05_Profile/B05_Profile_Engine_RouteInitial.py +++ b/B05_Profile/B05_Profile_Engine_RouteInitial.py @@ -188,6 +188,7 @@ def read_initial_choice(project_root: Path) -> dict[str, Any]: `criteria` 는 **사용자가 겹쳐 쓴 값만**(비운 칸은 안 담김) — 나머지는 프로젝트 기본값이라 화면이 다시 열 때 겹친 칸만 채우고 나머지는 placeholder(기본값)로 비워 둔다. + `with_profile` = [적용] 때 「종단 반영」(없으면 켬 · 420). """ try: data = json.loads(_choice_path(project_root).read_text(encoding="utf-8")) @@ -197,9 +198,15 @@ def read_initial_choice(project_root: Path) -> dict[str, Any]: "logic": logic, "weights": resolve_weights(logic, data.get("weights")), "criteria": criteria if isinstance(criteria, dict) and criteria else None, + "with_profile": data.get("with_profile") is not False, } except (OSError, ValueError, KeyError, TypeError, AttributeError): - return {"logic": "follow", "weights": resolve_weights("follow", None), "criteria": None} + return { + "logic": "follow", + "weights": resolve_weights("follow", None), + "criteria": None, + "with_profile": True, + } def write_initial_choice( @@ -207,6 +214,7 @@ def write_initial_choice( logic: str, weights: dict[str, float] | None, criteria: dict[str, float] | None = None, + with_profile: bool = True, ) -> dict[str, Any]: """[확인] 때만 부름 — 작업본 한 칸. 다시 열기 · 다른 PC 에서도 같은 초기값. @@ -217,6 +225,7 @@ def write_initial_choice( "logic": logic, "weights": resolve_weights(logic, weights), "criteria": criteria or None, + "with_profile": bool(with_profile), } path = _choice_path(project_root) path.parent.mkdir(parents=True, exist_ok=True) @@ -280,6 +289,7 @@ def generate_initial_route( design_speed: int = 20, pipes: list[tuple[float, float, float]] | None = None, with_profile: bool = True, + standard: dict[str, Any] | None = None, ) -> dict[str, Any]: """갈래 하나로 초기 계획노선을 만들고 비교값을 붙여 돌려준다. `max_grade` 는 비율(0.14). @@ -345,6 +355,7 @@ def generate_initial_route( design_speed=design_speed, design_curves=[curve.as_dict() for curve in route[0].curves], exact=exact, + standard=standard, ) bound = None diff --git a/B05_Profile/B05_Profile_Engine_RouteInitial_CrossEarth.py b/B05_Profile/B05_Profile_Engine_RouteInitial_CrossEarth.py new file mode 100644 index 000000000..8102431be --- /dev/null +++ b/B05_Profile/B05_Profile_Engine_RouteInitial_CrossEarth.py @@ -0,0 +1,177 @@ +"""노선 추천 횡단 토공(PLAN 420 · 요구 254). + +측점마다 공식 횡단 엔진(B06 `compute_cross_design` — [저장] 정본과 같은 파이썬)으로 절 · 성 +단면적을 재고 평균단면법으로 토량을 낸다. 띠 가늠(폭 4 m · 비탈 뺌 · `..._Metrics`)을 갈음하는 +추천 토공 — 후보 줄 세움 · 합치기 판정 · 결과 표 절 · 성. + +계산 자리 = 서버 단독(③). 측점 횡단 뜨기 · 단면적은 균형 종단(`..._Profile`)과 같은 함수 — +종단 반영 끔 = 계획고 = 지반고([적용] 뒤 1차 종단과 같음). +""" + +from __future__ import annotations + +from typing import TYPE_CHECKING, Any + +import numpy as np + +from B05_Profile.B05_Profile_Engine_RouteInitial_Profile import ( + STATION_M, + _station_areas, + route_sections, +) + +if TYPE_CHECKING: + from B05_Profile.B05_Profile_Engine_RouteInitial_Search import Terrain + from common_util.common_util_surface_sampler import SurfaceElevationSampler + + +def _end_area_weights(at: np.ndarray) -> np.ndarray: + """평균단면법 무게 — 측점마다 앞 · 뒤 구간 반씩(Σ (A_i + A_i+1) / 2 · L = Σ A_i · w_i).""" + spans = np.diff(at) + weight = np.zeros(len(at)) + weight[:-1] += spans / 2.0 + weight[1:] += spans / 2.0 + return weight + + +def _areas( + line: list[tuple[float, float]], + chainages: Any, + design_z: Any, + terrain: Terrain | SurfaceElevationSampler, + standard: dict[str, Any] | None, + design_curves: list[dict[str, Any]] | None, +) -> tuple[np.ndarray, np.ndarray, np.ndarray, np.ndarray]: + """(누가거리, 지반고, 계획고, [절 자연 · 절 다짐 · 성] ㎡ 행렬).""" + at, ground, sections, modes, widen = route_sections(line, terrain, design_curves, at=chainages) + plan = ground if design_z is None else np.asarray(design_z, dtype=np.float64) + if plan.shape != at.shape: + raise ValueError("계획고 수가 측점 수와 다릅니다.") + rows = np.asarray( + [ + _station_areas(sections[i], modes[i], float(plan[i]), widen[i], standard) + for i in range(len(at)) + ] + ).reshape(-1, 3) + return at, ground, plan, rows + + +def cross_earthwork( + line: list[tuple[float, float]], + chainages: Any, + design_z: Any, + terrain: Terrain | SurfaceElevationSampler, + standard: dict[str, Any] | None = None, + *, + design_curves: list[dict[str, Any]] | None = None, +) -> dict[str, Any]: + """노선 하나의 횡단 토공 — `{cut, cut_compacted, fill, per_station}`(㎥ · 절 = 자연 상태). + + `chainages` = 잴 누가거리(None = 20 m 측점 + 종점 · 균형 종단과 같은 틀) · `design_z` = + 그 측점 계획고(None = 지반고 · 종단 반영 끔) · `standard` = 프로젝트 저장 표준 횡단면 설정값 + (None = 설정 기본값) · `design_curves` = 곡선표(확폭 · 없으면 측점별 실측 반경).""" + at, ground, plan, rows = _areas(line, chainages, design_z, terrain, standard, design_curves) + weight = _end_area_weights(at) + cut, cut_c, fill = (float((weight * rows[:, k]).sum()) for k in range(3)) + return { + "cut": round(cut, 1), + "cut_compacted": round(cut_c, 1), + "fill": round(fill, 1), + "per_station": [ + { + "chainage_m": round(float(s), 2), + "ground_m": round(float(g), 3), + "plan_m": round(float(p), 3), + "cut_m2": round(float(r[0]), 4), + "cut_compacted_m2": round(float(r[1]), 4), + "fill_m2": round(float(r[2]), 4), + } + for s, g, p, r in zip(at, ground, plan, rows) + ], + } + + +def cross_earthwork_delta( + before: list[tuple[float, float]] | dict[str, Any], + after: list[tuple[float, float]], + changed_chainages: Any, + terrain: Terrain | SurfaceElevationSampler, + standard: dict[str, Any] | None = None, + *, + before_chainages: Any = None, + z_before: Any = None, + z_after: Any = None, + design_curves_before: list[dict[str, Any]] | None = None, + design_curves_after: list[dict[str, Any]] | None = None, +) -> dict[str, Any]: + """바뀐 측점만 횡단으로 잰 토공 차(새 − 옛 · ㎥) — 합치기 판정(3~6 측점). + + 측점 하나 = `STATION_M` 몫(가운데 측점 평균단면법 무게). `before` = 옛 노선 정점 또는 그 + 노선의 `cross_earthwork` 결과(그 측점 단면적을 다시 안 셈 · 누가거리 가장 가까운 것) · + `before_chainages` = 옛 노선에서 잴 누가거리(없으면 `changed_chainages`) · `z_*` = 그 측점 + 계획고(None = 지반고). 결과 `cut` · `fill` · `earth` = 차 · `before` 가 결과면 `*_total` = + 고친 뒤 노선 전체 어림(옛 합 + 차).""" + at_after = np.asarray(changed_chainages, dtype=np.float64).reshape(-1) + at_before = at_after if before_chainages is None else np.asarray(before_chainages, float) + if at_after.size == 0: + new = old = np.zeros((0, 3)) + else: + new = _areas(after, at_after, z_after, terrain, standard, design_curves_after)[3] + if isinstance(before, dict): + rows = before["per_station"] + held = np.asarray([row["chainage_m"] for row in rows]) + pick = [int(np.argmin(np.abs(held - s))) for s in at_before] + old = np.asarray( + [[rows[i]["cut_m2"], rows[i]["cut_compacted_m2"], rows[i]["fill_m2"]] for i in pick] + ).reshape(-1, 3) + else: + old = _areas(before, at_before, z_before, terrain, standard, design_curves_before)[3] + cut, cut_c, fill = (float(STATION_M * (new[:, k].sum() - old[:, k].sum())) for k in range(3)) + out = { + "cut": round(cut, 1), + "cut_compacted": round(cut_c, 1), + "fill": round(fill, 1), + "earth": round(cut + fill, 1), + "stations": int(at_after.size), + } + if isinstance(before, dict): + out["cut_total"] = round(before["cut"] + cut, 1) + out["fill_total"] = round(before["fill"] + fill, 1) + return out + + +def attach_cross_earth( + result: dict[str, Any], + terrain: Terrain | SurfaceElevationSampler, + standard: dict[str, Any] | None = None, +) -> None: + """추천 결과 표 절 · 성 = 횡단 값(420) — 종단 켬 = 균형 종단 횡단 ㎥ · 끔 = 계획고 지반고. + + 띠 가늠은 `band_cut_m3` · `band_fill_m3` 로 남긴다. 맞추기 전 선(`unsnapped`)도 같이.""" + for item in (result, result.get("unsnapped")): + if not item or not item.get("planned"): + continue + metrics = item["metrics"] + if "band_cut_m3" in metrics: + continue + profile = metrics.get("profile") + if profile and "error" not in profile: + cut, fill = float(profile["cut_m3"]), float(profile["fill_m3"]) + else: + try: + earth = cross_earthwork( + [tuple(p) for p in item["planned"]], + None, + None, + terrain, + standard, + design_curves=item.get("curves") or None, + ) + except ValueError: # 지반을 못 뜬 노선 — 띠 가늠 그대로 + continue + cut, fill = earth["cut"], earth["fill"] + metrics["band_cut_m3"] = metrics.get("cut_m3") + metrics["band_fill_m3"] = metrics.get("fill_m3") + metrics["cut_m3"] = round(cut, 1) + metrics["fill_m3"] = round(fill, 1) + metrics["surplus_m3"] = round(cut - fill, 1) diff --git a/B05_Profile/B05_Profile_Engine_RouteInitial_Profile.py b/B05_Profile/B05_Profile_Engine_RouteInitial_Profile.py index 095e63c8b..b1f639bd5 100644 --- a/B05_Profile/B05_Profile_Engine_RouteInitial_Profile.py +++ b/B05_Profile/B05_Profile_Engine_RouteInitial_Profile.py @@ -85,20 +85,25 @@ def route_sections( vertices: list[tuple[float, float]], terrain: Terrain | SurfaceElevationSampler, design_curves: list[dict[str, Any]] | None = None, + at: Any = None, ) -> tuple[np.ndarray, np.ndarray, list[list[dict[str, Any]]], list[str], list[dict[str, Any]]]: """측점 누가거리 · 중심 지반고 · 측점별 지반 횡단 샘플 · 단면유형(상단측 절토) · 곡선부 확폭. - 샘플은 `CROSS_WIDE_M` 까지 — ±`CROSS_HALF_M` 밖은 지표면이 끊기기 전까지만 담는다.""" + `at` = 뜰 누가거리(없으면 20 m 측점 + 종점) · 샘플은 `CROSS_WIDE_M` 까지 — ±`CROSS_HALF_M` + 밖은 지표면이 끊기기 전까지만 담는다.""" xy = np.asarray(vertices, dtype=np.float64) keep = np.r_[True, np.hypot(*np.diff(xy, axis=0).T) > 1e-9] xy = xy[keep] seg = np.hypot(*np.diff(xy, axis=0).T) chain = np.concatenate([[0.0], np.cumsum(seg)]) total = float(chain[-1]) - at = np.arange(0.0, total, STATION_M) - if len(at) > 1 and total - at[-1] < TAIL_MERGE_M: - at = at[:-1] - at = np.append(at, total) + if at is not None: + at = np.clip(np.asarray(at, dtype=np.float64), 0.0, total) + else: + at = np.arange(0.0, total, STATION_M) + if len(at) > 1 and total - at[-1] < TAIL_MERGE_M: + at = at[:-1] + at = np.append(at, total) index = np.clip(np.searchsorted(chain, at, side="right") - 1, 0, len(seg) - 1) tangent = np.diff(xy, axis=0)[index] / seg[index][:, None] left = np.column_stack([-tangent[:, 1], tangent[:, 0]]) @@ -165,10 +170,15 @@ def _widenings( def _station_areas( - samples: list[dict[str, Any]], mode: str, elevation: float, widen: dict[str, Any] + samples: list[dict[str, Any]], + mode: str, + elevation: float, + widen: dict[str, Any], + standard: dict[str, Any] | None = None, ) -> tuple[float, float, float]: """측점 하나 · 계획고 하나의 (절토 자연 ㎡, 절토 다짐 환산 ㎡, 성토 ㎡) — 본 설계 기본 설계와 - 같은 입력. 반폭에서 비탈이 안 닫히면 넓힌 샘플로 다시 셈(본 설계 `extend_unclosed`).""" + 같은 입력. 반폭에서 비탈이 안 닫히면 넓힌 샘플로 다시 셈(본 설계 `extend_unclosed`). + `standard` = 프로젝트에 저장된 표준 횡단면 설정값(없으면 설정 기본값 · 420).""" from B06_Section.B06_Section_Engine_Design import compute_cross_design def design(pick: list[dict[str, Any]]) -> dict[str, Any]: @@ -177,6 +187,7 @@ def _station_areas( elevation, ground_type=GROUND_TYPE, section_mode=mode, + standard=standard, rock_boundary_offset_m=STANDARD_ROCK_BOUNDARY_DEFAULT_OFFSET_M, **widen, ) @@ -204,6 +215,7 @@ def _area_curves( ground: np.ndarray, levels: np.ndarray, curves: list[dict[str, Any]] | None = None, + standard: dict[str, Any] | None = None, ) -> tuple[np.ndarray, np.ndarray, np.ndarray]: """측점 · 계획고 후보마다 (절토 자연 ㎡, 절토 다짐 ㎡, 성토 ㎡) — `AREA_STEP_M` 마디 보간.""" shape = levels.shape @@ -213,7 +225,9 @@ def _area_curves( delta = levels[i] - ground[i] lo = np.floor(delta.min() / AREA_STEP_M) * AREA_STEP_M at = np.arange(lo, delta.max() + AREA_STEP_M, AREA_STEP_M) - table = np.asarray([_station_areas(samples, mode, float(ground[i] + d), widen) for d in at]) + table = np.asarray( + [_station_areas(samples, mode, float(ground[i] + d), widen, standard) for d in at] + ) # 단면적은 계획고에 대해 볼록(대략 2차)이라 직선 보간은 크게 잡힌다 — 단조 3차(PCHIP) if len(at) > 1: curve = PchipInterpolator(at, table, axis=0)(delta) @@ -277,11 +291,13 @@ def balanced_profile( design_speed: int = 20, curves: list[dict[str, Any]] | None = None, exact: bool = True, + standard: dict[str, Any] | None = None, ) -> dict[str, Any]: """상한(종단기울기 · 종단곡선)을 지키며 절 · 성토를 횡단 ㎥ 로 맞춘 계획선과 토량. 시점 · 종점은 지반고(확정 체인 기본 오프셋 0 과 같음) — 그 길이 없을 때만 풀어 다시 찾는다. `exact` 끄면 보간값 그대로(후보 줄 세우기 — 빠름) · 켜면 고른 선을 바로 셈해 누가토량을 맞춘다. + `standard` = 프로젝트 저장 표준 횡단면 설정값(없으면 설정 기본값). """ started = time.perf_counter() spans = np.diff(chainage) @@ -291,7 +307,7 @@ def balanced_profile( offsets = np.arange(-LEVEL_RANGE_M, LEVEL_RANGE_M + 1e-9, LEVEL_STEP_M) levels = _reference(ground, spans, max_grade)[:, None] + offsets[None, :] cut, cut_c, fill = ( - weight[:, None] * a for a in _area_curves(sections, modes, ground, levels, curves) + weight[:, None] * a for a in _area_curves(sections, modes, ground, levels, curves, standard) ) criteria = ( FOREST_ROAD_PROFILE_CRITERIA["design_speed"].get(design_speed) @@ -312,7 +328,9 @@ def balanced_profile( plan = levels[np.arange(len(chainage)), result["path"]] rows = np.asarray( [ - _station_areas(sections[i], modes[i], float(plan[i]), curves[i] if curves else {}) + _station_areas( + sections[i], modes[i], float(plan[i]), curves[i] if curves else {}, standard + ) for i in range(len(chainage)) ] ) @@ -423,6 +441,7 @@ def route_profile( design_speed: int, design_curves: list[dict[str, Any]] | None = None, exact: bool = True, + standard: dict[str, Any] | None = None, ) -> dict[str, Any]: """후보 노선 하나의 균형 종단 — 측점 횡단을 뜨고 `balanced_profile`. @@ -439,6 +458,7 @@ def route_profile( design_speed=design_speed, curves=curves, exact=exact, + standard=standard, ) @@ -496,6 +516,7 @@ def attach_profile( design_speed: int, design_curves: list[dict[str, Any]] | None = None, exact: bool = True, + standard: dict[str, Any] | None = None, ) -> None: """후보 비교값에 균형 종단을 붙이고 토공 점수를 그것으로 바꾼다(묶음 캐시에 남음). @@ -513,6 +534,7 @@ def attach_profile( design_speed=design_speed, design_curves=design_curves, exact=exact, + standard=standard, ) except ValueError as error: metrics["profile"] = {"error": str(error)} diff --git a/B05_Profile/B05_Profile_Engine_RouteInitial_Store.py b/B05_Profile/B05_Profile_Engine_RouteInitial_Store.py index 49767ab1d..9cac21d99 100644 --- a/B05_Profile/B05_Profile_Engine_RouteInitial_Store.py +++ b/B05_Profile/B05_Profile_Engine_RouteInitial_Store.py @@ -8,14 +8,16 @@ [초기화]에 안 휩쓸리고, 자료 교체(`purge_output_folders`) · 프로젝트 삭제 때 폴더째 지워진다. 키 = 후보 묶음 지문(`_pool_key` — 지형 · 예상노선 · 계류 · 상한 · 최소 R · 이격 · 고름 · 목표 -기울기) + 갈래 · 비중 · 종단 반영 · 설계속도 · `ALGO_VERSION`. 관 지점은 뺀다 — 덮개 수는 꺼낼 때 -다시 센다. 추천 결과를 바꾸는 코드 고침은 `ALGO_VERSION` 을 올린다(옛 파일은 다른 키라 안 쓰임). +기울기) + 갈래 · 비중 · 종단 반영 · 설계속도 · 표준 횡단면 · `ALGO_VERSION` · 엔진 소스 지문. 관 +지점은 뺀다 — 덮개 수는 꺼낼 때 다시 센다. 추천 결과를 바꾸는 코드 고침은 `ALGO_VERSION` 을 +올린다 — 빠뜨려도 엔진 소스 지문이 바뀌어 새로 셈. 옛 판 · 옛 지문 파일은 미리 셈 때 지운다(420). 기본값 · 저장된 제약 결과(`default`)는 「최근 벌」을 따로 센다(351-1). """ from __future__ import annotations import asyncio +import functools import hashlib import json from contextlib import asynccontextmanager @@ -33,8 +35,16 @@ from B05_Profile.B05_Profile_Engine_RouteInitial import ( from common_util.common_util_json import atomic_write_json # 3 = 곡선 MC 일반측점(351-2) · 4 = 다투는 곡선 끔(364-5) · 5 = 후보마다 맞춤(383-1) · -# 6 = 같은 쪽 곡선 합치기 · 거리 한도 대신 직접 잼(406) · 7 = 측점 밖이 늘면 옛 차례 맞춤(406-4) -ALGO_VERSION = 7 +# 6 = 같은 쪽 곡선 합치기 · 거리 한도 대신 직접 잼(406) · 7 = 측점 밖이 늘면 옛 차례 맞춤(406-4) · +# 8 = 횡단 토공 · 표준 횡단면 = 저장 패널값 · 엔진 소스 지문(420) +ALGO_VERSION = 8 +# 엔진 소스 지문에 드는 파일(저장소 뿌리 기준) — 추천 결과를 낳는 코드 +SOURCE_GLOBS = ( + "B05_Profile/B05_Profile_Engine_RouteInitial*.py", + "B05_Profile/B05_Profile_Engine_Sections_Core.py", + "B06_Section/B06_Section_Engine_Design*.py", + "common_util/common_util_route_polyline.py", +) STORE_DIR = "B05_Profile/route_recommend" # 종류(갈래 · 지형) 마다 남길 벌 수 — 꺼낼 때 수정시각을 새로 해 최근에 쓴 것이 남음 KEEP_PER_KIND = 2 @@ -43,12 +53,25 @@ DEFAULT_MODES = ("gradeline", "earthwork", "gentle", "weighted", "follow") # 묶음 지문마다 잠금 — 진행 중 같은 묶음(같은 키 포함)은 기다렸다 꺼냄(두 번 셈 막음) _LOCKS: dict[str, asyncio.Lock] = {} +# 묶음 지문마다 그 묶음을 짤 때 표준 횡단면(`forget_pool_on_standard`) +_POOL_STANDARDS: dict[str, str] = {} def store_dir(project_root: Path) -> Path: return Path(project_root) / STORE_DIR +@functools.cache +def source_fingerprint() -> str: + """엔진 소스 지문 — `SOURCE_GLOBS` 파일 내용(줄끝 LF 로 맞춤 · 두 PC 같은 코드 = 같은 지문).""" + 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 request_fingerprint( expected: list[tuple[float, float]], terrain: Terrain, @@ -63,6 +86,7 @@ def request_fingerprint( target_grade: float | None, with_profile: bool, design_speed: int, + standard: dict[str, Any] | None = None, ) -> tuple[str, str]: """(저장 키, 묶음 지문) — `generate_initial_route` 와 같은 기본값 풀이로 같은 입력 = 같은 키.""" pool = _pool_key( @@ -74,7 +98,15 @@ def request_fingerprint( *resolve_inputs(max_grade, stream_offset_m, grade_smooth_m, target_grade), ).hex() used = resolve_weights(mode, weights) - tail = (mode, sorted(used.items()), bool(with_profile), int(design_speed), ALGO_VERSION) + tail = ( + mode, + sorted(used.items()), + bool(with_profile), + int(design_speed), + json.dumps(standard, sort_keys=True, default=str) if standard else None, + ALGO_VERSION, + source_fingerprint(), + ) key = hashlib.blake2b((pool + repr(tail)).encode(), digest_size=16).hexdigest() return key, pool @@ -102,14 +134,22 @@ def is_default( ) +def _current(data: Any) -> bool: + return ( + isinstance(data, dict) + and data.get("version") == ALGO_VERSION + and data.get("source") == source_fingerprint() + ) + + def read_stored(project_root: Path, key: str) -> dict[str, Any] | None: - """저장본 한 벌 — 없거나 못 읽거나 판이 다르면 None.""" + """저장본 한 벌 — 없거나 못 읽거나 판 · 소스 지문이 다르면 None.""" path = store_dir(project_root) / f"{key}.json" try: data = json.loads(path.read_text(encoding="utf-8")) except (OSError, ValueError): return None - if not isinstance(data, dict) or data.get("version") != ALGO_VERSION: + if not _current(data): return None try: path.touch() # 최근에 쓴 것 = 정리 때 남김 @@ -134,6 +174,7 @@ def write_stored( path, { "version": ALGO_VERSION, + "source": source_fingerprint(), "key": key, "pool": pool, "default": default, @@ -171,6 +212,45 @@ def prune_stored(project_root: Path, request: dict[str, Any], default: bool = Fa return max(0, len(same) - KEEP_PER_KIND) +def _remove_where(project_root: Path, drop: Any) -> int: + """저장본 중 `drop(data)` 가 참인 파일을 지운다(못 읽는 파일도 지움) — 지운 수.""" + removed = 0 + for path in store_dir(project_root).glob("*.json"): + try: + data = json.loads(path.read_text(encoding="utf-8")) + except (OSError, ValueError): + data = None + if data is None or drop(data): + path.unlink(missing_ok=True) + removed += 1 + return removed + + +def purge_outdated(project_root: Path) -> int: + """옛 판 · 옛 소스 지문 저장본을 지운다(미리 셈 첫머리 · 420) — 지운 수.""" + return _remove_where(project_root, lambda data: not _current(data)) + + +def drop_pool_results(project_root: Path, pool: str) -> int: + """[재계산] — 이 후보 묶음(다섯 갈래 · 종단 켬끔)의 저장본을 지우고 메모리 묶음도 버린다. + + `pool_lock(pool)` 안에서 부름(미리 셈이 다시 써 버리지 않게). 지운 수.""" + from B05_Profile.B05_Profile_Engine_RouteInitial import _POOLS + + _POOLS.pop(bytes.fromhex(pool), None) + return _remove_where(project_root, lambda data: data.get("pool") == pool) + + +def forget_pool_on_standard(pool: str, standard: dict[str, Any] | None) -> None: + """메모리 묶음 후보에 붙은 균형 종단은 그때 표준 횡단면으로 셈 — 바뀌면 묶음을 버린다.""" + from B05_Profile.B05_Profile_Engine_RouteInitial import _POOLS + + mark = json.dumps(standard, sort_keys=True, default=str) + if _POOL_STANDARDS.get(pool, mark) != mark: + _POOLS.pop(bytes.fromhex(pool), None) + _POOL_STANDARDS[pool] = mark + + @asynccontextmanager async def pool_lock(pool: str) -> AsyncIterator[None]: # ponytail: 잠금은 지우지 않음 — 입력 묶음 수만큼(작음) · 다중 프로세스면 파일 잠금으로 diff --git a/B05_Profile/B05_Profile_Router_Replan.py b/B05_Profile/B05_Profile_Router_Replan.py index 51e6cdced..45e184e66 100644 --- a/B05_Profile/B05_Profile_Router_Replan.py +++ b/B05_Profile/B05_Profile_Router_Replan.py @@ -627,6 +627,7 @@ async def replan_route(project_id: UUID, body: RouteSaveRequest) -> dict[str, An choice.mode, choice.weights.model_dump() if choice.weights else None, (choice.criteria.model_dump(exclude_none=True) or None) if choice.criteria else None, + request.balanced_profile, # 「종단 반영」 — 다시 열면 되살림(420) ) if choice else await asyncio.to_thread(read_initial_choice, project_root) diff --git a/B05_Profile/B05_Profile_Router_RouteInitial.py b/B05_Profile/B05_Profile_Router_RouteInitial.py index 962a83d30..118f07de6 100644 --- a/B05_Profile/B05_Profile_Router_RouteInitial.py +++ b/B05_Profile/B05_Profile_Router_RouteInitial.py @@ -6,6 +6,7 @@ import asyncio import functools +import json import logging import time from pathlib import Path @@ -44,17 +45,21 @@ async def compute_initial( """같은 입력이면 저장본을 꺼내고 없으면 셈해 더한다(PLAN 327) · 덮개 수는 늘 지금 관으로. 응답 `stored` = 저장본에서 꺼냈나. 진행 중인 같은 묶음은 기다렸다 꺼낸다(두 번 셈 막음). + `recompute` = [재계산] — 묶음 잠금 안에서 이 묶음 저장본 · 메모리 묶음을 버리고 새로 셈(420). """ from B05_Profile.B05_Profile_Engine_Grade import legal_grade_limit_pct from B05_Profile.B05_Profile_Engine_RouteInitial import ( generate_initial_route, load_terrain_and_streams, ) + from B05_Profile.B05_Profile_Engine_RouteInitial_CrossEarth import attach_cross_earth from B05_Profile.B05_Profile_Engine_RouteInitial_Profile import ( cover_shortfall, project_pipes, ) from B05_Profile.B05_Profile_Engine_RouteInitial_Store import ( + drop_pool_results, + forget_pool_on_standard, is_default, pool_lock, read_stored, @@ -115,6 +120,8 @@ async def compute_initial( "target_grade": None if target_pct is None else target_pct / 100.0, "design_speed": design_speed or 20, "with_profile": request.with_profile, + # 균형 종단 · 횡단 토공 표준 횡단면 = 프로젝트 저장 패널값(420) + "standard": await project_standard(project_id), } started = time.perf_counter() try: @@ -126,6 +133,10 @@ async def compute_initial( functools.partial(request_fingerprint, route, terrain, streams, **inputs) ) async with pool_lock(pool_key): + if request.recompute: # 잠금 안 — 밖이면 미리 셈이 지운 것을 다시 써 버림 + removed = await asyncio.to_thread(drop_pool_results, project_root, pool_key) + logger.info("추천 [재계산]: project_id=%s 저장본 %d 지움", project_id, removed) + forget_pool_on_standard(pool_key, inputs["standard"]) stored = await asyncio.to_thread(read_stored, project_root, key) if stored is not None: result = stored["result"] @@ -136,12 +147,14 @@ async def compute_initial( generate_initial_route, route, terrain, streams, pipes=None, **inputs ) ) + # 결과 표 절 · 성 = 횡단(420) — 저장본에 남김 + await asyncio.to_thread(attach_cross_earth, result, terrain, inputs["standard"]) await asyncio.to_thread( write_stored, project_root, key, pool_key, - request.model_dump(), + request.model_dump(exclude={"recompute"}), result, is_default( request.mode, @@ -180,13 +193,36 @@ async def compute_initial( } +async def project_standard(project_id: UUID) -> dict[str, Any] | None: + """프로젝트에 저장된 표준 횡단면 설정값(B06 패널 · 확정 때 종단 options) — 없으면 None. + + 공식 재계산(`B06_Section_Server_Calc_Prebuild` · `stored_standard_cross_section`)과 같은 + 자리 · 같은 읽기 — 마지막 종단 정본 행.""" + from B06_Section.B06_Section_Router_Design import stored_standard_cross_section + + pool = get_db_pool() + async with pool.acquire() as connection: + async with connection.cursor() as cursor: + await cursor.execute( + "SELECT data FROM longitudinal_sections WHERE project_id = %s " + "ORDER BY id DESC LIMIT 1", + (str(project_id),), + ) + row = await cursor.fetchone() + data = row[0] if row else None + if isinstance(data, str): + data = json.loads(data) + return stored_standard_cross_section({"data": data if isinstance(data, dict) else None}) + + async def precompute_defaults(project_id: UUID) -> None: """체인 끝 예약(PLAN 327-2 · 351-1) — 기본 다섯 알약 + 저장된 제약(initial_choice)의 다섯. 모달은 저장된 제약 · 비중으로 열리므로 그 값으로도 미리 셈한다. 있으면 꺼내기만. + 다섯 갈래 모두 종단 켬(420) · 첫머리에 옛 판 · 옛 소스 지문 저장본을 지운다. """ from B05_Profile.B05_Profile_Engine_RouteInitial import DEFAULT_WEIGHTS - from B05_Profile.B05_Profile_Engine_RouteInitial_Store import DEFAULT_MODES + from B05_Profile.B05_Profile_Engine_RouteInitial_Store import DEFAULT_MODES, purge_outdated started = time.perf_counter() variants: list[tuple[Any, Any]] = [(None, None)] @@ -196,6 +232,9 @@ async def precompute_defaults(project_id: UUID) -> None: except LookupError: paths = None if paths is not None: + removed = await asyncio.to_thread(purge_outdated, paths[0]) + if removed: + logger.info("추천 옛 판 저장본 지움: project_id=%s %d", project_id, removed) saved = await asyncio.to_thread(read_initial_choice, paths[0]) weights = saved["weights"] if saved["logic"] == "weighted" else None if saved["criteria"] or (weights and weights != DEFAULT_WEIGHTS): @@ -209,10 +248,10 @@ async def precompute_defaults(project_id: UUID) -> None: for mode in modes: if progress is not None: progress["mode"] = mode - # 화면과 같은 몸통 — 설계 지형 · 비중은 비중 갈래만 · 추적만 종단 끔(`withProfileOf`) + # 화면과 같은 몸통 — 설계 지형 · 비중은 비중 갈래만 · 「종단 반영」 기본 켬(420) request = replan.RouteInitialRequest( mode=mode, - with_profile=mode != "follow", + with_profile=True, criteria=criteria, weights=weights if mode == "weighted" else None, ) @@ -416,6 +455,7 @@ async def balanced_profile_of( pool = get_db_pool() async with pool.acquire() as connection: selection = await get_surface_confirmation_params(connection, str(project_id)) + standard = await project_standard(project_id) # 추천 창과 같은 저장 패널값(420) try: # 측점 생성과 같은 1 m 쌍선형 표본(추천 점수의 2 m 비용면 아님) surface = await asyncio.to_thread( @@ -433,6 +473,7 @@ async def balanced_profile_of( max_grade=max_grade, design_speed=design_speed or 20, design_curves=design_curves, + standard=standard, ) ) except (FileNotFoundError, ValueError): diff --git a/B05_Profile/B05_Profile_Schema.py b/B05_Profile/B05_Profile_Schema.py index 2ca2d0c92..d429a5aa9 100644 --- a/B05_Profile/B05_Profile_Schema.py +++ b/B05_Profile/B05_Profile_Schema.py @@ -367,6 +367,8 @@ class RouteInitialRequest(BaseModel): with_profile: bool = True # 분석 지형(PLAN 133-1) — None = 지금 설계 지형 · [확인] 때 설계 지형이 됨(133-2). terrain: Literal["las", "sheet"] | None = None + # [재계산](420) — 이 후보 묶음 저장본 · 메모리 묶음을 지우고 새로 셈. 저장본 request 엔 안 적힘. + recompute: bool = False class RouteReplanRequest(BaseModel): diff --git a/resources/tester/test_224_3_recommend_store.py b/resources/tester/test_224_3_recommend_store.py index 528f34605..5c2344e73 100644 --- a/resources/tester/test_224_3_recommend_store.py +++ b/resources/tester/test_224_3_recommend_store.py @@ -96,7 +96,16 @@ def test_no_overwrite(tmp_path: Path) -> None: assert not write_stored(tmp_path, "k", "p", {}, {"n": 2}, False) data = json.loads((store_dir(tmp_path) / "k.json").read_text(encoding="utf-8")) assert data["result"] == {"n": 1} and data["default"] is True - assert set(data) == {"version", "key", "pool", "default", "created_at", "request", "result"} + assert set(data) == { + "version", + "source", + "key", + "pool", + "default", + "created_at", + "request", + "result", + } def test_survives_reset_and_goes_with_data_swap(tmp_path: Path) -> None: @@ -117,6 +126,8 @@ def test_survives_reset_and_goes_with_data_swap(tmp_path: Path) -> None: class _Calls: def __init__(self) -> None: self.count = 0 + self.inputs: list[dict] = [] + self.standard: dict | None = None @pytest.fixture @@ -141,20 +152,26 @@ def wired(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> _Calls: async def surface_params(_connection, _pid): return {} + async def project_standard(_pid): + return calls.standard + class _Pool: @asynccontextmanager async def acquire(self): yield None - def generate(expected, terrain, streams, *, pipes, **_inputs): + def generate(expected, terrain, streams, *, pipes, **inputs): assert pipes is None # 관은 키 밖 — 덮개 없이 셈해 남김 calls.count += 1 + calls.inputs.append(inputs) return { "planned": [[x, y] for x, y in expected], "metrics": { "length_m": 38.0, "offset_avg_m": 0.0, "profile": { + "cut_m3": 10.0, + "fill_m3": 4.0, "stations_m": [0.0, 38.0], "plan_m": [1.0, 1.0], "ground_m": [0.0, 0.0], @@ -167,6 +184,7 @@ def wired(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> _Calls: monkeypatch.setattr(replan, "_road_settings", road_settings) monkeypatch.setattr(replan, "_vertices_of", lambda _path: [list(p) for p in EXPECTED]) monkeypatch.setattr(router, "get_db_pool", lambda: _Pool()) + monkeypatch.setattr(router, "project_standard", project_standard) monkeypatch.setattr(surface, "get_surface_confirmation_params", surface_params) monkeypatch.setattr(surface, "analysis_terrain_params", lambda selection, _t: selection) monkeypatch.setattr(grade, "legal_grade_limit_pct", lambda *_a: 14.0) @@ -225,7 +243,8 @@ def test_precompute_defaults_then_hits(wired: _Calls) -> None: ] assert sorted(item["request"]["mode"] for item in saved) == sorted(DEFAULT_MODES) assert all(item["default"] for item in saved) - # 화면 첫 누름(비중은 화면이 50/25/25 를 채워 보냄 · 추적은 종단 끔) — 셈 없이 저장본 + # 화면 첫 누름(비중은 화면이 50/25/25 를 채워 보냄 · 다섯 갈래 종단 켬 · 420) — 셈 없이 저장본 + assert all(item["request"]["with_profile"] for item in saved) hit = _ask(mode="weighted", weights={"expected": 50, "earthwork": 25, "grade": 25}) - follow = _ask(mode="follow", with_profile=False) + follow = _ask(mode="follow") assert hit["stored"] and follow["stored"] and wired.count == 5 diff --git a/resources/tester/test_364_4_precompute_on_open.py b/resources/tester/test_364_4_precompute_on_open.py index a13910ea5..f8728a393 100644 --- a/resources/tester/test_364_4_precompute_on_open.py +++ b/resources/tester/test_364_4_precompute_on_open.py @@ -90,6 +90,7 @@ def test_모달이_여는_갈래_먼저(monkeypatch) -> None: async def compute(_pid, request): asked.append(request.mode) + assert request.with_profile # 다섯 갈래 모두 종단 켬(420) return {} monkeypatch.setattr(router.replan, "_project_paths", paths) diff --git a/resources/tester/test_420_1_cross_earth.py b/resources/tester/test_420_1_cross_earth.py new file mode 100644 index 000000000..35a5f2c83 --- /dev/null +++ b/resources/tester/test_420_1_cross_earth.py @@ -0,0 +1,243 @@ +"""420 추천 횡단 토공 · 저장본 판 · [재계산] 서버(요구 254). + + ./venv/Scripts/python.exe -m pytest resources/tester/test_420_1_cross_earth.py -q + +횡단 토공 = 공식 횡단 단면적 평균단면법(저장 정본 ±6 %) · 차 = 바뀐 측점만 · 결과 표 절 · 성 = +횡단 · [재계산] = 잠금 안 지우고 새로 셈(셈 수 1 → 2) · 옛 판 · 옛 소스 지문 미리 셈 때 지움 · +표준 횡단면 = 저장 패널값(키 · 엔진에 감) · `initial_choice` with_profile 되살림. +""" + +import asyncio +import json +from pathlib import Path +from uuid import uuid4 + +import numpy as np +import pytest +from test_224_3_recommend_store import EXPECTED, STREAMS, _ask, _key, _terrain, wired # noqa: F401 + +from B05_Profile import B05_Profile_Router_RouteInitial as router +from B05_Profile.B05_Profile_Engine_RouteInitial import ( + Terrain, + read_initial_choice, + write_initial_choice, +) +from B05_Profile.B05_Profile_Engine_RouteInitial_CrossEarth import ( + attach_cross_earth, + cross_earthwork, + cross_earthwork_delta, +) +from B05_Profile.B05_Profile_Engine_RouteInitial_Profile import _station_areas, route_sections +from B05_Profile.B05_Profile_Engine_RouteInitial_Store import ( + ALGO_VERSION, + DEFAULT_MODES, + source_fingerprint, + store_dir, +) + +ROOT = Path(__file__).resolve().parents[2] +SUB2 = ROOT / "storage/1/3/cb15892a-5788-4a26-9316-20a15421f68b" + + +def _slope() -> Terrain: + """옆경사 30 % 사면(노선 = x 방향) — 절 · 성 둘 다 생김.""" + x = np.arange(0.0, 200.0, 2.0) + y = np.arange(-80.0, 80.0, 2.0) + xx, yy = np.meshgrid(x, y) + return Terrain(x, y, 0.05 * xx + 0.3 * yy, np.ones(xx.shape, dtype=bool)) + + +LINE = [(20.0, 0.0), (100.0, 0.0), (180.0, 0.0)] + + +def test_횡단_토공_평균단면법() -> None: + terrain = _slope() + earth = cross_earthwork(LINE, None, None, terrain) + rows = earth["per_station"] + assert [r["chainage_m"] for r in rows] == [float(s) for s in range(0, 161, 20)] + # 직접 셈한 단면적 · Σ (A_i + A_i+1) / 2 · L + at, ground, sections, modes, widen = route_sections(LINE, terrain) + areas = [_station_areas(sections[i], modes[i], ground[i], widen[i]) for i in range(len(at))] + cut = sum((areas[i][0] + areas[i + 1][0]) / 2 * 20.0 for i in range(len(at) - 1)) + fill = sum((areas[i][2] + areas[i + 1][2]) / 2 * 20.0 for i in range(len(at) - 1)) + assert earth["cut"] == pytest.approx(cut, abs=0.1) and earth["cut"] > 0 + assert earth["fill"] == pytest.approx(fill, abs=0.1) and earth["fill"] > 0 + # 계획고를 1 m 올리면 성토가 늚 · 측점 수와 다른 계획고는 거절 + up = cross_earthwork(LINE, None, np.asarray(ground) + 1.0, terrain) + assert up["fill"] > earth["fill"] and up["cut"] < earth["cut"] + with pytest.raises(ValueError): + cross_earthwork(LINE, None, [1.0, 2.0], terrain) + + +def test_표준_횡단면이_단면적을_바꿈() -> None: + from config.config_system_design import STANDARD_CROSS_SECTION + + terrain = _slope() + wide = json.loads(json.dumps(STANDARD_CROSS_SECTION)) + for group in wide.values(): + group["road_width_m"] += 2.0 + base = cross_earthwork(LINE, None, None, terrain) + assert cross_earthwork(LINE, None, None, terrain, wide)["cut"] > base["cut"] + + +def test_토공_차_바뀐_측점만() -> None: + terrain = _slope() + moved = [(20.0, 0.0), (100.0, 6.0), (180.0, 0.0)] + changed = [60.0, 80.0, 100.0] + same = cross_earthwork_delta(LINE, LINE, changed, terrain) + assert same["cut"] == 0 and same["fill"] == 0 and same["stations"] == 3 + delta = cross_earthwork_delta(LINE, moved, changed, terrain) + old = cross_earthwork(LINE, changed, None, terrain)["per_station"] + new = cross_earthwork(moved, changed, None, terrain)["per_station"] + expect = 20.0 * sum(n["fill_m2"] - o["fill_m2"] for n, o in zip(new, old)) + assert delta["fill"] == pytest.approx(expect, abs=0.2) + # 옛 결과를 넘기면 옛 단면적을 다시 안 셈 · 전체 어림 = 옛 합 + 차 + whole = cross_earthwork(LINE, None, None, terrain) + reuse = cross_earthwork_delta(whole, moved, changed, terrain) + assert reuse["fill"] == pytest.approx(delta["fill"], abs=0.2) + assert reuse["fill_total"] == pytest.approx(whole["fill"] + reuse["fill"], abs=0.2) + assert cross_earthwork_delta(LINE, moved, [], terrain)["earth"] == 0 + + +@pytest.mark.skipif(not (SUB2 / "B05_Profile").is_dir(), reason="검증_sub2 storage 없음") +def test_검증_sub2_저장_정본_6퍼센트() -> None: + """저장된 노선 측점 · 계획고 = 지반 — 공식 [저장] 단면적 합과 ±6 %(2 m 추천 지형 · 1 m 표본).""" + from B05_Profile.B05_Profile_Engine_RouteInitial import load_terrain_and_streams + from B05_Profile.B05_Profile_Engine_Solver import _MODELS_SUBDIR + from common_util.common_util_route_geometry import read_planned_route_csv + from common_util.common_util_surface_sampler import build_surface_sampler + + route = read_planned_route_csv(SUB2 / "B05_Profile/route/planned_route_initial.csv") + curves = json.loads( + (SUB2 / "B05_Profile/route/planned_route_initial_curves.json").read_text("utf-8") + ) + curves = curves.get("curves") if isinstance(curves, dict) else curves + line = [(v.x, v.y) for v in route.vertices] + files = sorted((SUB2 / "B06_Section/cross_sections").glob("cross_*.json")) + at = [json.loads(p.read_text("utf-8"))["chainage_m"] for p in files] + # 공식 [저장] 단면적(DB cross_sections.design · 2026-10-18 뜸 · 계획고 = 지반) 평균단면법 합 + stored_cut, stored_fill = 5166.0, 23079.3 + selection = {"source_filter": "classification", "method": "dtm", "smooth": True} + terrain, _ = load_terrain_and_streams(SUB2, selection) + surface = build_surface_sampler(SUB2 / _MODELS_SUBDIR, "classification", "dtm", True) + for ground in (terrain, surface): + earth = cross_earthwork(line, at, None, ground, design_curves=curves) + assert earth["cut"] == pytest.approx(stored_cut, rel=0.06) + assert earth["fill"] == pytest.approx(stored_fill, rel=0.06) + + +def test_결과_절성_횡단_값() -> None: + terrain = _slope() + plain = {"planned": [list(p) for p in LINE], "curves": [], "metrics": {"cut_m3": 1.0}} + profiled = { + "planned": [list(p) for p in LINE], + "metrics": {"cut_m3": 1.0, "fill_m3": 2.0, "profile": {"cut_m3": 30.0, "fill_m3": 20.0}}, + } + result = {**plain, "unsnapped": profiled} + attach_cross_earth(result, terrain) + earth = cross_earthwork(LINE, None, None, terrain) + assert result["metrics"]["cut_m3"] == earth["cut"] and result["metrics"]["band_cut_m3"] == 1.0 + assert result["metrics"]["surplus_m3"] == pytest.approx(earth["cut"] - earth["fill"], abs=0.2) + shown = result["unsnapped"]["metrics"] + assert (shown["cut_m3"], shown["fill_m3"], shown["band_fill_m3"]) == (30.0, 20.0, 2.0) + attach_cross_earth(result, terrain) # 두 번 붙여도 띠 가늠을 안 덮음 + assert result["metrics"]["band_cut_m3"] == 1.0 + + +def test_표준_횡단면_키_엔진에_감(wired) -> None: # noqa: F811 + assert _key(standard={"a": 1}) != _key() and _key(standard={"a": 1}) == _key(standard={"a": 1}) + wired.standard = {"road": {"width_m": 5.0}} + _ask(mode="earthwork") + assert wired.inputs[-1]["standard"] == wired.standard + + +def test_재계산_잠금_안_지우고_새로(wired) -> None: # noqa: F811 + first = _ask(mode="earthwork") + _ask(mode="gentle") + _ask(mode="earthwork", criteria={"min_radius_m": 15.0}) # 다른 묶음 + assert wired.count == 3 and first["stored"] is False + files = {p.name: p.read_text("utf-8") for p in store_dir(wired.root).glob("*.json")} + again = _ask(mode="earthwork", recompute=True) + assert again["stored"] is False and wired.count == 4 # 셈 1 → 2(earthwork) + now = {p.name: json.loads(p.read_text("utf-8")) for p in store_dir(wired.root).glob("*.json")} + # 같은 묶음 gentle 은 지워짐 · 다른 묶음은 그대로 · earthwork 파일은 새로 씀 + assert len(now) == 2 + other = [n for n, d in now.items() if d["request"]["criteria"]] + assert other and now[other[0]] == json.loads(files[other[0]]) + fresh = next(d for n, d in now.items() if not d["request"]["criteria"]) + assert "recompute" not in fresh["request"] and fresh["version"] == ALGO_VERSION + assert fresh["result"]["metrics"]["cut_m3"] == 10.0 # 횡단(균형 종단) 값 + assert _ask(mode="earthwork")["stored"] and wired.count == 4 + + +def test_재계산_메모리_묶음도_버림(wired) -> None: # noqa: F811 + from B05_Profile.B05_Profile_Engine_RouteInitial import _POOLS + from B05_Profile.B05_Profile_Engine_RouteInitial_Store import request_fingerprint + + _ask(mode="earthwork") + pool = request_fingerprint( + EXPECTED, + _terrain(), + STREAMS, + mode="earthwork", + weights=None, + max_grade=0.14, + min_radius_m=12.0, + stream_offset_m=None, + grade_smooth_m=None, + target_grade=None, + with_profile=True, + design_speed=20, + )[1] + _POOLS[bytes.fromhex(pool)] = "옛 묶음" + _ask(mode="earthwork") + assert bytes.fromhex(pool) in _POOLS + _ask(mode="earthwork", recompute=True) + assert bytes.fromhex(pool) not in _POOLS + + +def test_옛_판_옛_지문_미리_셈_때_지움(wired) -> None: # noqa: F811 + folder = store_dir(wired.root) + folder.mkdir(parents=True) + old = {"version": ALGO_VERSION - 1, "source": source_fingerprint(), "request": {}} + stale = {"version": ALGO_VERSION, "source": "0" * 16, "request": {}} + (folder / "old.json").write_text(json.dumps(old), encoding="utf-8") + (folder / "stale.json").write_text(json.dumps(stale), encoding="utf-8") + (folder / "broken.json").write_text("{", encoding="utf-8") + asyncio.run(router.precompute_defaults(uuid4())) + names = {p.stem for p in folder.glob("*.json")} + assert not names & {"old", "stale", "broken"} and len(names) == len(DEFAULT_MODES) + assert wired.count == len(DEFAULT_MODES) + + +def test_소스_지문이_바뀌면_다른_키(monkeypatch: pytest.MonkeyPatch) -> None: + import B05_Profile.B05_Profile_Engine_RouteInitial_Store as store + + before = _key() + monkeypatch.setattr(store, "source_fingerprint", lambda: "f" * 16) + assert _key() != before + + +def test_with_profile_되살림(tmp_path: Path) -> None: + assert read_initial_choice(tmp_path)["with_profile"] is True # 없으면 켬 + write_initial_choice(tmp_path, "gentle", None, None, False) + assert read_initial_choice(tmp_path)["with_profile"] is False + write_initial_choice(tmp_path, "gentle", None, None) + assert read_initial_choice(tmp_path)["with_profile"] is True + # 옛 파일(칸 없음) = 켬 + path = tmp_path / "B05_Profile/route/initial_choice.json" + path.write_text(json.dumps({"logic": "gentle"}), encoding="utf-8") + assert read_initial_choice(tmp_path)["with_profile"] is True + + +def test_적용이_종단_반영을_적음_소스() -> None: + source = (ROOT / "B05_Profile/B05_Profile_Router_Replan.py").read_text(encoding="utf-8") + call = source[source.index("write_initial_choice,\n") :][:600] + assert "request.balanced_profile" in call + initial = (ROOT / "B05_Profile/B05_Profile_Router_RouteInitial.py").read_text(encoding="utf-8") + # 지우기는 잠금 안 + lock = initial.index("async with pool_lock(pool_key):") + assert lock < initial.index("drop_pool_results, project_root, pool_key") < lock + 400 + assert 'model_dump(exclude={"recompute"})' in initial + # [적용] 균형 종단도 저장 패널값 + assert "standard=standard" in initial[initial.index("async def balanced_profile_of") :] diff --git a/resources/tester/test_route_initial.py b/resources/tester/test_route_initial.py index c29b0d5eb..9d467c82b 100644 --- a/resources/tester/test_route_initial.py +++ b/resources/tester/test_route_initial.py @@ -321,6 +321,7 @@ def test_initial_choice_roundtrip(tmp_path) -> None: "logic": "follow", "weights": {"expected": 100.0, "earthwork": 0.0, "grade": 0.0}, "criteria": None, + "with_profile": True, } for logic in ("gradeline", "safe", "crossings", "cost"): # 새 갈래도 적고 다시 읽음(23-5) write_initial_choice(tmp_path, logic, None, {"target_grade_pct": 10.0}) @@ -331,6 +332,7 @@ def test_initial_choice_roundtrip(tmp_path) -> None: "logic": "weighted", "weights": {"expected": 60.0, "earthwork": 20.0, "grade": 20.0}, "criteria": None, + "with_profile": True, } # 사용자가 화면 칸에 겹쳐 넣은 값만(빈 칸은 안 담김) — 남으면 다시 열 때 그 칸만 채운다. write_initial_choice( @@ -340,6 +342,7 @@ def test_initial_choice_roundtrip(tmp_path) -> None: "logic": "earthwork", "weights": {"expected": 0.0, "earthwork": 100.0, "grade": 0.0}, "criteria": {"stream_offset_m": 30.0, "grade_smooth_m": 2.0}, + "with_profile": True, } (tmp_path / "B05_Profile" / "route" / "initial_choice.json").write_text("{", encoding="utf-8") assert read_initial_choice(tmp_path)["logic"] == "follow"