feat(B05 추천 횡단 토공): CrossEarth 새 파일 · 표준 횡단면 = 저장 패널값 · 판 8 + 소스 지문 · [재계산] 서버 · with_profile 저장 (420)
- 새 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 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01QfHKQs7J3hQELKMWQMUJEy
This commit is contained in:
@@ -188,6 +188,7 @@ def read_initial_choice(project_root: Path) -> dict[str, Any]:
|
|||||||
|
|
||||||
`criteria` 는 **사용자가 겹쳐 쓴 값만**(비운 칸은 안 담김) — 나머지는 프로젝트 기본값이라
|
`criteria` 는 **사용자가 겹쳐 쓴 값만**(비운 칸은 안 담김) — 나머지는 프로젝트 기본값이라
|
||||||
화면이 다시 열 때 겹친 칸만 채우고 나머지는 placeholder(기본값)로 비워 둔다.
|
화면이 다시 열 때 겹친 칸만 채우고 나머지는 placeholder(기본값)로 비워 둔다.
|
||||||
|
`with_profile` = [적용] 때 「종단 반영」(없으면 켬 · 420).
|
||||||
"""
|
"""
|
||||||
try:
|
try:
|
||||||
data = json.loads(_choice_path(project_root).read_text(encoding="utf-8"))
|
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,
|
"logic": logic,
|
||||||
"weights": resolve_weights(logic, data.get("weights")),
|
"weights": resolve_weights(logic, data.get("weights")),
|
||||||
"criteria": criteria if isinstance(criteria, dict) and criteria else None,
|
"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):
|
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(
|
def write_initial_choice(
|
||||||
@@ -207,6 +214,7 @@ def write_initial_choice(
|
|||||||
logic: str,
|
logic: str,
|
||||||
weights: dict[str, float] | None,
|
weights: dict[str, float] | None,
|
||||||
criteria: dict[str, float] | None = None,
|
criteria: dict[str, float] | None = None,
|
||||||
|
with_profile: bool = True,
|
||||||
) -> dict[str, Any]:
|
) -> dict[str, Any]:
|
||||||
"""[확인] 때만 부름 — 작업본 한 칸. 다시 열기 · 다른 PC 에서도 같은 초기값.
|
"""[확인] 때만 부름 — 작업본 한 칸. 다시 열기 · 다른 PC 에서도 같은 초기값.
|
||||||
|
|
||||||
@@ -217,6 +225,7 @@ def write_initial_choice(
|
|||||||
"logic": logic,
|
"logic": logic,
|
||||||
"weights": resolve_weights(logic, weights),
|
"weights": resolve_weights(logic, weights),
|
||||||
"criteria": criteria or None,
|
"criteria": criteria or None,
|
||||||
|
"with_profile": bool(with_profile),
|
||||||
}
|
}
|
||||||
path = _choice_path(project_root)
|
path = _choice_path(project_root)
|
||||||
path.parent.mkdir(parents=True, exist_ok=True)
|
path.parent.mkdir(parents=True, exist_ok=True)
|
||||||
@@ -280,6 +289,7 @@ def generate_initial_route(
|
|||||||
design_speed: int = 20,
|
design_speed: int = 20,
|
||||||
pipes: list[tuple[float, float, float]] | None = None,
|
pipes: list[tuple[float, float, float]] | None = None,
|
||||||
with_profile: bool = True,
|
with_profile: bool = True,
|
||||||
|
standard: dict[str, Any] | None = None,
|
||||||
) -> dict[str, Any]:
|
) -> dict[str, Any]:
|
||||||
"""갈래 하나로 초기 계획노선을 만들고 비교값을 붙여 돌려준다. `max_grade` 는 비율(0.14).
|
"""갈래 하나로 초기 계획노선을 만들고 비교값을 붙여 돌려준다. `max_grade` 는 비율(0.14).
|
||||||
|
|
||||||
@@ -345,6 +355,7 @@ def generate_initial_route(
|
|||||||
design_speed=design_speed,
|
design_speed=design_speed,
|
||||||
design_curves=[curve.as_dict() for curve in route[0].curves],
|
design_curves=[curve.as_dict() for curve in route[0].curves],
|
||||||
exact=exact,
|
exact=exact,
|
||||||
|
standard=standard,
|
||||||
)
|
)
|
||||||
|
|
||||||
bound = None
|
bound = None
|
||||||
|
|||||||
@@ -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)
|
||||||
@@ -85,16 +85,21 @@ def route_sections(
|
|||||||
vertices: list[tuple[float, float]],
|
vertices: list[tuple[float, float]],
|
||||||
terrain: Terrain | SurfaceElevationSampler,
|
terrain: Terrain | SurfaceElevationSampler,
|
||||||
design_curves: list[dict[str, Any]] | None = None,
|
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]]]:
|
) -> 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)
|
xy = np.asarray(vertices, dtype=np.float64)
|
||||||
keep = np.r_[True, np.hypot(*np.diff(xy, axis=0).T) > 1e-9]
|
keep = np.r_[True, np.hypot(*np.diff(xy, axis=0).T) > 1e-9]
|
||||||
xy = xy[keep]
|
xy = xy[keep]
|
||||||
seg = np.hypot(*np.diff(xy, axis=0).T)
|
seg = np.hypot(*np.diff(xy, axis=0).T)
|
||||||
chain = np.concatenate([[0.0], np.cumsum(seg)])
|
chain = np.concatenate([[0.0], np.cumsum(seg)])
|
||||||
total = float(chain[-1])
|
total = float(chain[-1])
|
||||||
|
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)
|
at = np.arange(0.0, total, STATION_M)
|
||||||
if len(at) > 1 and total - at[-1] < TAIL_MERGE_M:
|
if len(at) > 1 and total - at[-1] < TAIL_MERGE_M:
|
||||||
at = at[:-1]
|
at = at[:-1]
|
||||||
@@ -165,10 +170,15 @@ def _widenings(
|
|||||||
|
|
||||||
|
|
||||||
def _station_areas(
|
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]:
|
) -> tuple[float, float, float]:
|
||||||
"""측점 하나 · 계획고 하나의 (절토 자연 ㎡, 절토 다짐 환산 ㎡, 성토 ㎡) — 본 설계 기본 설계와
|
"""측점 하나 · 계획고 하나의 (절토 자연 ㎡, 절토 다짐 환산 ㎡, 성토 ㎡) — 본 설계 기본 설계와
|
||||||
같은 입력. 반폭에서 비탈이 안 닫히면 넓힌 샘플로 다시 셈(본 설계 `extend_unclosed`)."""
|
같은 입력. 반폭에서 비탈이 안 닫히면 넓힌 샘플로 다시 셈(본 설계 `extend_unclosed`).
|
||||||
|
`standard` = 프로젝트에 저장된 표준 횡단면 설정값(없으면 설정 기본값 · 420)."""
|
||||||
from B06_Section.B06_Section_Engine_Design import compute_cross_design
|
from B06_Section.B06_Section_Engine_Design import compute_cross_design
|
||||||
|
|
||||||
def design(pick: list[dict[str, Any]]) -> dict[str, Any]:
|
def design(pick: list[dict[str, Any]]) -> dict[str, Any]:
|
||||||
@@ -177,6 +187,7 @@ def _station_areas(
|
|||||||
elevation,
|
elevation,
|
||||||
ground_type=GROUND_TYPE,
|
ground_type=GROUND_TYPE,
|
||||||
section_mode=mode,
|
section_mode=mode,
|
||||||
|
standard=standard,
|
||||||
rock_boundary_offset_m=STANDARD_ROCK_BOUNDARY_DEFAULT_OFFSET_M,
|
rock_boundary_offset_m=STANDARD_ROCK_BOUNDARY_DEFAULT_OFFSET_M,
|
||||||
**widen,
|
**widen,
|
||||||
)
|
)
|
||||||
@@ -204,6 +215,7 @@ def _area_curves(
|
|||||||
ground: np.ndarray,
|
ground: np.ndarray,
|
||||||
levels: np.ndarray,
|
levels: np.ndarray,
|
||||||
curves: list[dict[str, Any]] | None = None,
|
curves: list[dict[str, Any]] | None = None,
|
||||||
|
standard: dict[str, Any] | None = None,
|
||||||
) -> tuple[np.ndarray, np.ndarray, np.ndarray]:
|
) -> tuple[np.ndarray, np.ndarray, np.ndarray]:
|
||||||
"""측점 · 계획고 후보마다 (절토 자연 ㎡, 절토 다짐 ㎡, 성토 ㎡) — `AREA_STEP_M` 마디 보간."""
|
"""측점 · 계획고 후보마다 (절토 자연 ㎡, 절토 다짐 ㎡, 성토 ㎡) — `AREA_STEP_M` 마디 보간."""
|
||||||
shape = levels.shape
|
shape = levels.shape
|
||||||
@@ -213,7 +225,9 @@ def _area_curves(
|
|||||||
delta = levels[i] - ground[i]
|
delta = levels[i] - ground[i]
|
||||||
lo = np.floor(delta.min() / AREA_STEP_M) * AREA_STEP_M
|
lo = np.floor(delta.min() / AREA_STEP_M) * AREA_STEP_M
|
||||||
at = np.arange(lo, delta.max() + 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)
|
# 단면적은 계획고에 대해 볼록(대략 2차)이라 직선 보간은 크게 잡힌다 — 단조 3차(PCHIP)
|
||||||
if len(at) > 1:
|
if len(at) > 1:
|
||||||
curve = PchipInterpolator(at, table, axis=0)(delta)
|
curve = PchipInterpolator(at, table, axis=0)(delta)
|
||||||
@@ -277,11 +291,13 @@ def balanced_profile(
|
|||||||
design_speed: int = 20,
|
design_speed: int = 20,
|
||||||
curves: list[dict[str, Any]] | None = None,
|
curves: list[dict[str, Any]] | None = None,
|
||||||
exact: bool = True,
|
exact: bool = True,
|
||||||
|
standard: dict[str, Any] | None = None,
|
||||||
) -> dict[str, Any]:
|
) -> dict[str, Any]:
|
||||||
"""상한(종단기울기 · 종단곡선)을 지키며 절 · 성토를 횡단 ㎥ 로 맞춘 계획선과 토량.
|
"""상한(종단기울기 · 종단곡선)을 지키며 절 · 성토를 횡단 ㎥ 로 맞춘 계획선과 토량.
|
||||||
|
|
||||||
시점 · 종점은 지반고(확정 체인 기본 오프셋 0 과 같음) — 그 길이 없을 때만 풀어 다시 찾는다.
|
시점 · 종점은 지반고(확정 체인 기본 오프셋 0 과 같음) — 그 길이 없을 때만 풀어 다시 찾는다.
|
||||||
`exact` 끄면 보간값 그대로(후보 줄 세우기 — 빠름) · 켜면 고른 선을 바로 셈해 누가토량을 맞춘다.
|
`exact` 끄면 보간값 그대로(후보 줄 세우기 — 빠름) · 켜면 고른 선을 바로 셈해 누가토량을 맞춘다.
|
||||||
|
`standard` = 프로젝트 저장 표준 횡단면 설정값(없으면 설정 기본값).
|
||||||
"""
|
"""
|
||||||
started = time.perf_counter()
|
started = time.perf_counter()
|
||||||
spans = np.diff(chainage)
|
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)
|
offsets = np.arange(-LEVEL_RANGE_M, LEVEL_RANGE_M + 1e-9, LEVEL_STEP_M)
|
||||||
levels = _reference(ground, spans, max_grade)[:, None] + offsets[None, :]
|
levels = _reference(ground, spans, max_grade)[:, None] + offsets[None, :]
|
||||||
cut, cut_c, fill = (
|
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 = (
|
criteria = (
|
||||||
FOREST_ROAD_PROFILE_CRITERIA["design_speed"].get(design_speed)
|
FOREST_ROAD_PROFILE_CRITERIA["design_speed"].get(design_speed)
|
||||||
@@ -312,7 +328,9 @@ def balanced_profile(
|
|||||||
plan = levels[np.arange(len(chainage)), result["path"]]
|
plan = levels[np.arange(len(chainage)), result["path"]]
|
||||||
rows = np.asarray(
|
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))
|
for i in range(len(chainage))
|
||||||
]
|
]
|
||||||
)
|
)
|
||||||
@@ -423,6 +441,7 @@ def route_profile(
|
|||||||
design_speed: int,
|
design_speed: int,
|
||||||
design_curves: list[dict[str, Any]] | None = None,
|
design_curves: list[dict[str, Any]] | None = None,
|
||||||
exact: bool = True,
|
exact: bool = True,
|
||||||
|
standard: dict[str, Any] | None = None,
|
||||||
) -> dict[str, Any]:
|
) -> dict[str, Any]:
|
||||||
"""후보 노선 하나의 균형 종단 — 측점 횡단을 뜨고 `balanced_profile`.
|
"""후보 노선 하나의 균형 종단 — 측점 횡단을 뜨고 `balanced_profile`.
|
||||||
|
|
||||||
@@ -439,6 +458,7 @@ def route_profile(
|
|||||||
design_speed=design_speed,
|
design_speed=design_speed,
|
||||||
curves=curves,
|
curves=curves,
|
||||||
exact=exact,
|
exact=exact,
|
||||||
|
standard=standard,
|
||||||
)
|
)
|
||||||
|
|
||||||
|
|
||||||
@@ -496,6 +516,7 @@ def attach_profile(
|
|||||||
design_speed: int,
|
design_speed: int,
|
||||||
design_curves: list[dict[str, Any]] | None = None,
|
design_curves: list[dict[str, Any]] | None = None,
|
||||||
exact: bool = True,
|
exact: bool = True,
|
||||||
|
standard: dict[str, Any] | None = None,
|
||||||
) -> None:
|
) -> None:
|
||||||
"""후보 비교값에 균형 종단을 붙이고 토공 점수를 그것으로 바꾼다(묶음 캐시에 남음).
|
"""후보 비교값에 균형 종단을 붙이고 토공 점수를 그것으로 바꾼다(묶음 캐시에 남음).
|
||||||
|
|
||||||
@@ -513,6 +534,7 @@ def attach_profile(
|
|||||||
design_speed=design_speed,
|
design_speed=design_speed,
|
||||||
design_curves=design_curves,
|
design_curves=design_curves,
|
||||||
exact=exact,
|
exact=exact,
|
||||||
|
standard=standard,
|
||||||
)
|
)
|
||||||
except ValueError as error:
|
except ValueError as error:
|
||||||
metrics["profile"] = {"error": str(error)}
|
metrics["profile"] = {"error": str(error)}
|
||||||
|
|||||||
@@ -8,14 +8,16 @@
|
|||||||
[초기화]에 안 휩쓸리고, 자료 교체(`purge_output_folders`) · 프로젝트 삭제 때 폴더째 지워진다.
|
[초기화]에 안 휩쓸리고, 자료 교체(`purge_output_folders`) · 프로젝트 삭제 때 폴더째 지워진다.
|
||||||
|
|
||||||
키 = 후보 묶음 지문(`_pool_key` — 지형 · 예상노선 · 계류 · 상한 · 최소 R · 이격 · 고름 · 목표
|
키 = 후보 묶음 지문(`_pool_key` — 지형 · 예상노선 · 계류 · 상한 · 최소 R · 이격 · 고름 · 목표
|
||||||
기울기) + 갈래 · 비중 · 종단 반영 · 설계속도 · `ALGO_VERSION`. 관 지점은 뺀다 — 덮개 수는 꺼낼 때
|
기울기) + 갈래 · 비중 · 종단 반영 · 설계속도 · 표준 횡단면 · `ALGO_VERSION` · 엔진 소스 지문. 관
|
||||||
다시 센다. 추천 결과를 바꾸는 코드 고침은 `ALGO_VERSION` 을 올린다(옛 파일은 다른 키라 안 쓰임).
|
지점은 뺀다 — 덮개 수는 꺼낼 때 다시 센다. 추천 결과를 바꾸는 코드 고침은 `ALGO_VERSION` 을
|
||||||
|
올린다 — 빠뜨려도 엔진 소스 지문이 바뀌어 새로 셈. 옛 판 · 옛 지문 파일은 미리 셈 때 지운다(420).
|
||||||
기본값 · 저장된 제약 결과(`default`)는 「최근 벌」을 따로 센다(351-1).
|
기본값 · 저장된 제약 결과(`default`)는 「최근 벌」을 따로 센다(351-1).
|
||||||
"""
|
"""
|
||||||
|
|
||||||
from __future__ import annotations
|
from __future__ import annotations
|
||||||
|
|
||||||
import asyncio
|
import asyncio
|
||||||
|
import functools
|
||||||
import hashlib
|
import hashlib
|
||||||
import json
|
import json
|
||||||
from contextlib import asynccontextmanager
|
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
|
from common_util.common_util_json import atomic_write_json
|
||||||
|
|
||||||
# 3 = 곡선 MC 일반측점(351-2) · 4 = 다투는 곡선 끔(364-5) · 5 = 후보마다 맞춤(383-1) ·
|
# 3 = 곡선 MC 일반측점(351-2) · 4 = 다투는 곡선 끔(364-5) · 5 = 후보마다 맞춤(383-1) ·
|
||||||
# 6 = 같은 쪽 곡선 합치기 · 거리 한도 대신 직접 잼(406) · 7 = 측점 밖이 늘면 옛 차례 맞춤(406-4)
|
# 6 = 같은 쪽 곡선 합치기 · 거리 한도 대신 직접 잼(406) · 7 = 측점 밖이 늘면 옛 차례 맞춤(406-4) ·
|
||||||
ALGO_VERSION = 7
|
# 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"
|
STORE_DIR = "B05_Profile/route_recommend"
|
||||||
# 종류(갈래 · 지형) 마다 남길 벌 수 — 꺼낼 때 수정시각을 새로 해 최근에 쓴 것이 남음
|
# 종류(갈래 · 지형) 마다 남길 벌 수 — 꺼낼 때 수정시각을 새로 해 최근에 쓴 것이 남음
|
||||||
KEEP_PER_KIND = 2
|
KEEP_PER_KIND = 2
|
||||||
@@ -43,12 +53,25 @@ DEFAULT_MODES = ("gradeline", "earthwork", "gentle", "weighted", "follow")
|
|||||||
|
|
||||||
# 묶음 지문마다 잠금 — 진행 중 같은 묶음(같은 키 포함)은 기다렸다 꺼냄(두 번 셈 막음)
|
# 묶음 지문마다 잠금 — 진행 중 같은 묶음(같은 키 포함)은 기다렸다 꺼냄(두 번 셈 막음)
|
||||||
_LOCKS: dict[str, asyncio.Lock] = {}
|
_LOCKS: dict[str, asyncio.Lock] = {}
|
||||||
|
# 묶음 지문마다 그 묶음을 짤 때 표준 횡단면(`forget_pool_on_standard`)
|
||||||
|
_POOL_STANDARDS: dict[str, str] = {}
|
||||||
|
|
||||||
|
|
||||||
def store_dir(project_root: Path) -> Path:
|
def store_dir(project_root: Path) -> Path:
|
||||||
return Path(project_root) / STORE_DIR
|
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(
|
def request_fingerprint(
|
||||||
expected: list[tuple[float, float]],
|
expected: list[tuple[float, float]],
|
||||||
terrain: Terrain,
|
terrain: Terrain,
|
||||||
@@ -63,6 +86,7 @@ def request_fingerprint(
|
|||||||
target_grade: float | None,
|
target_grade: float | None,
|
||||||
with_profile: bool,
|
with_profile: bool,
|
||||||
design_speed: int,
|
design_speed: int,
|
||||||
|
standard: dict[str, Any] | None = None,
|
||||||
) -> tuple[str, str]:
|
) -> tuple[str, str]:
|
||||||
"""(저장 키, 묶음 지문) — `generate_initial_route` 와 같은 기본값 풀이로 같은 입력 = 같은 키."""
|
"""(저장 키, 묶음 지문) — `generate_initial_route` 와 같은 기본값 풀이로 같은 입력 = 같은 키."""
|
||||||
pool = _pool_key(
|
pool = _pool_key(
|
||||||
@@ -74,7 +98,15 @@ def request_fingerprint(
|
|||||||
*resolve_inputs(max_grade, stream_offset_m, grade_smooth_m, target_grade),
|
*resolve_inputs(max_grade, stream_offset_m, grade_smooth_m, target_grade),
|
||||||
).hex()
|
).hex()
|
||||||
used = resolve_weights(mode, weights)
|
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()
|
key = hashlib.blake2b((pool + repr(tail)).encode(), digest_size=16).hexdigest()
|
||||||
return key, pool
|
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:
|
def read_stored(project_root: Path, key: str) -> dict[str, Any] | None:
|
||||||
"""저장본 한 벌 — 없거나 못 읽거나 판이 다르면 None."""
|
"""저장본 한 벌 — 없거나 못 읽거나 판 · 소스 지문이 다르면 None."""
|
||||||
path = store_dir(project_root) / f"{key}.json"
|
path = store_dir(project_root) / f"{key}.json"
|
||||||
try:
|
try:
|
||||||
data = json.loads(path.read_text(encoding="utf-8"))
|
data = json.loads(path.read_text(encoding="utf-8"))
|
||||||
except (OSError, ValueError):
|
except (OSError, ValueError):
|
||||||
return None
|
return None
|
||||||
if not isinstance(data, dict) or data.get("version") != ALGO_VERSION:
|
if not _current(data):
|
||||||
return None
|
return None
|
||||||
try:
|
try:
|
||||||
path.touch() # 최근에 쓴 것 = 정리 때 남김
|
path.touch() # 최근에 쓴 것 = 정리 때 남김
|
||||||
@@ -134,6 +174,7 @@ def write_stored(
|
|||||||
path,
|
path,
|
||||||
{
|
{
|
||||||
"version": ALGO_VERSION,
|
"version": ALGO_VERSION,
|
||||||
|
"source": source_fingerprint(),
|
||||||
"key": key,
|
"key": key,
|
||||||
"pool": pool,
|
"pool": pool,
|
||||||
"default": default,
|
"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)
|
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
|
@asynccontextmanager
|
||||||
async def pool_lock(pool: str) -> AsyncIterator[None]:
|
async def pool_lock(pool: str) -> AsyncIterator[None]:
|
||||||
# ponytail: 잠금은 지우지 않음 — 입력 묶음 수만큼(작음) · 다중 프로세스면 파일 잠금으로
|
# ponytail: 잠금은 지우지 않음 — 입력 묶음 수만큼(작음) · 다중 프로세스면 파일 잠금으로
|
||||||
|
|||||||
@@ -627,6 +627,7 @@ async def replan_route(project_id: UUID, body: RouteSaveRequest) -> dict[str, An
|
|||||||
choice.mode,
|
choice.mode,
|
||||||
choice.weights.model_dump() if choice.weights else None,
|
choice.weights.model_dump() if choice.weights else None,
|
||||||
(choice.criteria.model_dump(exclude_none=True) or None) if choice.criteria else None,
|
(choice.criteria.model_dump(exclude_none=True) or None) if choice.criteria else None,
|
||||||
|
request.balanced_profile, # 「종단 반영」 — 다시 열면 되살림(420)
|
||||||
)
|
)
|
||||||
if choice
|
if choice
|
||||||
else await asyncio.to_thread(read_initial_choice, project_root)
|
else await asyncio.to_thread(read_initial_choice, project_root)
|
||||||
|
|||||||
@@ -6,6 +6,7 @@
|
|||||||
|
|
||||||
import asyncio
|
import asyncio
|
||||||
import functools
|
import functools
|
||||||
|
import json
|
||||||
import logging
|
import logging
|
||||||
import time
|
import time
|
||||||
from pathlib import Path
|
from pathlib import Path
|
||||||
@@ -44,17 +45,21 @@ async def compute_initial(
|
|||||||
"""같은 입력이면 저장본을 꺼내고 없으면 셈해 더한다(PLAN 327) · 덮개 수는 늘 지금 관으로.
|
"""같은 입력이면 저장본을 꺼내고 없으면 셈해 더한다(PLAN 327) · 덮개 수는 늘 지금 관으로.
|
||||||
|
|
||||||
응답 `stored` = 저장본에서 꺼냈나. 진행 중인 같은 묶음은 기다렸다 꺼낸다(두 번 셈 막음).
|
응답 `stored` = 저장본에서 꺼냈나. 진행 중인 같은 묶음은 기다렸다 꺼낸다(두 번 셈 막음).
|
||||||
|
`recompute` = [재계산] — 묶음 잠금 안에서 이 묶음 저장본 · 메모리 묶음을 버리고 새로 셈(420).
|
||||||
"""
|
"""
|
||||||
from B05_Profile.B05_Profile_Engine_Grade import legal_grade_limit_pct
|
from B05_Profile.B05_Profile_Engine_Grade import legal_grade_limit_pct
|
||||||
from B05_Profile.B05_Profile_Engine_RouteInitial import (
|
from B05_Profile.B05_Profile_Engine_RouteInitial import (
|
||||||
generate_initial_route,
|
generate_initial_route,
|
||||||
load_terrain_and_streams,
|
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 (
|
from B05_Profile.B05_Profile_Engine_RouteInitial_Profile import (
|
||||||
cover_shortfall,
|
cover_shortfall,
|
||||||
project_pipes,
|
project_pipes,
|
||||||
)
|
)
|
||||||
from B05_Profile.B05_Profile_Engine_RouteInitial_Store import (
|
from B05_Profile.B05_Profile_Engine_RouteInitial_Store import (
|
||||||
|
drop_pool_results,
|
||||||
|
forget_pool_on_standard,
|
||||||
is_default,
|
is_default,
|
||||||
pool_lock,
|
pool_lock,
|
||||||
read_stored,
|
read_stored,
|
||||||
@@ -115,6 +120,8 @@ async def compute_initial(
|
|||||||
"target_grade": None if target_pct is None else target_pct / 100.0,
|
"target_grade": None if target_pct is None else target_pct / 100.0,
|
||||||
"design_speed": design_speed or 20,
|
"design_speed": design_speed or 20,
|
||||||
"with_profile": request.with_profile,
|
"with_profile": request.with_profile,
|
||||||
|
# 균형 종단 · 횡단 토공 표준 횡단면 = 프로젝트 저장 패널값(420)
|
||||||
|
"standard": await project_standard(project_id),
|
||||||
}
|
}
|
||||||
started = time.perf_counter()
|
started = time.perf_counter()
|
||||||
try:
|
try:
|
||||||
@@ -126,6 +133,10 @@ async def compute_initial(
|
|||||||
functools.partial(request_fingerprint, route, terrain, streams, **inputs)
|
functools.partial(request_fingerprint, route, terrain, streams, **inputs)
|
||||||
)
|
)
|
||||||
async with pool_lock(pool_key):
|
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)
|
stored = await asyncio.to_thread(read_stored, project_root, key)
|
||||||
if stored is not None:
|
if stored is not None:
|
||||||
result = stored["result"]
|
result = stored["result"]
|
||||||
@@ -136,12 +147,14 @@ async def compute_initial(
|
|||||||
generate_initial_route, route, terrain, streams, pipes=None, **inputs
|
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(
|
await asyncio.to_thread(
|
||||||
write_stored,
|
write_stored,
|
||||||
project_root,
|
project_root,
|
||||||
key,
|
key,
|
||||||
pool_key,
|
pool_key,
|
||||||
request.model_dump(),
|
request.model_dump(exclude={"recompute"}),
|
||||||
result,
|
result,
|
||||||
is_default(
|
is_default(
|
||||||
request.mode,
|
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:
|
async def precompute_defaults(project_id: UUID) -> None:
|
||||||
"""체인 끝 예약(PLAN 327-2 · 351-1) — 기본 다섯 알약 + 저장된 제약(initial_choice)의 다섯.
|
"""체인 끝 예약(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 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()
|
started = time.perf_counter()
|
||||||
variants: list[tuple[Any, Any]] = [(None, None)]
|
variants: list[tuple[Any, Any]] = [(None, None)]
|
||||||
@@ -196,6 +232,9 @@ async def precompute_defaults(project_id: UUID) -> None:
|
|||||||
except LookupError:
|
except LookupError:
|
||||||
paths = None
|
paths = None
|
||||||
if paths is not 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])
|
saved = await asyncio.to_thread(read_initial_choice, paths[0])
|
||||||
weights = saved["weights"] if saved["logic"] == "weighted" else None
|
weights = saved["weights"] if saved["logic"] == "weighted" else None
|
||||||
if saved["criteria"] or (weights and weights != DEFAULT_WEIGHTS):
|
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:
|
for mode in modes:
|
||||||
if progress is not None:
|
if progress is not None:
|
||||||
progress["mode"] = mode
|
progress["mode"] = mode
|
||||||
# 화면과 같은 몸통 — 설계 지형 · 비중은 비중 갈래만 · 추적만 종단 끔(`withProfileOf`)
|
# 화면과 같은 몸통 — 설계 지형 · 비중은 비중 갈래만 · 「종단 반영」 기본 켬(420)
|
||||||
request = replan.RouteInitialRequest(
|
request = replan.RouteInitialRequest(
|
||||||
mode=mode,
|
mode=mode,
|
||||||
with_profile=mode != "follow",
|
with_profile=True,
|
||||||
criteria=criteria,
|
criteria=criteria,
|
||||||
weights=weights if mode == "weighted" else None,
|
weights=weights if mode == "weighted" else None,
|
||||||
)
|
)
|
||||||
@@ -416,6 +455,7 @@ async def balanced_profile_of(
|
|||||||
pool = get_db_pool()
|
pool = get_db_pool()
|
||||||
async with pool.acquire() as connection:
|
async with pool.acquire() as connection:
|
||||||
selection = await get_surface_confirmation_params(connection, str(project_id))
|
selection = await get_surface_confirmation_params(connection, str(project_id))
|
||||||
|
standard = await project_standard(project_id) # 추천 창과 같은 저장 패널값(420)
|
||||||
try:
|
try:
|
||||||
# 측점 생성과 같은 1 m 쌍선형 표본(추천 점수의 2 m 비용면 아님)
|
# 측점 생성과 같은 1 m 쌍선형 표본(추천 점수의 2 m 비용면 아님)
|
||||||
surface = await asyncio.to_thread(
|
surface = await asyncio.to_thread(
|
||||||
@@ -433,6 +473,7 @@ async def balanced_profile_of(
|
|||||||
max_grade=max_grade,
|
max_grade=max_grade,
|
||||||
design_speed=design_speed or 20,
|
design_speed=design_speed or 20,
|
||||||
design_curves=design_curves,
|
design_curves=design_curves,
|
||||||
|
standard=standard,
|
||||||
)
|
)
|
||||||
)
|
)
|
||||||
except (FileNotFoundError, ValueError):
|
except (FileNotFoundError, ValueError):
|
||||||
|
|||||||
@@ -367,6 +367,8 @@ class RouteInitialRequest(BaseModel):
|
|||||||
with_profile: bool = True
|
with_profile: bool = True
|
||||||
# 분석 지형(PLAN 133-1) — None = 지금 설계 지형 · [확인] 때 설계 지형이 됨(133-2).
|
# 분석 지형(PLAN 133-1) — None = 지금 설계 지형 · [확인] 때 설계 지형이 됨(133-2).
|
||||||
terrain: Literal["las", "sheet"] | None = None
|
terrain: Literal["las", "sheet"] | None = None
|
||||||
|
# [재계산](420) — 이 후보 묶음 저장본 · 메모리 묶음을 지우고 새로 셈. 저장본 request 엔 안 적힘.
|
||||||
|
recompute: bool = False
|
||||||
|
|
||||||
|
|
||||||
class RouteReplanRequest(BaseModel):
|
class RouteReplanRequest(BaseModel):
|
||||||
|
|||||||
@@ -96,7 +96,16 @@ def test_no_overwrite(tmp_path: Path) -> None:
|
|||||||
assert not write_stored(tmp_path, "k", "p", {}, {"n": 2}, False)
|
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"))
|
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 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:
|
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:
|
class _Calls:
|
||||||
def __init__(self) -> None:
|
def __init__(self) -> None:
|
||||||
self.count = 0
|
self.count = 0
|
||||||
|
self.inputs: list[dict] = []
|
||||||
|
self.standard: dict | None = None
|
||||||
|
|
||||||
|
|
||||||
@pytest.fixture
|
@pytest.fixture
|
||||||
@@ -141,20 +152,26 @@ def wired(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> _Calls:
|
|||||||
async def surface_params(_connection, _pid):
|
async def surface_params(_connection, _pid):
|
||||||
return {}
|
return {}
|
||||||
|
|
||||||
|
async def project_standard(_pid):
|
||||||
|
return calls.standard
|
||||||
|
|
||||||
class _Pool:
|
class _Pool:
|
||||||
@asynccontextmanager
|
@asynccontextmanager
|
||||||
async def acquire(self):
|
async def acquire(self):
|
||||||
yield None
|
yield None
|
||||||
|
|
||||||
def generate(expected, terrain, streams, *, pipes, **_inputs):
|
def generate(expected, terrain, streams, *, pipes, **inputs):
|
||||||
assert pipes is None # 관은 키 밖 — 덮개 없이 셈해 남김
|
assert pipes is None # 관은 키 밖 — 덮개 없이 셈해 남김
|
||||||
calls.count += 1
|
calls.count += 1
|
||||||
|
calls.inputs.append(inputs)
|
||||||
return {
|
return {
|
||||||
"planned": [[x, y] for x, y in expected],
|
"planned": [[x, y] for x, y in expected],
|
||||||
"metrics": {
|
"metrics": {
|
||||||
"length_m": 38.0,
|
"length_m": 38.0,
|
||||||
"offset_avg_m": 0.0,
|
"offset_avg_m": 0.0,
|
||||||
"profile": {
|
"profile": {
|
||||||
|
"cut_m3": 10.0,
|
||||||
|
"fill_m3": 4.0,
|
||||||
"stations_m": [0.0, 38.0],
|
"stations_m": [0.0, 38.0],
|
||||||
"plan_m": [1.0, 1.0],
|
"plan_m": [1.0, 1.0],
|
||||||
"ground_m": [0.0, 0.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, "_road_settings", road_settings)
|
||||||
monkeypatch.setattr(replan, "_vertices_of", lambda _path: [list(p) for p in EXPECTED])
|
monkeypatch.setattr(replan, "_vertices_of", lambda _path: [list(p) for p in EXPECTED])
|
||||||
monkeypatch.setattr(router, "get_db_pool", lambda: _Pool())
|
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, "get_surface_confirmation_params", surface_params)
|
||||||
monkeypatch.setattr(surface, "analysis_terrain_params", lambda selection, _t: selection)
|
monkeypatch.setattr(surface, "analysis_terrain_params", lambda selection, _t: selection)
|
||||||
monkeypatch.setattr(grade, "legal_grade_limit_pct", lambda *_a: 14.0)
|
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 sorted(item["request"]["mode"] for item in saved) == sorted(DEFAULT_MODES)
|
||||||
assert all(item["default"] for item in saved)
|
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})
|
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
|
assert hit["stored"] and follow["stored"] and wired.count == 5
|
||||||
|
|||||||
@@ -90,6 +90,7 @@ def test_모달이_여는_갈래_먼저(monkeypatch) -> None:
|
|||||||
|
|
||||||
async def compute(_pid, request):
|
async def compute(_pid, request):
|
||||||
asked.append(request.mode)
|
asked.append(request.mode)
|
||||||
|
assert request.with_profile # 다섯 갈래 모두 종단 켬(420)
|
||||||
return {}
|
return {}
|
||||||
|
|
||||||
monkeypatch.setattr(router.replan, "_project_paths", paths)
|
monkeypatch.setattr(router.replan, "_project_paths", paths)
|
||||||
|
|||||||
@@ -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") :]
|
||||||
@@ -321,6 +321,7 @@ def test_initial_choice_roundtrip(tmp_path) -> None:
|
|||||||
"logic": "follow",
|
"logic": "follow",
|
||||||
"weights": {"expected": 100.0, "earthwork": 0.0, "grade": 0.0},
|
"weights": {"expected": 100.0, "earthwork": 0.0, "grade": 0.0},
|
||||||
"criteria": None,
|
"criteria": None,
|
||||||
|
"with_profile": True,
|
||||||
}
|
}
|
||||||
for logic in ("gradeline", "safe", "crossings", "cost"): # 새 갈래도 적고 다시 읽음(23-5)
|
for logic in ("gradeline", "safe", "crossings", "cost"): # 새 갈래도 적고 다시 읽음(23-5)
|
||||||
write_initial_choice(tmp_path, logic, None, {"target_grade_pct": 10.0})
|
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",
|
"logic": "weighted",
|
||||||
"weights": {"expected": 60.0, "earthwork": 20.0, "grade": 20.0},
|
"weights": {"expected": 60.0, "earthwork": 20.0, "grade": 20.0},
|
||||||
"criteria": None,
|
"criteria": None,
|
||||||
|
"with_profile": True,
|
||||||
}
|
}
|
||||||
# 사용자가 화면 칸에 겹쳐 넣은 값만(빈 칸은 안 담김) — 남으면 다시 열 때 그 칸만 채운다.
|
# 사용자가 화면 칸에 겹쳐 넣은 값만(빈 칸은 안 담김) — 남으면 다시 열 때 그 칸만 채운다.
|
||||||
write_initial_choice(
|
write_initial_choice(
|
||||||
@@ -340,6 +342,7 @@ def test_initial_choice_roundtrip(tmp_path) -> None:
|
|||||||
"logic": "earthwork",
|
"logic": "earthwork",
|
||||||
"weights": {"expected": 0.0, "earthwork": 100.0, "grade": 0.0},
|
"weights": {"expected": 0.0, "earthwork": 100.0, "grade": 0.0},
|
||||||
"criteria": {"stream_offset_m": 30.0, "grade_smooth_m": 2.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")
|
(tmp_path / "B05_Profile" / "route" / "initial_choice.json").write_text("{", encoding="utf-8")
|
||||||
assert read_initial_choice(tmp_path)["logic"] == "follow"
|
assert read_initial_choice(tmp_path)["logic"] == "follow"
|
||||||
|
|||||||
Reference in New Issue
Block a user