Files
Aislo/resources/tester/test_route_initial.py
T
eomsangdonandClaude Opus 5.5 c14717d0c2 feat(B05): 비중 = 탐색 비용의 가중합 · 짧은 노선 · 흙 균형 갈래 · 같은 노선 알림(PLAN 23-4)
- 비중 계산을 후보 고르기에서 탐색 한 번으로 — 몫(이격 · 토공 · 기울기)을 끝 노선 셋(추적 · 토공 최소 · 기울기 순한) 사이 범위로 크기 맞춤 · 제약은 (1 − 예상 몫)만큼 벌점(막지 않음) · 넘은 곳은 「비중 계산」 사유로 목록
- 끝값 100/0/0 · 0/100/0 · 0/0/100 = 추적 · 토공 최소 · 기울기 순한 그 선 · 예상 99 → 추적에서 5.0 m
- 새 갈래 short(연장 최소) · balance(|절토 − 성토| 최소) · 비교값 surplus_m3(흙 남음)
- 응답 similar(10 m 안인 다른 갈래 · 가장 먼 거리) · band_width_m(제약 안 띠 폭 어림)
- 후보 묶음을 새 모듈 _Pool 로 · 토공 ↔ 기울기 쓸기 11 벌을 묶음에 넣어 갈래가 사이 비중보다 제 목표에서 좋게
- 같은 상한 탐색은 간선 틀을 다시 씀 · 같은 노드는 다시 안 잼(첫 요청 엔진 7.1 → 2.2 초)
- 시험 test_route_initial 27

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_019ACSGaHdLgnkEkoA4LDtMU
2026-09-30 08:00:42 +09:00

449 lines
19 KiB
Python
Raw Blame History

This file contains ambiguous Unicode characters
This file contains Unicode characters that might be confused with other characters. If you think that this is intentional, you can safely ignore this warning. Use the Escape button to reveal them.
"""초기 계획노선 갈래(PLAN 20장) — `B05_Profile_Engine_RouteInitial` 합성 지형 시험.
./venv/Scripts/python.exe -m pytest resources/tester/test_route_initial.py -q
공통 제약(종단기울기 상한 · 최소 곡선반지름 · 계류 이격)은 넘지 못함 — 지형상 불가능할 때만
`metrics.violations` 에 측점 · 값 · 사유로 뜨고 `over_limit.total` 은 그 수와 같다.
예상노선 추적(follow)은 전처리 초기값과 같은 선 — 제약 탐색 없이 넘은 곳만 목록(PLAN 23-1).
갈래는 제 목표 비교값이 가장 좋은 후보 — 실프로젝트(랩탑_보조 · 읽기만)에서 갈래가 갈림.
"""
from pathlib import Path
import numpy as np
import pytest
from shapely.geometry import LineString
from B05_Profile.B05_Profile_Engine_RouteInitial import (
MODES,
PENDING,
STREAM_OFFSET_M,
Terrain,
generate_initial_route,
read_initial_choice,
resolve_weights,
stream_lines,
write_initial_choice,
)
from B05_Profile.B05_Profile_Engine_RouteInitial_Metrics import grade_line, station_label
from common_util.common_util_route_polyline import build_planned_polyline
SEARCHED = [mode for mode in MODES if mode != "follow"]
# 제약을 지키는 갈래(후보 묶음에서 고름) — 비중은 제약을 벌점으로만(PLAN 23-4a)
BRANCHES = ["earthwork", "gentle", "short", "balance"]
GRADE = 0.14
RADIUS = 12.0
def _terrain(z_of) -> Terrain:
x = np.arange(0.0, 802.0, 2.0)
y = np.arange(0.0, 402.0, 2.0)
xx, yy = np.meshgrid(x, y)
return Terrain(x, y, z_of(xx, yy), np.ones(xx.shape, dtype=bool))
def _line(y: float) -> list[tuple[float, float]]:
return [(float(x), y) for x in np.arange(50.0, 752.0, 4.0)]
def _run(terrain: Terrain, mode: str, streams=(), weights=None) -> dict:
return generate_initial_route(
_line(200.0),
terrain,
list(streams),
mode=mode,
weights=weights,
max_grade=GRADE,
min_radius_m=RADIUS,
)
def _kept(result: dict) -> None:
"""제약을 지켰거나, 못 지킨 곳이 목록에 있고 수가 같음."""
metrics = result["metrics"]
assert metrics["over_limit"]["total"] == len(metrics["violations"])
if not metrics["violations"]:
assert metrics["max_grade_pct"] <= GRADE * 100.0 + 1e-6
assert result["relaxed"] is False
for curve in result["curves"]:
assert curve["radius_m"] >= RADIUS - 1e-6 or curve["violations"]
@pytest.fixture(scope="module")
def hill() -> Terrain:
"""가운데 솟은 언덕(높이 40 m) — 곧은 예상노선이 꼭대기(측면 27 %)를 넘는다."""
return _terrain(
lambda xx, yy: 40.0 * np.exp(-((xx - 400.0) ** 2 + (yy - 200.0) ** 2) / 90.0**2)
)
@pytest.mark.parametrize("mode", BRANCHES)
def test_every_mode_keeps_grade_limit_on_hill(hill: Terrain, mode: str) -> None:
result = _run(hill, mode)
_kept(result)
metrics = result["metrics"]
assert metrics["over_limit"]["total"] == 0
assert metrics["max_grade_pct"] <= GRADE * 100.0 + 1e-6
# 기점 · 종점은 옮기지 않고 복도(100 m) 밖으로 안 나감
assert result["planned"][0] == [50.0, 200.0]
assert result["planned"][-1] == [750.0, 200.0]
assert metrics["offset_max_m"] <= result["criteria"]["corridor_m"] + 4.0
def test_follow_is_initial_polyline_and_lists_violations(hill: Terrain) -> None:
"""추종 = 전처리 초기값 규칙(`build_planned_polyline` 기본값) — 언덕을 넘는 곳은 목록에."""
result = _run(hill, "follow")
initial = build_planned_polyline(_line(200.0), min_radius_m=RADIUS)
assert result["planned"] == [[round(x, 4), round(y, 4)] for x, y in initial.vertices]
assert result["nodes"] == [node.as_dict() for node in initial.nodes]
assert result["relaxed"] is False and result["terrain_bound_pct"] is None
grades = [v for v in result["metrics"]["violations"] if v["kind"] == "grade"]
assert grades and all("예상노선 추적" in v["reason"] for v in grades)
assert result["metrics"]["over_limit"]["total"] == len(result["metrics"]["violations"])
def _earth(result: dict) -> float:
return result["metrics"]["cut_m3"] + result["metrics"]["fill_m3"]
def _check_branch_goals(runs: dict) -> None:
"""토공 최소 = 절성토 · 기울기 순한 = 최대 기울기 · 짧은 노선 = 연장 · 흙 균형 = |절토 − 성토|
가 갈래(같은 후보 묶음) 가운데 가장 좋음 · 흙 남음 = 절토 − 성토."""
branches = [runs[mode] for mode in BRANCHES]
assert all(_earth(runs["earthwork"]) <= _earth(r) + 1e-6 for r in branches)
top = runs["gentle"]["metrics"]["max_grade_pct"]
assert all(top <= r["metrics"]["max_grade_pct"] + 1e-6 for r in branches)
shortest = runs["short"]["metrics"]["length_m"]
assert all(shortest <= r["metrics"]["length_m"] + 1e-6 for r in branches)
surplus = abs(runs["balance"]["metrics"]["surplus_m3"])
assert all(surplus <= abs(r["metrics"]["surplus_m3"]) + 0.1 for r in branches)
for result in runs.values():
metrics = result["metrics"]
assert abs(metrics["surplus_m3"] - (metrics["cut_m3"] - metrics["fill_m3"])) <= 0.11
def test_each_branch_best_at_own_goal(hill: Terrain) -> None:
_check_branch_goals({mode: _run(hill, mode) for mode in MODES})
# 비중 끝값 · 단조(PLAN 23-3e) — 예상 · 토공 · 기울기 한쪽 100 이면 그 갈래와 같은 선
ENDS = {
"follow": {"expected": 100, "earthwork": 0, "grade": 0},
"earthwork": {"expected": 0, "earthwork": 100, "grade": 0},
"gentle": {"expected": 0, "earthwork": 0, "grade": 100},
}
# 예상 100 → 0(나머지 반반) · 기울기 100 → 토공 100
TOWARD_OTHERS = [
(100, 0, 0),
(90, 5, 5),
(75, 12.5, 12.5),
(50, 25, 25),
(25, 37.5, 37.5),
(0, 50, 50),
]
TOWARD_EARTH = [(0, 0, 100), (0, 25, 75), (0, 50, 50), (0, 75, 25), (0, 100, 0)]
def _gap(first: dict, second: dict) -> float:
return LineString(first["planned"]).hausdorff_distance(LineString(second["planned"]))
def _check_weight_ends(run) -> None:
for mode, weights in ENDS.items():
end, own = run("weighted", weights), run(mode)
assert _gap(end, own) < 0.01, (mode, _gap(end, own))
assert end["nodes"] == own["nodes"]
assert end["metrics"]["length_m"] == own["metrics"]["length_m"]
def _check_weight_monotone(run) -> None:
"""예상 몫을 줄이면 예상노선 이격이 늘기만 함(비중 탐색 = 가중합 최소의 성질) · 토공 ↔ 기울기
끝값(갈래)은 그 사이 어느 비중보다 제 목표에서 좋음(쓸기가 후보 묶음에 들어 있음)."""
def metrics(steps):
keys = ("expected", "earthwork", "grade")
return [run("weighted", dict(zip(keys, step)))["metrics"] for step in steps]
offsets = [m["offset_avg_m"] for m in metrics(TOWARD_OTHERS)]
assert offsets == sorted(offsets), offsets
earth = metrics(TOWARD_EARTH)
cuts = [m["cut_m3"] + m["fill_m3"] for m in earth]
tops = [m["max_grade_pct"] for m in earth]
assert cuts[-1] <= min(cuts) + 1e-6, cuts # 끝 0/100/0 = 토공 최소
assert tops[0] <= min(tops) + 1e-6, tops # 끝 0/0/100 = 기울기 순한
def test_weight_ends_match_branches(hill: Terrain) -> None:
_check_weight_ends(lambda mode, weights=None: _run(hill, mode, weights=weights))
def test_weight_moves_monotone(hill: Terrain) -> None:
_check_weight_monotone(lambda mode, weights=None: _run(hill, mode, weights=weights))
def test_follow_stays_on_expected_where_allowed() -> None:
"""고른 경사(2 %) — 추종은 예상노선을 그대로(격자 칸 반 안쪽)."""
ramp = _terrain(lambda xx, yy: 0.02 * xx)
result = _run(ramp, "follow")
_kept(result)
assert result["metrics"]["offset_max_m"] <= 3.0
def test_weighted_expected_share_pulls_route_back(hill: Terrain) -> None:
near = _run(hill, "weighted", weights={"expected": 90, "earthwork": 5, "grade": 5})
far = _run(hill, "weighted", weights={"expected": 10, "earthwork": 45, "grade": 45})
assert near["metrics"]["offset_avg_m"] <= far["metrics"]["offset_avg_m"]
def test_stream_offset_is_hard() -> None:
"""평지 · 예상노선 20 m 옆을 나란히 흐르는 소하천 — 모든 갈래가 버퍼를 비켜 감."""
flat = _terrain(lambda xx, yy: np.zeros_like(xx))
stream = [[(0.0, 220.0), (800.0, 220.0)]]
for mode in BRANCHES:
result = _run(flat, mode, streams=stream)
_kept(result)
assert result["metrics"]["over_limit"]["stream"] == 0
assert result["metrics"]["offset_avg_m"] > STREAM_OFFSET_M - 20.0 - 30.0
def test_stream_crossing_is_allowed() -> None:
flat = _terrain(lambda xx, yy: np.zeros_like(xx))
crossing = [[(400.0, 0.0), (400.0, 400.0)]]
result = _run(flat, "follow", streams=crossing)
_kept(result)
assert result["metrics"]["over_limit"]["total"] == 0
def test_terrain_impossible_is_listed() -> None:
"""복도 전체를 가로지르는 20 m 벽 — 어떤 길도 상한을 넘음 → 목록 · 사유 · 막은 기울기."""
wall = _terrain(lambda xx, yy: np.where(xx > 400.0, 20.0, 0.0))
result = _run(wall, "earthwork")
_kept(result)
assert result["relaxed"] is True
assert result["terrain_bound_pct"] > GRADE * 100.0
grades = [v for v in result["metrics"]["violations"] if v["kind"] == "grade"]
assert grades and all("지형상 불가능" in v["reason"] for v in grades)
assert all(v["station"].startswith("No.") for v in grades)
def test_grade_line_stays_within_limit() -> None:
ground = np.array([0.0, 0.0, 5.0, 10.0, 10.0, 2.0, 0.0])
line = grade_line(ground, 2.0, GRADE)
assert np.all(np.abs(np.diff(line)) <= GRADE * 2.0 + 1e-9)
assert station_label(125.0) == "No.6+5.0"
def test_weights_and_stream_filter() -> None:
assert resolve_weights("weighted", None) == {"expected": 50.0, "earthwork": 25.0, "grade": 25.0}
assert resolve_weights("gentle", {"expected": 99})["grade"] == 100.0
assert set(MODES) == {"follow", "earthwork", "gentle", "weighted", "short", "balance"}
assert resolve_weights("short", {"expected": 99})["expected"] == 0.0
line = {"type": "LineString", "coordinates": [[0, 0], [1, 1]]}
features = [
{"properties": {"구분": "세류"}, "geometry": line},
{"properties": {"구분": "소하천"}, "geometry": line},
]
assert len(stream_lines(features)) == 1
def test_unknown_mode_rejected(hill: Terrain) -> None:
with pytest.raises(ValueError):
_run(hill, "shortest")
def test_criteria_confirmed(hill: Terrain) -> None:
"""기준값 확정(2026-09-29 사용자 결정) — `pending` 은 이제 늘 빈 목록."""
assert _run(hill, "follow")["criteria"]["pending"] == [] == list(PENDING)
def test_criteria_overrides_change_result(hill: Terrain) -> None:
"""화면 제약 칸(계류 이격 · 지반 고름)을 겹치면 그 값으로 셈하고, 안 겹치면 기본값."""
default = generate_initial_route(
_line(200.0),
hill,
[],
mode="follow",
weights=None,
max_grade=GRADE,
min_radius_m=RADIUS,
)
assert default["criteria"]["stream_offset_m"] == STREAM_OFFSET_M
overridden = generate_initial_route(
_line(200.0),
hill,
[],
mode="follow",
weights=None,
max_grade=GRADE,
min_radius_m=RADIUS,
stream_offset_m=30.0,
grade_smooth_m=2.0,
)
assert overridden["criteria"]["stream_offset_m"] == 30.0
assert overridden["criteria"]["grade_smooth_m"] == 2.0
assert overridden["criteria"]["max_grade_pct"] == default["criteria"]["max_grade_pct"]
_kept(overridden)
def test_initial_choice_roundtrip(tmp_path) -> None:
"""[확인] 때 적은 현재 초기값(갈래 · 비중 · 제약 겹침)을 다시 읽음 · 없거나 깨지면 추종."""
assert read_initial_choice(tmp_path) == {
"logic": "follow",
"weights": {"expected": 100.0, "earthwork": 0.0, "grade": 0.0},
"criteria": None,
}
write_initial_choice(tmp_path, "weighted", {"expected": 60, "earthwork": 20, "grade": 20})
assert read_initial_choice(tmp_path) == {
"logic": "weighted",
"weights": {"expected": 60.0, "earthwork": 20.0, "grade": 20.0},
"criteria": None,
}
# 사용자가 화면 칸에 겹쳐 넣은 값만(빈 칸은 안 담김) — 남으면 다시 열 때 그 칸만 채운다.
write_initial_choice(
tmp_path, "earthwork", None, {"stream_offset_m": 30.0, "grade_smooth_m": 2.0}
)
assert read_initial_choice(tmp_path) == {
"logic": "earthwork",
"weights": {"expected": 0.0, "earthwork": 100.0, "grade": 0.0},
"criteria": {"stream_offset_m": 30.0, "grade_smooth_m": 2.0},
}
(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)["criteria"] is None
def test_request_accepts_logic_name() -> None:
from B05_Profile.B05_Profile_Router_Replan import RouteInitialRequest, RouteReplanRequest
assert RouteInitialRequest.model_validate({"logic": "gentle"}).mode == "gentle"
assert RouteInitialRequest.model_validate({"mode": "earthwork"}).mode == "earthwork"
assert RouteInitialRequest.model_validate({"logic": "short"}).mode == "short"
assert RouteInitialRequest.model_validate({"logic": "balance"}).mode == "balance"
replan = RouteReplanRequest.model_validate({"vertices": [], "initial": {"logic": "weighted"}})
assert replan.initial is not None and replan.initial.mode == "weighted"
def test_request_accepts_criteria_overrides() -> None:
"""화면 제약 칸 — 비운 칸은 None(서버 기본값), 채운 칸만 값이 실린다(계획서 20)."""
from B05_Profile.B05_Profile_Router_Replan import RouteInitialRequest, RouteReplanRequest
request = RouteInitialRequest.model_validate(
{"mode": "earthwork", "criteria": {"stream_offset_m": 30, "grade_smooth_m": 2}}
)
assert request.criteria is not None
assert request.criteria.stream_offset_m == 30
assert request.criteria.grade_smooth_m == 2
assert request.criteria.max_grade_pct is None
assert request.criteria.min_radius_m is None
replan = RouteReplanRequest.model_validate(
{"vertices": [], "initial": {"mode": "weighted", "criteria": {"max_grade_pct": 12}}}
)
assert replan.initial is not None and replan.initial.criteria is not None
assert replan.initial.criteria.max_grade_pct == 12
# ── 실프로젝트(랩탑_보조) — 읽기만 · storage 가 없으면 건너뜀 ──────────────────────────
PROJECT = Path(__file__).resolve().parents[2] / "storage/1/3/936be972-11bc-46c2-8bf3-b15d8de7df0d"
SURFACE = PROJECT / "B05_Profile/route/cost_surface_classification_dtm_smooth.npz"
@pytest.fixture(scope="module")
def project_runs() -> dict:
"""네 갈래 결과 — 비용면 캐시 · 계류 · 예상노선을 읽기만(파일 안 씀). 기준 14 % · R12."""
if not SURFACE.is_file():
pytest.skip("랩탑_보조 storage 없음")
from pyproj import Transformer
from B04_PreProcess.B04_PreProcess_Router_Watershed import (
STREAM_FILE,
_load_features,
_reproject_features,
)
from common_util.common_util_crs import resolve_project_crs
from common_util.common_util_route_geometry import (
expected_route_csv_path,
read_planned_route_csv,
)
cached = np.load(SURFACE, allow_pickle=False)
terrain = Terrain(cached["x"], cached["y"], cached["z"], cached["valid_mask"])
to_metric = Transformer.from_crs("EPSG:4326", resolve_project_crs(PROJECT), always_xy=True)
features = _load_features(PROJECT / "B04_PreProcess" / "processed", STREAM_FILE)
streams = stream_lines(_reproject_features(features, to_metric))
route = read_planned_route_csv(expected_route_csv_path(PROJECT))
expected = [(float(v.x), float(v.y)) for v in route.vertices]
def run(mode: str, weights: dict | None = None) -> dict:
return generate_initial_route(
expected,
terrain,
streams,
mode=mode,
weights=weights,
max_grade=GRADE,
min_radius_m=RADIUS,
)
return {**{mode: run(mode) for mode in MODES}, "run": run}
def test_project_follow_is_preprocess_initial(project_runs: dict) -> None:
"""추종 = 전처리 초기값 `planned_route_initial.csv`(1078 m) 그대로."""
from common_util.common_util_route_geometry import (
planned_route_initial_path,
read_planned_route_csv,
)
initial = read_planned_route_csv(planned_route_initial_path(PROJECT))
stored = LineString([(v.x, v.y) for v in initial.vertices])
follow = LineString(project_runs["follow"]["planned"])
assert follow.hausdorff_distance(stored) < 0.01
assert abs(follow.length - stored.length) < 0.01
def test_project_branches_differ_by_goal(project_runs: dict) -> None:
"""각 갈래가 제 목표 비교값에서 가장 좋음 · 토공 최소와 기울기 순한은 선이 갈림(6.1 m)."""
_check_branch_goals({mode: project_runs[mode] for mode in MODES})
assert _gap(project_runs["earthwork"], project_runs["gentle"]) >= 3.0
for mode in MODES:
metrics = project_runs[mode]["metrics"]
assert metrics["over_limit"]["total"] == len(metrics["violations"])
def test_project_similar_and_band(project_runs: dict) -> None:
"""가파른 지형 — 제약 안 길이 한 띠(띠 폭 < 20 m) · 10 m 안인 다른 갈래를 `similar` 로 알림."""
earthwork = project_runs["earthwork"]
assert earthwork["band_width_m"] < 20.0
near = {item["logic"]: item["max_gap_m"] for item in earthwork["similar"]}
assert near["gentle"] == round(_gap(earthwork, project_runs["gentle"]), 1)
assert all(gap < 10.0 for gap in near.values()) and "earthwork" not in near
assert "follow" not in near # 추적은 40 m 넘게 떨어짐
def test_project_expected_axis_moves_little_by_little(project_runs: dict) -> None:
"""예상 100 → 0(토공 : 기울기 1 : 1) — 99 는 추적에서 몇 m 안 · 이웃 단계마다 노선이 움직임 ·
추적에서 점점 멀어짐(이격 늘기만) · 비중 탐색은 넘은 곳을 「비중 계산」 사유로 알림."""
run = project_runs["run"]
near = run("weighted", {"expected": 99, "earthwork": 0.5, "grade": 0.5})
assert 0.0 < _gap(near, project_runs["follow"]) < 10.0
steps = [
run("weighted", {"expected": e, "earthwork": (100 - e) / 2, "grade": (100 - e) / 2})
for e in range(100, -1, -10)
]
assert all(_gap(a, b) > 0.0 for a, b in zip(steps, steps[1:]))
offsets = [step["metrics"]["offset_avg_m"] for step in steps]
assert offsets == sorted(offsets), offsets
reasons = [v["reason"] for v in steps[5]["metrics"]["violations"] if v["kind"] == "grade"]
assert reasons and all(reason.startswith("비중 계산") for reason in reasons)
def test_project_weight_ends_match_branches(project_runs: dict) -> None:
"""랩탑_보조 — 비중 100/0/0 = 추적(1078.01 m) · 0/100/0 = 토공 최소 · 0/0/100 = 기울기 순한."""
_check_weight_ends(project_runs["run"])
assert project_runs["run"]("weighted", ENDS["follow"])["metrics"]["length_m"] == 1078.01
def test_project_weight_moves_monotone(project_runs: dict) -> None:
_check_weight_monotone(project_runs["run"])