[Pather] add strategy arg

This commit is contained in:
Jan Petykiewicz 2026-07-10 14:20:04 -07:00
commit cf3f72b828
5 changed files with 188 additions and 29 deletions

View file

@ -58,6 +58,7 @@ from .planner.interface import (
route_error_is_fatal,
)
from .planner import (
RouteTieBreakStrategy,
RoutingPlanner,
)
from .planner.bounds import resolved_position_bound
@ -517,6 +518,7 @@ class Pather(PortList):
length: float | None = None,
*,
spacing: float | ArrayLike | None = None,
strategy: RouteTieBreakStrategy | str | None = None,
**bounds: Any,
) -> Self:
"""
@ -530,16 +532,18 @@ class Pather(PortList):
For a single port with no length or bound, legal primitive-offer
candidates are evaluated at their minimum legal length-like parameters,
then cost selects among those minimum-length candidates. `out_ptype`,
when provided, constrains only the final route endpoint.
when provided, constrains only the final route endpoint. `strategy`
controls straight-first vs turn-first ordering only after cost and
structural tie-breakers.
`spacing` and `set_rotation` are only valid when using a bundle bound.
"""
with self._logger.log_operation(self, 'trace', portspec, ccw=ccw, length=length, spacing=spacing, **bounds):
with self._logger.log_operation(self, 'trace', portspec, ccw=ccw, length=length, spacing=spacing, strategy=strategy, **bounds):
if isinstance(portspec, str):
portspec = [portspec]
contexts = self._route_contexts(portspec)
try:
result = self.planner.plan_trace_route(contexts, ccw, length, spacing=spacing, **bounds)
result = self.planner.plan_trace_route(contexts, ccw, length, spacing=spacing, strategy=strategy, **bounds)
except (BuildError, NotImplementedError) as err:
if not self._dead or route_error_is_fatal(err):
raise
@ -576,6 +580,7 @@ class Pather(PortList):
ccw: SupportsBool | None,
*,
spacing: float | ArrayLike | None = None,
strategy: RouteTieBreakStrategy | str | None = None,
**bounds: Any,
) -> Self:
"""
@ -587,13 +592,15 @@ class Pather(PortList):
With no positional or bundle bound, single-port `trace_to()` uses the
same omitted minimum-length primitive-offer behavior as `trace()`.
`strategy` controls straight-first vs turn-first ordering only after
cost and structural tie-breakers.
"""
with self._logger.log_operation(self, 'trace_to', portspec, ccw=ccw, spacing=spacing, **bounds):
with self._logger.log_operation(self, 'trace_to', portspec, ccw=ccw, spacing=spacing, strategy=strategy, **bounds):
if isinstance(portspec, str):
portspec = [portspec]
contexts = self._route_contexts(portspec)
try:
result = self.planner.plan_trace_to_route(contexts, ccw, spacing=spacing, **bounds)
result = self.planner.plan_trace_to_route(contexts, ccw, spacing=spacing, strategy=strategy, **bounds)
except (BuildError, NotImplementedError) as err:
if not self._dead or len(contexts) != 1 or route_error_is_fatal(err):
raise
@ -640,6 +647,7 @@ class Pather(PortList):
length: float | None = None,
*,
spacing: float | ArrayLike | None = None,
strategy: RouteTieBreakStrategy | str | None = None,
**bounds: Any,
) -> Self:
"""
@ -655,14 +663,16 @@ class Pather(PortList):
Multi-port jogs require `spacing`; the innermost first-bend port uses
the base `length` or omitted-length solve, and other ports derive exact
route lengths and offsets from that base route. `out_ptype`, when
provided, constrains only each final route endpoint.
provided, constrains only each final route endpoint. `strategy`
controls straight-first vs S-first ordering only after cost and
structural tie-breakers.
"""
with self._logger.log_operation(self, 'jog', portspec, offset=offset, length=length, spacing=spacing, **bounds):
with self._logger.log_operation(self, 'jog', portspec, offset=offset, length=length, spacing=spacing, strategy=strategy, **bounds):
if isinstance(portspec, str):
portspec = [portspec]
contexts = self._route_contexts(portspec)
try:
result = self.planner.plan_jog_route(contexts, offset, length, spacing=spacing, **bounds)
result = self.planner.plan_jog_route(contexts, offset, length, spacing=spacing, strategy=strategy, **bounds)
except (BuildError, NotImplementedError) as err:
if not self._dead or len(contexts) != 1 or route_error_is_fatal(err):
raise
@ -701,6 +711,7 @@ class Pather(PortList):
length: float | None = None,
*,
spacing: float | ArrayLike | None = None,
strategy: RouteTieBreakStrategy | str | None = None,
**bounds: Any,
) -> Self:
"""
@ -714,14 +725,16 @@ class Pather(PortList):
other ports derive exact lengths and offsets from it. Use `length=0` to
request the old zero-public-length U-turn shape. Positional and
bundle-bound keywords are not supported for this operation. `out_ptype`,
when provided, constrains only each final route endpoint.
when provided, constrains only each final route endpoint. `strategy`
controls straight-first vs U-first ordering only after cost and
structural tie-breakers.
"""
with self._logger.log_operation(self, 'uturn', portspec, offset=offset, length=length, spacing=spacing, **bounds):
with self._logger.log_operation(self, 'uturn', portspec, offset=offset, length=length, spacing=spacing, strategy=strategy, **bounds):
if isinstance(portspec, str):
portspec = [portspec]
contexts = self._route_contexts(portspec)
try:
result = self.planner.plan_uturn_route(contexts, offset, length, spacing=spacing, **bounds)
result = self.planner.plan_uturn_route(contexts, offset, length, spacing=spacing, strategy=strategy, **bounds)
except (BuildError, NotImplementedError) as err:
if (
not self._dead
@ -752,6 +765,7 @@ class Pather(PortList):
out_ptype: str | None = None,
plug_destination: bool = True,
thru: str | None = None,
strategy: RouteTieBreakStrategy | str | None = None,
**kwargs: Any,
) -> Self:
"""
@ -760,9 +774,10 @@ class Pather(PortList):
The current baseline searches bounded primitive-offer routes with up to
four bend roles, including straight, single-bend, S-like, U-like, and
dogleg topologies. The lowest-cost legal bounded candidate is selected;
bend count, step count, and search order are tie-breakers only.
bend count, step count, and search order are tie-breakers only. `strategy`
controls straight-first vs turn-first search order within those ties.
Route-shape kwargs such as `length`, `offset`, `ccw`, positional bounds,
and bundle bounds are reserved for this internal solve; other tool
bundle bounds, and `strategy` are reserved for this internal solve; other tool
kwargs are forwarded to primitive offer generation.
If `plug_destination` is `True`, the destination port is consumed by the final step.
@ -779,6 +794,7 @@ class Pather(PortList):
out_ptype=out_ptype,
plug_destination=plug_destination,
thru=thru,
strategy=strategy,
**kwargs,
):
result = self.planner.plan_trace_into(
@ -788,6 +804,7 @@ class Pather(PortList):
out_ptype = out_ptype,
plug_destination = plug_destination,
thru = thru,
strategy = strategy,
**kwargs,
)
self._apply_route_result(result)

View file

@ -12,4 +12,5 @@ from .interface import (
RoutePortContext as RoutePortContext,
route_error_is_fatal as route_error_is_fatal,
)
from .planner import RouteTieBreakStrategy as RouteTieBreakStrategy
from .planner import RoutingPlanner as RoutingPlanner

View file

@ -53,6 +53,15 @@ from .interface import (
route_error_is_fatal,
)
RouteTieBreakStrategy = Literal['straight_first', 'turn_first']
def validate_strategy(strategy: RouteTieBreakStrategy | str) -> RouteTieBreakStrategy:
"""Return a supported route tie-break strategy or raise a routing error."""
if strategy in ('straight_first', 'turn_first'):
return strategy
raise BuildError(f'Invalid route strategy {strategy!r}; expected straight_first or turn_first')
def is_close(a: float, b: float) -> bool:
"""Compare route-solver scalars with the planner tolerance."""
@ -245,6 +254,8 @@ class RouteRequest:
"""Whether bend-family trace_into routes must also match `jog`."""
max_bends: int | None = None
"""Optional override for grammar bend budget."""
strategy: RouteTieBreakStrategy = 'straight_first'
"""Discovery-order tie-break strategy for straight-vs-turn placement."""
@property
def route_name(self) -> str:
@ -715,7 +726,10 @@ class Solver:
residual_jog: float,
) -> Iterable[tuple[SelectedPrimitive, ...]]:
"""Recursively enumerate normal/adapter/turn blocks within the bend budget."""
for normal in self.straight_options(steps):
straight_options = self.straight_options(steps)
if self.request.strategy == 'straight_first':
straight_options = (*straight_options[1:], straight_options[0])
for normal in straight_options:
after_normal = (*steps, *normal)
suffix_options = (
self.adapter_options(after_normal, residual_jog=0)
@ -949,6 +963,16 @@ class RoutingPlanner:
"""
TRACE_INTO_MAX_BENDS: int = 4
DEFAULT_STRATEGY: RouteTieBreakStrategy = 'straight_first'
def __init__(self, strategy: RouteTieBreakStrategy = DEFAULT_STRATEGY) -> None:
self.strategy = validate_strategy(strategy)
def resolve_strategy(self, strategy: RouteTieBreakStrategy | str | None) -> RouteTieBreakStrategy:
"""Return the per-route strategy or the planner default."""
if strategy is None:
return getattr(self, 'strategy', self.DEFAULT_STRATEGY)
return validate_strategy(strategy)
def trace_into_bend_bands(self, family: PrimitiveKind) -> tuple[tuple[int, int], ...]:
"""Return non-overlapping trace_into bend-budget bands for staged solving."""
@ -972,6 +996,7 @@ class RoutingPlanner:
ccw: SupportsBool | None = None,
constrain_jog: bool = False,
max_bends: int | None = None,
strategy: RouteTieBreakStrategy | str | None = None,
**kwargs: Any,
) -> RouteRequest:
"""Build normalized solver input for one route leg."""
@ -986,6 +1011,7 @@ class RoutingPlanner:
out_ptype=kwargs.get('out_ptype'),
constrain_jog=constrain_jog,
max_bends=max_bends,
strategy=self.resolve_strategy(strategy),
)
def solver_for_request(self, request: RouteRequest) -> Solver:
@ -1019,6 +1045,7 @@ class RoutingPlanner:
plug_into: str | None = None,
constrain_jog: bool = False,
max_bends: int | None = None,
strategy: RouteTieBreakStrategy | str | None = None,
**kwargs: Any,
) -> RouteLeg:
"""Solve one route leg and attach it to its source Pather context."""
@ -1030,6 +1057,7 @@ class RoutingPlanner:
ccw=ccw,
constrain_jog=constrain_jog,
max_bends=max_bends,
strategy=strategy,
**kwargs,
)
try:
@ -1096,6 +1124,7 @@ class RoutingPlanner:
length: float | None = None,
*,
spacing: float | ArrayLike | None = None,
strategy: RouteTieBreakStrategy | str | None = None,
**bounds: Any,
) -> PreparedRouteResult:
"""Plan straight or single-bend traces, including `each` and bundle-bound modes."""
@ -1104,21 +1133,21 @@ class RoutingPlanner:
planner_bounds.validate_trace_args(portspec, length=length, spacing=spacing, bounds=route_bounds)
family: Literal['straight', 'bend'] = 'straight' if ccw is None else 'bend'
if length is not None:
leg = self.plan_leg(family, contexts[0], length=length, ccw=ccw, **route_bounds)
leg = self.plan_leg(family, contexts[0], length=length, ccw=ccw, strategy=strategy, **route_bounds)
return self.prepared_result_from_legs((leg,))
if route_bounds.get('each') is not None:
each = route_bounds.pop('each')
return PreparedRouteResult(tuple(
self.prepared_route_action_from_leg(
self.plan_leg(family, context, length=each, ccw=ccw, **route_bounds),
self.plan_leg(family, context, length=each, ccw=ccw, strategy=strategy, **route_bounds),
)
for context in contexts
))
bundle_bounds = planner_bounds.present_bundle_bounds(route_bounds)
if not bundle_bounds:
leg = self.plan_leg(family, contexts[0], length=None, ccw=ccw, **route_bounds)
leg = self.plan_leg(family, contexts[0], length=None, ccw=ccw, strategy=strategy, **route_bounds)
return self.prepared_result_from_legs((leg,))
bound_type = bundle_bounds[0]
@ -1135,7 +1164,7 @@ class RoutingPlanner:
actions = []
for port_name, route_length in extensions.items():
context = next(context for context in contexts if context.portspec == port_name)
leg = self.plan_leg(family, context, length=route_length, ccw=ccw, **route_bounds)
leg = self.plan_leg(family, context, length=route_length, ccw=ccw, strategy=strategy, **route_bounds)
actions.append(self.prepared_route_action_from_leg(leg))
return PreparedRouteResult(tuple(actions))
@ -1145,6 +1174,7 @@ class RoutingPlanner:
ccw: SupportsBool | None,
*,
spacing: float | ArrayLike | None = None,
strategy: RouteTieBreakStrategy | str | None = None,
**bounds: Any,
) -> PreparedRouteResult:
"""Plan `trace_to()` by resolving positional targets or delegating to `trace()` modes."""
@ -1156,7 +1186,7 @@ class RoutingPlanner:
if any(route_bounds.get(key) is not None for key in planner_bounds.POSITION_KEYS):
raise BuildError('Position bounds only allowed with a single port')
if resolved is None:
return self.plan_trace_route(contexts, ccw, spacing=spacing, **route_bounds)
return self.plan_trace_route(contexts, ccw, spacing=spacing, strategy=strategy, **route_bounds)
planner_bounds.validate_trace_to_positional_args(spacing=spacing, bounds=route_bounds)
_key, _value, length = resolved
@ -1166,7 +1196,7 @@ class RoutingPlanner:
if key not in planner_bounds.POSITION_KEYS and key != 'length'
}
family: Literal['straight', 'bend'] = 'straight' if ccw is None else 'bend'
leg = self.plan_leg(family, contexts[0], length=length, ccw=ccw, **other_bounds)
leg = self.plan_leg(family, contexts[0], length=length, ccw=ccw, strategy=strategy, **other_bounds)
return self.prepared_result_from_legs((leg,))
def plan_jog_route(
@ -1176,11 +1206,12 @@ class RoutingPlanner:
length: float | None = None,
*,
spacing: float | ArrayLike | None = None,
strategy: RouteTieBreakStrategy | str | None = None,
**bounds: Any,
) -> PreparedRouteResult:
"""Plan S-bend routes for single ports or spaced bundles."""
if numpy.isclose(offset, 0):
return self.plan_trace_to_route(contexts, None, length=length, spacing=spacing, **bounds)
return self.plan_trace_to_route(contexts, None, length=length, spacing=spacing, strategy=strategy, **bounds)
route_bounds = dict(bounds)
portspec = tuple(context.portspec for context in contexts)
planner_bounds.validate_jog_args(portspec, length=length, spacing=spacing, bounds=route_bounds)
@ -1193,9 +1224,9 @@ class RoutingPlanner:
if len(contexts) > 1:
return PreparedRouteResult(tuple(
self.prepared_route_action_from_leg(leg)
for leg in self.plan_su_bundle_routes('s', contexts, offset, length, spacing, **other_bounds)
for leg in self.plan_su_bundle_routes('s', contexts, offset, length, spacing, strategy=strategy, **other_bounds)
))
leg = self.plan_leg('s', contexts[0], length=length, jog=offset, **other_bounds)
leg = self.plan_leg('s', contexts[0], length=length, jog=offset, strategy=strategy, **other_bounds)
return self.prepared_result_from_legs((leg,))
def plan_uturn_route(
@ -1205,6 +1236,7 @@ class RoutingPlanner:
length: float | None = None,
*,
spacing: float | ArrayLike | None = None,
strategy: RouteTieBreakStrategy | str | None = None,
**bounds: Any,
) -> PreparedRouteResult:
"""Plan U-turn routes for single ports or spaced bundles."""
@ -1214,9 +1246,9 @@ class RoutingPlanner:
if len(contexts) > 1:
return PreparedRouteResult(tuple(
self.prepared_route_action_from_leg(leg)
for leg in self.plan_su_bundle_routes('u', contexts, offset, length, spacing, **route_bounds)
for leg in self.plan_su_bundle_routes('u', contexts, offset, length, spacing, strategy=strategy, **route_bounds)
))
leg = self.plan_leg('u', contexts[0], length=length, jog=offset, **route_bounds)
leg = self.plan_leg('u', contexts[0], length=length, jog=offset, strategy=strategy, **route_bounds)
return self.prepared_result_from_legs((leg,))
def plan_su_bundle_routes(
@ -1226,6 +1258,8 @@ class RoutingPlanner:
offset: float,
length: float | None,
spacing: float | ArrayLike | None,
*,
strategy: RouteTieBreakStrategy | str | None = None,
**kwargs: Any,
) -> tuple[RouteLeg, ...]:
"""
@ -1236,27 +1270,34 @@ class RoutingPlanner:
and offset so all legs can be planned independently.
"""
if len(contexts) == 1:
return (self.plan_leg(kind, contexts[0], length=length, jog=offset, **kwargs),)
return (self.plan_leg(kind, contexts[0], length=length, jog=offset, strategy=strategy, **kwargs),)
route_name = 'jog' if kind == 's' else 'uturn'
if kind == 'u' and is_close(offset, 0):
raise BuildError('multi-port uturn() requires nonzero offset to determine bundle ordering')
contexts_by_name = {context.portspec: context for context in contexts}
initial_specs = planner_bounds.su_bundle_specs(contexts, offset, 0, spacing, route_name=route_name)
anchor_portspec, _anchor_length, _anchor_offset = initial_specs[0]
anchor = self.plan_leg(kind, contexts_by_name[anchor_portspec], length=length, jog=offset, **kwargs)
anchor = self.plan_leg(kind, contexts_by_name[anchor_portspec], length=length, jog=offset, strategy=strategy, **kwargs)
base_length = anchor.candidate.public_length
specs = planner_bounds.su_bundle_specs(contexts, offset, base_length, spacing, route_name=route_name)
first_portspec, _first_length, _first_offset = specs[0]
routes_by_name = {first_portspec: anchor}
for spec_portspec, spec_length, spec_offset in specs[1:]:
if kind == 's' and is_close(spec_offset, 0):
routes_by_name[spec_portspec] = self.plan_leg('straight', contexts_by_name[spec_portspec], length=spec_length, **kwargs)
routes_by_name[spec_portspec] = self.plan_leg(
'straight',
contexts_by_name[spec_portspec],
length=spec_length,
strategy=strategy,
**kwargs,
)
else:
routes_by_name[spec_portspec] = self.plan_leg(
kind,
contexts_by_name[spec_portspec],
length=spec_length,
jog=spec_offset,
strategy=strategy,
**kwargs,
)
return tuple(routes_by_name[spec_portspec] for spec_portspec, _length, _offset in specs)
@ -1270,6 +1311,7 @@ class RoutingPlanner:
out_ptype: str | None,
plug_destination: bool,
thru: str | None,
strategy: RouteTieBreakStrategy | str | None = None,
**kwargs: Any,
) -> PreparedRouteResult:
"""Plan a bounded route from one source port into a destination port."""
@ -1297,6 +1339,7 @@ class RoutingPlanner:
ccw=ccw,
constrain_jog=family == 'bend',
max_bends=self.TRACE_INTO_MAX_BENDS,
strategy=strategy,
**(dict(kwargs) | {'out_ptype': out_ptype}),
)
solver = self.solver_for_request(request)

View file

@ -6,6 +6,7 @@ import pytest
from numpy import pi
from masque import Library, Path, Port, Pather
from masque.builder.planner import RoutingPlanner
from masque.builder.tools import (
BendOffer,
PathTool,
@ -422,6 +423,103 @@ def test_pather_selects_lowest_cost_offer() -> None:
assert numpy.allclose(p.ports['A'].offset, (-7, 0))
class StrategyTieTool(PlanningOnlyTool):
def __init__(self) -> None:
self.seen_kwargs: list[dict[str, Any]] = []
def primitive_offers(
self,
kind: Literal['straight', 'bend', 's', 'u'],
*,
in_ptype: str | None = None,
out_ptype: str | None = None,
**kwargs: Any,
) -> tuple[PrimitiveOffer, ...]:
self.seen_kwargs.append(dict(kwargs))
endpoint_ptype = out_ptype or in_ptype
if kind == 'straight':
return (StraightOffer(
in_ptype=in_ptype,
out_ptype=endpoint_ptype,
**offer_callbacks(lambda length: (
Port((length, 0), rotation=pi, ptype=endpoint_ptype),
{'kind': 'straight', 'length': length},
)),
),)
if kind == 's':
return (SOffer(
in_ptype=in_ptype,
out_ptype=endpoint_ptype,
**offer_callbacks(lambda jog: (
Port((3, jog), rotation=pi, ptype=endpoint_ptype),
{'kind': 's', 'jog': jog},
)),
),)
return ()
def pather_with_strategy_tool(
planner: RoutingPlanner | None = None,
) -> tuple[Pather, StrategyTieTool]:
tool = StrategyTieTool()
pather = Pather(Library(), tools=tool, planner=planner, render='deferred')
pather.ports['A'] = Port((0, 0), rotation=0, ptype='wire')
return pather, tool
def test_pather_route_strategy_defaults_to_straight_first() -> None:
pather, _tool = pather_with_strategy_tool()
pather.jog('A', 4, length=10)
assert [step.data['kind'] for step in pather._paths['A']] == ['straight', 's']
assert pather._paths['A'][0].data['length'] == 7
def test_pather_route_strategy_uses_planner_default() -> None:
pather, _tool = pather_with_strategy_tool(RoutingPlanner(strategy='turn_first'))
pather.jog('A', 4, length=10)
assert [step.data['kind'] for step in pather._paths['A']] == ['s', 'straight']
assert pather._paths['A'][1].data['length'] == 7
def test_pather_route_strategy_per_route_overrides_planner_default() -> None:
pather, _tool = pather_with_strategy_tool(RoutingPlanner(strategy='turn_first'))
pather.jog('A', 4, length=10, strategy='straight_first')
assert [step.data['kind'] for step in pather._paths['A']] == ['straight', 's']
def test_pather_route_strategy_per_route_can_request_turn_first() -> None:
pather, _tool = pather_with_strategy_tool()
pather.jog('A', 4, length=10, strategy='turn_first')
assert [step.data['kind'] for step in pather._paths['A']] == ['s', 'straight']
def test_pather_route_strategy_is_not_forwarded_to_tool() -> None:
pather, tool = pather_with_strategy_tool()
pather.jog('A', 4, length=10, strategy='turn_first', marker='sentinel')
assert tool.seen_kwargs
assert all('strategy' not in kwargs for kwargs in tool.seen_kwargs)
assert any(kwargs.get('marker') == 'sentinel' for kwargs in tool.seen_kwargs)
def test_pather_route_strategy_rejects_invalid_values() -> None:
with pytest.raises(BuildError, match='Invalid route strategy'):
RoutingPlanner(strategy='sideways')
pather, _tool = pather_with_strategy_tool()
with pytest.raises(BuildError, match='Invalid route strategy'):
pather.jog('A', 4, length=10, strategy='sideways')
def test_solver_rejects_rotation_impossible_candidates_before_parameter_solving() -> None:
invalid_parameters: list[float] = []

View file

@ -76,7 +76,7 @@ def test_deferred_render_jog_uses_lowest_cost_two_bend_route(deferred_render_set
rp.render()
path_shape = cast("Path", rp.pattern.shapes[(1, 0)][0])
assert_allclose(path_shape.vertices, [[0, 0], [0, -1], [1, -1], [3, -1], [4, -1], [4, -2], [4, -10]], atol=1e-10)
assert_allclose(path_shape.vertices, [[0, 0], [0, -8], [0, -9], [1, -9], [3, -9], [4, -9], [4, -10]], atol=1e-10)
assert_allclose(rp.ports["start"].offset, [4, -10], atol=1e-10)
def test_deferred_render_mirror_preserves_planned_bend_geometry(deferred_render_setup: tuple[Pather, PathTool, Library]) -> None: