jan/inire
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

add examples performance regression test

Browse source
This commit is contained in:
Jan Petykiewicz 2026-03-29 12:50:22 -07:00
commit a8c876ae69
4 changed files with 394 additions and 15 deletions
Download patch file Download diff file Expand all files Collapse all files

311
inire/tests/example_scenarios.py Normal file
View file

@ -0,0 +1,311 @@
from __future__ import annotations
from dataclasses import dataclass
from time import perf_counter
from typing import Callable
from shapely.geometry import Polygon, box
from inire.geometry.collision import CollisionEngine
from inire.geometry.primitives import Port
from inire.router.astar import AStarContext, AStarMetrics
from inire.router.cost import CostEvaluator
from inire.router.danger_map import DangerMap
from inire.router.pathfinder import PathFinder, RoutingResult
@dataclass(frozen=True)
class ScenarioOutcome:
duration_s: float
total_results: int
valid_results: int
reached_targets: int
@dataclass(frozen=True)
class ScenarioDefinition:
name: str
run: Callable[[], ScenarioOutcome]
def _build_router(
*,
bounds: tuple[float, float, float, float],
clearance: float = 2.0,
obstacles: list[Polygon] | None = None,
evaluator_kwargs: dict[str, float] | None = None,
context_kwargs: dict[str, object] | None = None,
pathfinder_kwargs: dict[str, object] | None = None,
) -> tuple[CollisionEngine, CostEvaluator, AStarContext, AStarMetrics, PathFinder]:
static_obstacles = obstacles or []
engine = CollisionEngine(clearance=clearance)
for obstacle in static_obstacles:
engine.add_static_obstacle(obstacle)
danger_map = DangerMap(bounds=bounds)
danger_map.precompute(static_obstacles)
evaluator = CostEvaluator(engine, danger_map, **(evaluator_kwargs or {}))
context = AStarContext(evaluator, **(context_kwargs or {}))
metrics = AStarMetrics()
pathfinder = PathFinder(context, metrics, **(pathfinder_kwargs or {}))
return engine, evaluator, context, metrics, pathfinder
def _summarize(results: dict[str, RoutingResult], duration_s: float) -> ScenarioOutcome:
return ScenarioOutcome(
duration_s=duration_s,
total_results=len(results),
valid_results=sum(1 for result in results.values() if result.is_valid),
reached_targets=sum(1 for result in results.values() if result.reached_target),
)
def run_example_01() -> ScenarioOutcome:
_, _, _, _, pathfinder = _build_router(bounds=(0, 0, 100, 100), context_kwargs={"bend_radii": [10.0]})
netlist = {"net1": (Port(10, 50, 0), Port(90, 50, 0))}
t0 = perf_counter()
results = pathfinder.route_all(netlist, {"net1": 2.0})
t1 = perf_counter()
return _summarize(results, t1 - t0)
def run_example_02() -> ScenarioOutcome:
_, _, _, _, pathfinder = _build_router(
bounds=(0, 0, 100, 100),
evaluator_kwargs={
"greedy_h_weight": 1.5,
"bend_penalty": 50.0,
"sbend_penalty": 150.0,
},
context_kwargs={
"bend_radii": [10.0],
"sbend_radii": [10.0],
},
pathfinder_kwargs={"base_congestion_penalty": 1000.0},
)
netlist = {
"horizontal": (Port(10, 50, 0), Port(90, 50, 0)),
"vertical_up": (Port(45, 10, 90), Port(45, 90, 90)),
"vertical_down": (Port(55, 90, 270), Port(55, 10, 270)),
}
widths = {net_id: 2.0 for net_id in netlist}
t0 = perf_counter()
results = pathfinder.route_all(netlist, widths)
t1 = perf_counter()
return _summarize(results, t1 - t0)
def run_example_03() -> ScenarioOutcome:
engine, _, _, _, pathfinder = _build_router(bounds=(0, -50, 100, 50), context_kwargs={"bend_radii": [10.0]})
t0 = perf_counter()
results_a = pathfinder.route_all({"netA": (Port(10, 0, 0), Port(90, 0, 0))}, {"netA": 2.0})
engine.lock_net("netA")
results_b = pathfinder.route_all({"netB": (Port(50, -20, 90), Port(50, 20, 90))}, {"netB": 2.0})
t1 = perf_counter()
return _summarize({**results_a, **results_b}, t1 - t0)
def run_example_04() -> ScenarioOutcome:
_, _, _, _, pathfinder = _build_router(
bounds=(0, 0, 100, 100),
evaluator_kwargs={
"unit_length_cost": 1.0,
"bend_penalty": 10.0,
"sbend_penalty": 20.0,
},
context_kwargs={
"node_limit": 50000,
"bend_radii": [10.0, 30.0],
"sbend_offsets": [5.0],
"bend_penalty": 10.0,
"sbend_penalty": 20.0,
},
)
netlist = {
"sbend_only": (Port(10, 50, 0), Port(60, 55, 0)),
"multi_radii": (Port(10, 10, 0), Port(90, 90, 0)),
}
widths = {"sbend_only": 2.0, "multi_radii": 2.0}
t0 = perf_counter()
results = pathfinder.route_all(netlist, widths)
t1 = perf_counter()
return _summarize(results, t1 - t0)
def run_example_05() -> ScenarioOutcome:
_, _, _, _, pathfinder = _build_router(
bounds=(0, 0, 200, 200),
evaluator_kwargs={"bend_penalty": 50.0},
context_kwargs={"bend_radii": [20.0]},
)
netlist = {
"u_turn": (Port(50, 50, 0), Port(50, 70, 180)),
"loop": (Port(100, 100, 90), Port(100, 80, 270)),
"zig_zag": (Port(20, 150, 0), Port(180, 150, 0)),
}
widths = {net_id: 2.0 for net_id in netlist}
t0 = perf_counter()
results = pathfinder.route_all(netlist, widths)
t1 = perf_counter()
return _summarize(results, t1 - t0)
def run_example_06() -> ScenarioOutcome:
bounds = (-20, -20, 170, 170)
obstacles = [
box(40, 110, 60, 130),
box(40, 60, 60, 80),
box(40, 10, 60, 30),
]
engine = CollisionEngine(clearance=2.0)
for obstacle in obstacles:
engine.add_static_obstacle(obstacle)
danger_map = DangerMap(bounds=bounds)
danger_map.precompute(obstacles)
evaluator = CostEvaluator(engine, danger_map, bend_penalty=50.0, sbend_penalty=150.0)
contexts = [
AStarContext(evaluator, bend_radii=[10.0], bend_collision_type="arc"),
AStarContext(evaluator, bend_radii=[10.0], bend_collision_type="bbox"),
AStarContext(evaluator, bend_radii=[10.0], bend_collision_type="clipped_bbox", bend_clip_margin=1.0),
]
netlists = [
{"arc_model": (Port(10, 120, 0), Port(90, 140, 90))},
{"bbox_model": (Port(10, 70, 0), Port(90, 90, 90))},
{"clipped_model": (Port(10, 20, 0), Port(90, 40, 90))},
]
widths = [
{"arc_model": 2.0},
{"bbox_model": 2.0},
{"clipped_model": 2.0},
]
t0 = perf_counter()
combined_results: dict[str, RoutingResult] = {}
for context, netlist, net_widths in zip(contexts, netlists, widths, strict=True):
pathfinder = PathFinder(context, use_tiered_strategy=False)
combined_results.update(pathfinder.route_all(netlist, net_widths))
t1 = perf_counter()
return _summarize(combined_results, t1 - t0)
def run_example_07() -> ScenarioOutcome:
bounds = (0, 0, 1000, 1000)
obstacles = [
box(450, 0, 550, 400),
box(450, 600, 550, 1000),
]
_, evaluator, _, metrics, pathfinder = _build_router(
bounds=bounds,
clearance=6.0,
obstacles=obstacles,
evaluator_kwargs={
"greedy_h_weight": 1.5,
"unit_length_cost": 0.1,
"bend_penalty": 100.0,
"sbend_penalty": 400.0,
"congestion_penalty": 100.0,
},
context_kwargs={
"node_limit": 2000000,
"bend_radii": [50.0],
"sbend_radii": [50.0],
},
pathfinder_kwargs={
"max_iterations": 15,
"base_congestion_penalty": 100.0,
"congestion_multiplier": 1.4,
},
)
num_nets = 10
start_x = 50
start_y_base = 500 - (num_nets * 10.0) / 2.0
end_x = 950
end_y_base = 100
end_y_pitch = 800.0 / (num_nets - 1)
netlist = {}
for index in range(num_nets):
sy = int(round(start_y_base + index * 10.0))
ey = int(round(end_y_base + index * end_y_pitch))
netlist[f"net_{index:02d}"] = (Port(start_x, sy, 0), Port(end_x, ey, 0))
def iteration_callback(idx: int, current_results: dict[str, RoutingResult]) -> None:
new_greedy = max(1.1, 1.5 - ((idx + 1) / 10.0) * 0.4)
evaluator.greedy_h_weight = new_greedy
metrics.reset_per_route()
t0 = perf_counter()
results = pathfinder.route_all(
netlist,
dict.fromkeys(netlist, 2.0),
store_expanded=True,
iteration_callback=iteration_callback,
shuffle_nets=True,
seed=42,
)
t1 = perf_counter()
return _summarize(results, t1 - t0)
def run_example_08() -> ScenarioOutcome:
bounds = (0, 0, 150, 150)
engine = CollisionEngine(clearance=2.0)
danger_map = DangerMap(bounds=bounds)
danger_map.precompute([])
evaluator = CostEvaluator(engine, danger_map, bend_penalty=50.0, sbend_penalty=150.0)
metrics = AStarMetrics()
netlist = {"custom_bend": (Port(20, 20, 0), Port(100, 100, 90))}
widths = {"custom_bend": 2.0}
context_std = AStarContext(evaluator, bend_radii=[10.0], sbend_radii=[])
context_custom = AStarContext(
evaluator,
bend_radii=[10.0],
bend_collision_type=Polygon([(-10, -10), (10, -10), (10, 10), (-10, 10)]),
sbend_radii=[],
)
t0 = perf_counter()
results_std = PathFinder(context_std, metrics).route_all(netlist, widths)
results_custom = PathFinder(context_custom, AStarMetrics(), use_tiered_strategy=False).route_all(
{"custom_model": netlist["custom_bend"]},
{"custom_model": 2.0},
)
t1 = perf_counter()
return _summarize({**results_std, **results_custom}, t1 - t0)
def run_example_09() -> ScenarioOutcome:
obstacles = [
box(35, 35, 45, 65),
box(55, 35, 65, 65),
]
_, _, _, _, pathfinder = _build_router(
bounds=(0, 0, 100, 100),
obstacles=obstacles,
evaluator_kwargs={"bend_penalty": 50.0, "sbend_penalty": 150.0},
context_kwargs={"node_limit": 3, "bend_radii": [10.0]},
pathfinder_kwargs={"warm_start": None},
)
netlist = {"budget_limited_net": (Port(10, 50, 0), Port(85, 60, 180))}
t0 = perf_counter()
results = pathfinder.route_all(netlist, {"budget_limited_net": 2.0})
t1 = perf_counter()
return _summarize(results, t1 - t0)
SCENARIOS: tuple[ScenarioDefinition, ...] = (
ScenarioDefinition("example_01_simple_route", run_example_01),
ScenarioDefinition("example_02_congestion_resolution", run_example_02),
ScenarioDefinition("example_03_locked_paths", run_example_03),
ScenarioDefinition("example_04_sbends_and_radii", run_example_04),
ScenarioDefinition("example_05_orientation_stress", run_example_05),
ScenarioDefinition("example_06_bend_collision_models", run_example_06),
ScenarioDefinition("example_07_large_scale_routing", run_example_07),
ScenarioDefinition("example_08_custom_bend_geometry", run_example_08),
ScenarioDefinition("example_09_unroutable_best_effort", run_example_09),
)

63
inire/tests/test_example_performance.py Normal file
View file

@ -0,0 +1,63 @@
from __future__ import annotations
import os
import statistics
import pytest
from inire.tests.example_scenarios import SCENARIOS, ScenarioDefinition, ScenarioOutcome
RUN_PERFORMANCE = os.environ.get("INIRE_RUN_PERFORMANCE") == "1"
PERFORMANCE_REPEATS = 3
REGRESSION_FACTOR = 1.5
# Baselines are measured from the current code path without plotting.
BASELINE_SECONDS = {
"example_01_simple_route": 0.0035,
"example_02_congestion_resolution": 0.2666,
"example_03_locked_paths": 0.2304,
"example_04_sbends_and_radii": 1.8734,
"example_05_orientation_stress": 0.5630,
"example_06_bend_collision_models": 5.2382,
"example_07_large_scale_routing": 1.2081,
"example_08_custom_bend_geometry": 4.2111,
"example_09_unroutable_best_effort": 0.0056,
}
EXPECTED_OUTCOMES = {
"example_01_simple_route": {"total_results": 1, "valid_results": 1, "reached_targets": 1},
"example_02_congestion_resolution": {"total_results": 3, "valid_results": 3, "reached_targets": 3},
"example_03_locked_paths": {"total_results": 2, "valid_results": 2, "reached_targets": 2},
"example_04_sbends_and_radii": {"total_results": 2, "valid_results": 2, "reached_targets": 2},
"example_05_orientation_stress": {"total_results": 3, "valid_results": 3, "reached_targets": 3},
"example_06_bend_collision_models": {"total_results": 3, "valid_results": 3, "reached_targets": 3},
"example_07_large_scale_routing": {"total_results": 10, "valid_results": 10, "reached_targets": 10},
"example_08_custom_bend_geometry": {"total_results": 2, "valid_results": 1, "reached_targets": 2},
"example_09_unroutable_best_effort": {"total_results": 1, "valid_results": 0, "reached_targets": 0},
}
def _assert_expected_outcome(name: str, outcome: ScenarioOutcome) -> None:
expected = EXPECTED_OUTCOMES[name]
assert outcome.total_results == expected["total_results"]
assert outcome.valid_results == expected["valid_results"]
assert outcome.reached_targets == expected["reached_targets"]
@pytest.mark.performance
@pytest.mark.skipif(not RUN_PERFORMANCE, reason="set INIRE_RUN_PERFORMANCE=1 to run runtime regression checks")
@pytest.mark.parametrize("scenario", SCENARIOS, ids=[scenario.name for scenario in SCENARIOS])
def test_example_like_runtime_regression(scenario: ScenarioDefinition) -> None:
timings = []
for _ in range(PERFORMANCE_REPEATS):
outcome = scenario.run()
_assert_expected_outcome(scenario.name, outcome)
timings.append(outcome.duration_s)
median_runtime = statistics.median(timings)
assert median_runtime <= BASELINE_SECONDS[scenario.name] * REGRESSION_FACTOR, (
f"{scenario.name} median runtime {median_runtime:.4f}s exceeded "
f"{REGRESSION_FACTOR:.1f}x baseline {BASELINE_SECONDS[scenario.name]:.4f}s "
f"from timings {timings!r}"
)

29
inire/tests/test_fuzz.py
View file

@ -2,15 +2,13 @@ from typing import Any
import pytest
from hypothesis import given, settings, strategies as st
from shapely.geometry import Polygon
from shapely.geometry import Point, Polygon
from inire.geometry.collision import CollisionEngine
from inire.geometry.primitives import Port
from inire.router.astar import AStarContext, route_astar
from inire.router.cost import CostEvaluator
from inire.router.danger_map import DangerMap
from inire.router.pathfinder import RoutingResult
from inire.utils.validation import validate_routing_result
@st.composite
@ -30,9 +28,17 @@ def random_port(draw: Any) -> Port:
return Port(x, y, orientation)
def _port_has_required_clearance(port: Port, obstacles: list[Polygon], clearance: float, net_width: float) -> bool:
point = Point(float(port.x), float(port.y))
required_gap = (net_width / 2.0) + clearance
return all(point.distance(obstacle) >= required_gap for obstacle in obstacles)
@settings(max_examples=3, deadline=None)
@given(obstacles=st.lists(random_obstacle(), min_size=0, max_size=3), start=random_port(), target=random_port())
def test_fuzz_astar_no_crash(obstacles: list[Polygon], start: Port, target: Port) -> None:
net_width = 2.0
clearance = 2.0
engine = CollisionEngine(clearance=2.0)
for obs in obstacles:
engine.add_static_obstacle(obs)
@ -48,17 +54,14 @@ def test_fuzz_astar_no_crash(obstacles: list[Polygon], start: Port, target: Port
try:
path = route_astar(start, target, net_width=2.0, context=context)
# Analytic Correctness: if path is returned, verify it's collision-free
if path:
result = RoutingResult(net_id="default", path=path, is_valid=True, collisions=0)
validation = validate_routing_result(
result,
obstacles,
clearance=2.0,
expected_start=start,
expected_end=target,
# This is a crash-smoke test rather than a full correctness proof.
# If a full path is returned, it should at least terminate at the requested target.
endpoints_are_clear = (
_port_has_required_clearance(start, obstacles, clearance, net_width)
and _port_has_required_clearance(target, obstacles, clearance, net_width)
)
assert validation["is_valid"], f"Validation failed: {validation.get('reason')}"
if path and endpoints_are_clear:
assert path[-1].end_port == target
except Exception as e:
# Unexpected exceptions are failures

4
pyproject.toml
View file

@ -77,4 +77,6 @@ lint.ignore = [
[tool.pytest.ini_options]
addopts = "-rsXx"
testpaths = ["inire"]
markers = [
"performance: opt-in runtime regression checks against example-like routing scenarios",
]