further performance improvements

This commit is contained in:
Jan Petykiewicz 2026-03-09 22:16:34 -07:00
commit c36bce9978
8 changed files with 168 additions and 58 deletions

Binary file not shown.

Before

Width:  |  Height:  |  Size: 63 KiB

After

Width:  |  Height:  |  Size: 43 KiB

Before After
Before After

View file

@ -18,7 +18,8 @@ def main() -> None:
danger_map.precompute([])
evaluator = CostEvaluator(engine, danger_map, greedy_h_weight=1.1)
router = AStarRouter(evaluator, node_limit=100000)
# router = AStarRouter(evaluator, node_limit=100000)
router = AStarRouter(evaluator, node_limit=100000, bend_collision_type="clipped_bbox", bend_clip_margin=1.0)
pf = PathFinder(router, evaluator)
# 2. Define Netlist with various orientation challenges

Binary file not shown.

Before

Width:  |  Height:  |  Size: 43 KiB

After

Width:  |  Height:  |  Size: 46 KiB

Before After
Before After

View file

@ -17,7 +17,7 @@ class CollisionEngine:
__slots__ = (
'clearance', 'max_net_width', 'safety_zone_radius',
'static_index', 'static_geometries', 'static_dilated', 'static_prepared', '_static_id_counter',
'dynamic_index', 'dynamic_geometries', 'dynamic_dilated', '_dynamic_id_counter'
'dynamic_index', 'dynamic_geometries', 'dynamic_dilated', 'dynamic_prepared', '_dynamic_id_counter'
)
clearance: float
@ -60,6 +60,7 @@ class CollisionEngine:
self.dynamic_geometries: dict[int, tuple[str, Polygon]] = {}
# obj_id -> dilated_geometry (by clearance/2)
self.dynamic_dilated: dict[int, Polygon] = {}
self.dynamic_prepared: dict[int, PreparedGeometry] = {}
self._dynamic_id_counter = 0
def add_static_obstacle(self, polygon: Polygon) -> None:
@ -96,6 +97,7 @@ class CollisionEngine:
self.dynamic_geometries[obj_id] = (net_id, poly)
self.dynamic_dilated[obj_id] = dil
self.dynamic_prepared[obj_id] = prep(dil)
self.dynamic_index.insert(obj_id, dil.bounds)
def remove_path(self, net_id: str) -> None:
@ -109,6 +111,7 @@ class CollisionEngine:
for obj_id in to_remove:
nid, poly = self.dynamic_geometries.pop(obj_id)
dilated = self.dynamic_dilated.pop(obj_id)
self.dynamic_prepared.pop(obj_id)
self.dynamic_index.delete(obj_id, dilated.bounds)
def lock_net(self, net_id: str) -> None:
@ -122,6 +125,7 @@ class CollisionEngine:
for obj_id in to_move:
nid, poly = self.dynamic_geometries.pop(obj_id)
dilated = self.dynamic_dilated.pop(obj_id)
self.dynamic_prepared.pop(obj_id)
self.dynamic_index.delete(obj_id, dilated.bounds)
# Re-buffer for static clearance if necessary.
# Note: dynamic is clearance/2, static is clearance.
@ -178,20 +182,28 @@ class CollisionEngine:
for obj_id in candidates:
if self.static_prepared[obj_id].intersects(geometry):
if start_port or end_port:
# Optimization: Instead of expensive buffer + intersection,
# use distance() and check if it's within clearance only near ports.
raw_obstacle = self.static_geometries[obj_id]
# If the intersection is within clearance, distance will be < clearance.
# We already know it intersects the dilated obstacle, so distance < clearance.
is_safe = False
# Optimization: Skip expensive intersection if neither port is near the obstacle's bounds
# (Plus a small margin for safety zone)
sz = self.safety_zone_radius
is_near_port = False
for p in [start_port, end_port]:
if p:
# Quick bounds check
b = self.static_dilated[obj_id].bounds
if (b[0] - sz <= p.x <= b[2] + sz and
b[1] - sz <= p.y <= b[3] + sz):
is_near_port = True
break
# Use intersection bounds to check proximity to ports
# We need the intersection of the geometry and the RAW obstacle
if not is_near_port:
return True # Collision, and not near any port safety zone
# Only if near port, do the expensive check
raw_obstacle = self.static_geometries[obj_id]
intersection = geometry.intersection(raw_obstacle)
if not intersection.is_empty:
ix_minx, ix_miny, ix_maxx, ix_maxy = intersection.bounds
is_safe = False
for p in [start_port, end_port]:
if p and (abs(ix_minx - p.x) < sz and
abs(ix_maxx - p.x) < sz and
@ -200,8 +212,8 @@ class CollisionEngine:
is_safe = True
break
if is_safe:
continue
if is_safe:
continue
return True
return False
@ -213,6 +225,6 @@ class CollisionEngine:
count = 0
for obj_id in candidates:
other_net_id, _ = self.dynamic_geometries[obj_id]
if other_net_id != net_id and test_poly.intersects(self.dynamic_dilated[obj_id]):
if other_net_id != net_id and self.dynamic_prepared[obj_id].intersects(test_poly):
count += 1
return count

View file

@ -2,6 +2,7 @@ from __future__ import annotations
from typing import Literal, cast
import numpy
import shapely
from shapely.geometry import Polygon, box
from shapely.ops import unary_union
@ -44,10 +45,10 @@ class ComponentResult:
length: float
""" Physical length of the component path """
bounds: list[tuple[float, float, float, float]]
bounds: numpy.ndarray
""" Pre-calculated bounds for each polygon in geometry """
dilated_bounds: list[tuple[float, float, float, float]] | None
dilated_bounds: numpy.ndarray | None
""" Pre-calculated bounds for each polygon in dilated_geometry """
def __init__(
@ -61,16 +62,26 @@ class ComponentResult:
self.dilated_geometry = dilated_geometry
self.end_port = end_port
self.length = length
self.bounds = [p.bounds for p in geometry]
self.dilated_bounds = [p.bounds for p in dilated_geometry] if dilated_geometry else None
# Vectorized bounds calculation
self.bounds = shapely.bounds(geometry)
self.dilated_bounds = shapely.bounds(dilated_geometry) if dilated_geometry else None
def translate(self, dx: float, dy: float) -> ComponentResult:
"""
Create a new ComponentResult translated by (dx, dy).
"""
# Vectorized translation if possible, else list comp
# Shapely 2.x affinity functions still work on single geometries efficiently
geoms = self.geometry
if self.dilated_geometry:
geoms = geoms + self.dilated_geometry
from shapely.affinity import translate
new_geom = [translate(p, dx, dy) for p in self.geometry]
new_dil = [translate(p, dx, dy) for p in self.dilated_geometry] if self.dilated_geometry else None
translated = [translate(p, dx, dy) for p in geoms]
new_geom = translated[:len(self.geometry)]
new_dil = translated[len(self.geometry):] if self.dilated_geometry else None
new_port = Port(self.end_port.x + dx, self.end_port.y + dy, self.end_port.orientation)
return ComponentResult(new_geom, new_port, self.length, new_dil)

View file

@ -24,7 +24,7 @@ class AStarNode:
"""
A node in the A* search graph.
"""
__slots__ = ('port', 'g_cost', 'h_cost', 'f_cost', 'parent', 'component_result', 'count')
__slots__ = ('port', 'g_cost', 'h_cost', 'f_cost', 'parent', 'component_result', 'count', 'path_bbox')
port: Port
""" Port representing the state at this node """
@ -47,6 +47,9 @@ class AStarNode:
count: int
""" Unique insertion order for tie-breaking """
path_bbox: tuple[float, float, float, float] | None
""" Bounding box of the entire path up to this node """
_count = 0
def __init__(
@ -66,6 +69,33 @@ class AStarNode:
self.count = AStarNode._count
AStarNode._count += 1
# Calculate path_bbox
if parent is None:
self.path_bbox = None
else:
# Union of parent's bbox and current move's bbox
if component_result:
# Merge all polygon bounds in the result
minx, miny, maxx, maxy = 1e15, 1e15, -1e15, -1e15
for b in component_result.dilated_bounds if component_result.dilated_bounds is not None else component_result.bounds:
minx = min(minx, b[0])
miny = min(miny, b[1])
maxx = max(maxx, b[2])
maxy = max(maxy, b[3])
if parent.path_bbox:
self.path_bbox = (
min(minx, parent.path_bbox[0]),
min(miny, parent.path_bbox[1]),
max(maxx, parent.path_bbox[2]),
max(maxy, parent.path_bbox[3])
)
else:
self.path_bbox = (minx, miny, maxx, maxy)
else:
self.path_bbox = parent.path_bbox
def __lt__(self, other: AStarNode) -> bool:
@ -142,7 +172,7 @@ class AStarRouter:
self.cost_evaluator = cost_evaluator
self.config = RouterConfig(
node_limit=node_limit,
straight_lengths=straight_lengths if straight_lengths is not None else [1.0, 5.0, 25.0],
straight_lengths=straight_lengths if straight_lengths is not None else [1.0, 5.0, 25.0, 100.0],
bend_radii=bend_radii if bend_radii is not None else [10.0],
sbend_offsets=sbend_offsets if sbend_offsets is not None else [-5.0, -2.0, 2.0, 5.0],
sbend_radii=sbend_radii if sbend_radii is not None else [10.0],
@ -283,7 +313,14 @@ class AStarRouter:
# Level 2: Relative cache (orientation only)
rel_key = (base_ori, 'S', length, net_width, self._self_dilation)
if rel_key in self._move_cache:
res = self._move_cache[rel_key].translate(cp.x, cp.y)
res_rel = self._move_cache[rel_key]
# Check closed set before translating
ex = res_rel.end_port.x + cp.x
ey = res_rel.end_port.y + cp.y
end_state = (round(ex, 3), round(ey, 3), round(res_rel.end_port.orientation, 2))
if end_state in closed_set:
continue
res = res_rel.translate(cp.x, cp.y)
else:
res_rel = Straight.generate(Port(0, 0, base_ori), length, net_width, dilation=self._self_dilation)
self._move_cache[rel_key] = res_rel
@ -300,7 +337,14 @@ class AStarRouter:
else:
rel_key = (base_ori, 'B', radius, direction, net_width, self.config.bend_collision_type, self._self_dilation)
if rel_key in self._move_cache:
res = self._move_cache[rel_key].translate(cp.x, cp.y)
res_rel = self._move_cache[rel_key]
# Check closed set before translating
ex = res_rel.end_port.x + cp.x
ey = res_rel.end_port.y + cp.y
end_state = (round(ex, 3), round(ey, 3), round(res_rel.end_port.orientation, 2))
if end_state in closed_set:
continue
res = res_rel.translate(cp.x, cp.y)
else:
res_rel = Bend90.generate(
Port(0, 0, base_ori),
@ -325,14 +369,21 @@ class AStarRouter:
else:
rel_key = (base_ori, 'SB', offset, radius, net_width, self.config.bend_collision_type, self._self_dilation)
if rel_key in self._move_cache:
res = self._move_cache[rel_key].translate(cp.x, cp.y)
res_rel = self._move_cache[rel_key]
# Check closed set before translating
ex = res_rel.end_port.x + cp.x
ey = res_rel.end_port.y + cp.y
end_state = (round(ex, 3), round(ey, 3), round(res_rel.end_port.orientation, 2))
if end_state in closed_set:
continue
res = res_rel.translate(cp.x, cp.y)
else:
try:
res_rel = SBend.generate(
Port(0, 0, base_ori),
offset,
radius,
net_width,
width=net_width,
collision_type=self.config.bend_collision_type,
clip_margin=self.config.bend_clip_margin,
dilation=self._self_dilation
@ -387,29 +438,48 @@ class AStarRouter:
return
# 3. Check for Self-Intersection (Limited to last 100 segments for performance)
# Optimization: use pre-dilated geometries and pre-calculated bounds
if result.dilated_geometry:
for dm_idx, dilated_move in enumerate(result.dilated_geometry):
dm_bounds = result.dilated_bounds[dm_idx]
curr_p: AStarNode | None = parent
seg_idx = 0
while curr_p and curr_p.component_result and seg_idx < 100:
if seg_idx > 0:
res_p = curr_p.component_result
if res_p.dilated_geometry:
for dp_idx, dilated_prev in enumerate(res_p.dilated_geometry):
dp_bounds = res_p.dilated_bounds[dp_idx]
# Quick bounds overlap check
if not (dm_bounds[0] > dp_bounds[2] or
dm_bounds[2] < dp_bounds[0] or
dm_bounds[1] > dp_bounds[3] or
dm_bounds[3] < dp_bounds[1]):
if dilated_move.intersects(dilated_prev):
overlap = dilated_move.intersection(dilated_prev)
if not overlap.is_empty and overlap.area > 1e-6:
return
curr_p = curr_p.parent
seg_idx += 1
# Union of current move's bounds for fast path-wide pruning
m_minx, m_miny, m_maxx, m_maxy = 1e15, 1e15, -1e15, -1e15
for b in result.dilated_bounds if result.dilated_bounds is not None else result.bounds:
m_minx = min(m_minx, b[0])
m_miny = min(m_miny, b[1])
m_maxx = max(m_maxx, b[2])
m_maxy = max(m_maxy, b[3])
# If current move doesn't overlap the entire parent path bbox, we can skip individual checks
# (Except the immediate parent which we usually skip anyway)
if parent.path_bbox and not (m_minx > parent.path_bbox[2] or
m_maxx < parent.path_bbox[0] or
m_miny > parent.path_bbox[3] or
m_maxy < parent.path_bbox[1]):
for dm_idx, dilated_move in enumerate(result.dilated_geometry):
dm_bounds = result.dilated_bounds[dm_idx]
curr_p: AStarNode | None = parent
seg_idx = 0
while curr_p and curr_p.component_result and seg_idx < 100:
# Skip immediate parent to avoid tangent/port-safety issues
if seg_idx > 0:
res_p = curr_p.component_result
if res_p.dilated_geometry:
for dp_idx, dilated_prev in enumerate(res_p.dilated_geometry):
dp_bounds = res_p.dilated_bounds[dp_idx]
# Quick bounds overlap check
if not (dm_bounds[0] > dp_bounds[2] or
dm_bounds[2] < dp_bounds[0] or
dm_bounds[1] > dp_bounds[3] or
dm_bounds[3] < dp_bounds[1]):
# Use intersects() which is much faster than intersection()
if dilated_move.intersects(dilated_prev):
# Only do expensive area check if absolutely necessary
overlap = dilated_move.intersection(dilated_prev)
if not overlap.is_empty and overlap.area > 1e-6:
return
curr_p = curr_p.parent
seg_idx += 1
move_cost = self.cost_evaluator.evaluate_move(
result.geometry,
@ -418,7 +488,8 @@ class AStarRouter:
net_id,
start_port=parent.port,
length=result.length,
dilated_geometry=result.dilated_geometry
dilated_geometry=result.dilated_geometry,
skip_static=True
)
if move_cost > 1e12:

View file

@ -28,3 +28,4 @@ class CostConfig:
unit_length_cost: float = 1.0
greedy_h_weight: float = 1.1
congestion_penalty: float = 10000.0
bend_penalty: float = 50.0

View file

@ -39,6 +39,7 @@ class CostEvaluator:
unit_length_cost: float = 1.0,
greedy_h_weight: float = 1.1,
congestion_penalty: float = 10000.0,
bend_penalty: float = 50.0,
) -> None:
"""
Initialize the Cost Evaluator.
@ -49,6 +50,7 @@ class CostEvaluator:
unit_length_cost: Cost multiplier per micrometer of path length.
greedy_h_weight: Heuristic weighting (A* greedy factor).
congestion_penalty: Multiplier for path overlaps in negotiated congestion.
bend_penalty: Base cost for 90-degree bends.
"""
self.collision_engine = collision_engine
self.danger_map = danger_map
@ -56,6 +58,7 @@ class CostEvaluator:
unit_length_cost=unit_length_cost,
greedy_h_weight=greedy_h_weight,
congestion_penalty=congestion_penalty,
bend_penalty=bend_penalty,
)
# Use config values
@ -63,6 +66,7 @@ class CostEvaluator:
self.greedy_h_weight = self.config.greedy_h_weight
self.congestion_penalty = self.config.congestion_penalty
def g_proximity(self, x: float, y: float) -> float:
"""
Get proximity cost from the Danger Map.
@ -86,14 +90,21 @@ class CostEvaluator:
Returns:
Heuristic cost estimate.
"""
dist = abs(current.x - target.x) + abs(current.y - target.y)
dx = abs(current.x - target.x)
dy = abs(current.y - target.y)
dist = dx + dy
# Orientation penalty if not aligned with target entry
# If we need to turn, the cost is at least min_bend_radius * pi/2
# But we also need to account for the physical distance required for the turn.
penalty = 0.0
if current.orientation != target.orientation:
penalty += 50.0 # Arbitrary high cost for mismatch
# 90-degree turn cost: radius 10 -> ~15.7 um + penalty
penalty += 15.7 + self.config.bend_penalty
# Add 1.5 multiplier for greediness (faster search)
return 1.5 * (dist + penalty)
return self.greedy_h_weight * (dist + penalty)
def evaluate_move(
self,
@ -104,6 +115,7 @@ class CostEvaluator:
start_port: Port | None = None,
length: float = 0.0,
dilated_geometry: list[Polygon] | None = None,
skip_static: bool = False,
) -> float:
"""
Calculate the cost of a single move (Straight, Bend, SBend).
@ -116,6 +128,7 @@ class CostEvaluator:
start_port: Port at the start of the move.
length: Physical path length of the move.
dilated_geometry: Pre-calculated dilated polygons.
skip_static: If True, bypass static collision checks (e.g. if already done).
Returns:
Total cost of the move, or 1e15 if invalid.
@ -131,11 +144,12 @@ class CostEvaluator:
for i, poly in enumerate(geometry):
dil_poly = dilated_geometry[i] if dilated_geometry else None
# Hard Collision (Static obstacles)
if self.collision_engine.check_collision(
poly, net_id, buffer_mode='static', start_port=start_port, end_port=end_port,
dilated_geometry=dil_poly
):
return 1e15
if not skip_static:
if self.collision_engine.check_collision(
poly, net_id, buffer_mode='static', start_port=start_port, end_port=end_port,
dilated_geometry=dil_poly
):
return 1e15
# Soft Collision (Negotiated Congestion)
overlaps = self.collision_engine.check_collision(