diff --git a/masque/builder/pather.py b/masque/builder/pather.py index 89bff0e..cbf87f4 100644 --- a/masque/builder/pather.py +++ b/masque/builder/pather.py @@ -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) diff --git a/masque/builder/planner/__init__.py b/masque/builder/planner/__init__.py index fc4d22a..f2df9e6 100644 --- a/masque/builder/planner/__init__.py +++ b/masque/builder/planner/__init__.py @@ -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 diff --git a/masque/builder/planner/planner.py b/masque/builder/planner/planner.py index f7c3cca..913758e 100644 --- a/masque/builder/planner/planner.py +++ b/masque/builder/planner/planner.py @@ -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) diff --git a/masque/test/test_pather_primitive_offers.py b/masque/test/test_pather_primitive_offers.py index 7d99355..d1945c1 100644 --- a/masque/test/test_pather_primitive_offers.py +++ b/masque/test/test_pather_primitive_offers.py @@ -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] = [] diff --git a/masque/test/test_pather_rendering.py b/masque/test/test_pather_rendering.py index c5ddd68..186faf4 100644 --- a/masque/test/test_pather_rendering.py +++ b/masque/test/test_pather_rendering.py @@ -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: