This commit is contained in:
Jan Petykiewicz 2026-03-09 03:19:01 -07:00
commit 8eb0dbf64a
10 changed files with 910 additions and 391 deletions

View file

@ -1,24 +1,52 @@
from __future__ import annotations
from typing import TYPE_CHECKING, Literal
import rtree
from shapely.geometry import Point, Polygon
from shapely.prepared import prep
if TYPE_CHECKING:
from shapely.geometry import Polygon
from shapely.prepared import PreparedGeometry
from inire.geometry.primitives import Port
class CollisionEngine:
"""Manages spatial queries for collision detection with unified dilation logic."""
"""
Manages spatial queries for collision detection with unified dilation logic.
"""
__slots__ = (
'clearance', 'max_net_width', 'safety_zone_radius',
'static_index', 'static_geometries', 'static_prepared', '_static_id_counter',
'dynamic_index', 'dynamic_geometries', '_dynamic_id_counter'
)
def __init__(self, clearance: float, max_net_width: float = 2.0, safety_zone_radius: float = 0.0021) -> None:
clearance: float
""" Minimum required distance between any two waveguides or obstacles """
max_net_width: float
""" Maximum width of any net in the session (used for pre-dilation) """
safety_zone_radius: float
""" Radius around ports where collisions are ignored """
def __init__(
self,
clearance: float,
max_net_width: float = 2.0,
safety_zone_radius: float = 0.0021,
) -> None:
"""
Initialize the Collision Engine.
Args:
clearance: Minimum required distance (um).
max_net_width: Maximum net width (um).
safety_zone_radius: Safety radius around ports (um).
"""
self.clearance = clearance
self.max_net_width = max_net_width
self.safety_zone_radius = safety_zone_radius
# Static obstacles: store raw geometries to avoid double-dilation
self.static_index = rtree.index.Index()
self.static_geometries: dict[int, Polygon] = {} # ID -> Polygon
@ -32,7 +60,12 @@ class CollisionEngine:
self._dynamic_id_counter = 0
def add_static_obstacle(self, polygon: Polygon) -> None:
"""Add a static obstacle (raw geometry) to the engine."""
"""
Add a static obstacle (raw geometry) to the engine.
Args:
polygon: Raw obstacle geometry.
"""
obj_id = self._static_id_counter
self._static_id_counter += 1
@ -41,7 +74,13 @@ class CollisionEngine:
self.static_index.insert(obj_id, polygon.bounds)
def add_path(self, net_id: str, geometry: list[Polygon]) -> None:
"""Add a net's routed path (raw geometry) to the dynamic index."""
"""
Add a net's routed path (raw geometry) to the dynamic index.
Args:
net_id: Identifier for the net.
geometry: List of raw polygons in the path.
"""
for poly in geometry:
obj_id = self._dynamic_id_counter
self._dynamic_id_counter += 1
@ -49,14 +88,24 @@ class CollisionEngine:
self.dynamic_index.insert(obj_id, poly.bounds)
def remove_path(self, net_id: str) -> None:
"""Remove a net's path from the dynamic index."""
"""
Remove a net's path from the dynamic index.
Args:
net_id: Identifier for the net to remove.
"""
to_remove = [obj_id for obj_id, (nid, _) in self.dynamic_geometries.items() if nid == net_id]
for obj_id in to_remove:
nid, poly = self.dynamic_geometries.pop(obj_id)
self.dynamic_index.delete(obj_id, poly.bounds)
def lock_net(self, net_id: str) -> None:
"""Move a net's dynamic path to static obstacles permanently."""
"""
Move a net's dynamic path to static obstacles permanently.
Args:
net_id: Identifier for the net to lock.
"""
to_move = [obj_id for obj_id, (nid, _) in self.dynamic_geometries.items() if nid == net_id]
for obj_id in to_move:
nid, poly = self.dynamic_geometries.pop(obj_id)
@ -64,59 +113,81 @@ class CollisionEngine:
self.add_static_obstacle(poly)
def is_collision(
self,
geometry: Polygon,
net_width: float = 2.0,
start_port: Port | None = None,
end_port: Port | None = None
) -> bool:
"""Alias for check_collision(buffer_mode='static') for backward compatibility."""
self,
geometry: Polygon,
net_width: float = 2.0,
start_port: Port | None = None,
end_port: Port | None = None,
) -> bool:
"""
Alias for check_collision(buffer_mode='static') for backward compatibility.
Args:
geometry: Move geometry to check.
net_width: Width of the net (unused).
start_port: Starting port for safety check.
end_port: Ending port for safety check.
Returns:
True if collision detected.
"""
_ = net_width
res = self.check_collision(geometry, "default", buffer_mode="static", start_port=start_port, end_port=end_port)
res = self.check_collision(geometry, 'default', buffer_mode='static', start_port=start_port, end_port=end_port)
return bool(res)
def count_congestion(self, geometry: Polygon, net_id: str) -> int:
"""Alias for check_collision(buffer_mode='congestion') for backward compatibility."""
res = self.check_collision(geometry, net_id, buffer_mode="congestion")
"""
Alias for check_collision(buffer_mode='congestion') for backward compatibility.
Args:
geometry: Move geometry to check.
net_id: Identifier for the net.
Returns:
Number of overlapping nets.
"""
res = self.check_collision(geometry, net_id, buffer_mode='congestion')
return int(res)
def check_collision(
self,
geometry: Polygon,
net_id: str,
buffer_mode: Literal["static", "congestion"] = "static",
start_port: Port | None = None,
end_port: Port | None = None
) -> bool | int:
self,
geometry: Polygon,
net_id: str,
buffer_mode: Literal['static', 'congestion'] = 'static',
start_port: Port | None = None,
end_port: Port | None = None,
) -> bool | int:
"""
Check for collisions using unified dilation logic.
If buffer_mode == "static":
Returns True if geometry collides with static obstacles (buffered by full clearance).
If buffer_mode == "congestion":
Returns count of other nets colliding with geometry (both buffered by clearance/2).
Args:
geometry: Raw geometry to check.
net_id: Identifier for the net.
buffer_mode: 'static' (full clearance) or 'congestion' (shared).
start_port: Optional start port for safety zone.
end_port: Optional end port for safety zone.
Returns:
Boolean if static, integer count if congestion.
"""
if buffer_mode == "static":
# Buffered move vs raw static obstacle
# Distance must be >= clearance
if buffer_mode == 'static':
test_poly = geometry.buffer(self.clearance)
candidates = self.static_index.intersection(test_poly.bounds)
for obj_id in candidates:
if self.static_prepared[obj_id].intersects(test_poly):
# Safety zone check (using exact intersection area/bounds)
if start_port or end_port:
intersection = test_poly.intersection(self.static_geometries[obj_id])
if intersection.is_empty:
continue
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) < self.safety_zone_radius and
if p and (abs(ix_minx - p.x) < self.safety_zone_radius and
abs(ix_maxx - p.x) < self.safety_zone_radius and
abs(ix_miny - p.y) < self.safety_zone_radius and
abs(ix_miny - p.y) < self.safety_zone_radius and
abs(ix_maxy - p.y) < self.safety_zone_radius):
is_safe = True
break
@ -125,17 +196,14 @@ class CollisionEngine:
return True
return False
else: # buffer_mode == "congestion"
# Both paths buffered by clearance/2 => Total separation = clearance
dilation = self.clearance / 2.0
test_poly = geometry.buffer(dilation)
candidates = self.dynamic_index.intersection(test_poly.bounds)
count = 0
for obj_id in candidates:
other_net_id, other_poly = self.dynamic_geometries[obj_id]
if other_net_id != net_id:
# Buffer the other path segment too
if test_poly.intersects(other_poly.buffer(dilation)):
count += 1
return count
# buffer_mode == 'congestion'
dilation = self.clearance / 2.0
test_poly = geometry.buffer(dilation)
candidates = self.dynamic_index.intersection(test_poly.bounds)
count = 0
for obj_id in candidates:
other_net_id, other_poly = self.dynamic_geometries[obj_id]
if other_net_id != net_id and test_poly.intersects(other_poly.buffer(dilation)):
count += 1
return count