diff --git a/examples/tutorial/pather.py b/examples/tutorial/pather.py index 280f9e5..7813529 100644 --- a/examples/tutorial/pather.py +++ b/examples/tutorial/pather.py @@ -178,7 +178,7 @@ class PrimitiveWireTool(Tool): def _transition_offers(self, in_ptype: str | None) -> tuple[StraightOffer, ...]: offers: list[StraightOffer] = [] - for index, spec in enumerate(self.transitions): + for spec in self.transitions: if spec.out_port.ptype != self.ptype: continue if in_ptype not in (None, 'unk', spec.in_port.ptype): @@ -208,7 +208,6 @@ class PrimitiveWireTool(Tool): offers.append(StraightOffer( in_ptype = spec.in_port.ptype, out_ptype = spec.out_port.ptype, - priority_bias = index * 1e7, length_domain = (length, length), endpoint_planner = endpoint_planner, commit_planner = commit_planner, @@ -220,7 +219,7 @@ class PrimitiveWireTool(Tool): return () offers: list[StraightOffer] = [] - for index, spec in enumerate(self.transitions): + for spec in self.transitions: if spec.in_port.ptype != self.ptype: continue if out_ptype is not None and spec.out_port.ptype != out_ptype: @@ -255,7 +254,6 @@ class PrimitiveWireTool(Tool): offers.append(StraightOffer( in_ptype = self.ptype, out_ptype = spec.out_port.ptype, - priority_bias = index * 1e7, length_domain = (transition_length, numpy.inf), endpoint_planner = endpoint_planner, commit_planner = commit_planner, diff --git a/masque/builder/__init__.py b/masque/builder/__init__.py index eba23ba..4a92753 100644 --- a/masque/builder/__init__.py +++ b/masque/builder/__init__.py @@ -67,6 +67,7 @@ from .tools import ( PathTool as PathTool, RenderStep as RenderStep, PrimitiveKind as PrimitiveKind, + CostCallable as CostCallable, GeneratedEndpointFn as GeneratedEndpointFn, PrimitiveOffer as PrimitiveOffer, StraightOffer as StraightOffer, diff --git a/masque/builder/planner/planner.py b/masque/builder/planner/planner.py index 8f483a7..1d5837a 100644 --- a/masque/builder/planner/planner.py +++ b/masque/builder/planner/planner.py @@ -350,11 +350,16 @@ class Solver: ) def offer_key(offer: PrimitiveOffer) -> tuple[Any, ...]: + cost_key: tuple[str, float | int] + if callable(offer.cost): + cost_key = ('callable', id(offer.cost)) + else: + cost_key = ('factor', round(float(offer.cost), 9)) return ( type(offer).__qualname__, offer.in_ptype, offer.out_ptype, - round(float(offer.priority_bias), 9), + cost_key, tuple(round(float(value), 9) for value in offer.parameter_domain), getattr(offer, 'ccw', None), id(offer.endpoint_planner), diff --git a/masque/builder/tools.py b/masque/builder/tools.py index 9967e62..588c8f0 100644 --- a/masque/builder/tools.py +++ b/masque/builder/tools.py @@ -112,8 +112,32 @@ CommitCallable = Callable[[float], Any] BBoxCallable = Callable[[float], NDArray[numpy.float64]] DataCallable = Callable[[float], Any] BBoxForDataCallable = Callable[[Any], NDArray[numpy.float64]] +CostCallable = Callable[[float, Port], float] PrimitiveKind = Literal['straight', 'bend', 's', 'u'] -BUILTIN_PRIORITY_STEP = 1e7 + + +def _validated_cost(cost: CostCallable | float) -> CostCallable | float: + """Return a canonical primitive cost policy or raise `BuildError`.""" + if callable(cost): + return cost + try: + factor = float(cost) + except (TypeError, ValueError, OverflowError) as err: + raise BuildError(f'Primitive cost factor must be a number or callable, got {cost!r}') from err + if not scalar_isfinite(factor) or factor < 0: + raise BuildError(f'Primitive cost factor must be nonnegative and finite, got {factor:g}') + return factor + + +def _validated_cost_value(value: Any) -> float: + """Return a canonical evaluated cost or raise `BuildError`.""" + try: + cost = float(value) + except (TypeError, ValueError, OverflowError) as err: + raise BuildError(f'Primitive cost must be a number, got {value!r}') from err + if not scalar_isfinite(cost) or cost < 0: + raise BuildError(f'Primitive cost must be nonnegative and finite, got {cost:g}') + return cost def _generated_offer_callbacks( @@ -188,7 +212,7 @@ class PrimitiveOffer(ABC): """ in_ptype: str | None out_ptype: str | None - priority_bias: float = 0.0 + cost: CostCallable | float = 1.0 bbox_planner: BBoxCallable | None = None parameterized_bbox: Any | None = None """Reserved footprint metadata; the current solver does not inspect it.""" @@ -201,8 +225,7 @@ class PrimitiveOffer(ABC): if has_endpoint != has_commit: raise BuildError('PrimitiveOffer split callbacks require both endpoint_planner and commit_planner') - if not numpy.isfinite(self.priority_bias) or self.priority_bias < 0: - raise BuildError(f'PrimitiveOffer priority_bias must be nonnegative and finite, got {self.priority_bias:g}') + object.__setattr__(self, 'cost', _validated_cost(self.cost)) @property @abstractmethod @@ -234,15 +257,21 @@ class PrimitiveOffer(ABC): """ Return this primitive's additive planning cost. - Lower cost is preferred before internal tie-breakers. The default cost - is based on local endpoint displacement plus `priority_bias`. + Lower cost is preferred before internal tie-breakers. A numeric `cost` + scales the default local-displacement cost; a callable receives the + canonical parameter and local endpoint and returns the complete cost. """ selected = self.canonicalize_parameter(parameter) if self.endpoint_planner is not None: out_port = self.endpoint_planner(selected) else: out_port = self.endpoint_at(selected) - return self.priority_bias + abs(float(out_port.x)) + (pi / 2) * abs(float(out_port.y)) + if callable(self.cost): + value = self.cost(selected, out_port) + else: + default_cost = abs(float(out_port.x)) + (pi / 2) * abs(float(out_port.y)) + value = self.cost * default_cost + return _validated_cost_value(value) def bbox_at(self, parameter: float) -> NDArray[numpy.float64]: """ @@ -294,7 +323,7 @@ class StraightOffer(PrimitiveOffer): ptype: str | None, data_at: DataCallable, *, - priority_bias: float = 0.0, + cost: CostCallable | float = 1.0, bbox_for_data: BBoxForDataCallable | None = None, length_domain: tuple[float, float] = (0.0, numpy.inf), ) -> Self: @@ -312,7 +341,7 @@ class StraightOffer(PrimitiveOffer): return cls( in_ptype = ptype, out_ptype = ptype, - priority_bias = priority_bias, + cost = cost, bbox_planner = bbox_planner, endpoint_planner = endpoint_planner, commit_planner = commit_planner, @@ -327,7 +356,7 @@ class StraightOffer(PrimitiveOffer): endpoint: Port, data: Any, *, - priority_bias: float = 0.0, + cost: CostCallable | float = 1.0, bbox_for_data: BBoxForDataCallable | None = None, ) -> Self: """ @@ -341,7 +370,7 @@ class StraightOffer(PrimitiveOffer): return cls( in_ptype = in_ptype, out_ptype = out_ptype, - priority_bias = priority_bias, + cost = cost, bbox_planner = bbox_planner, endpoint_planner = endpoint_planner, commit_planner = commit_planner, @@ -371,7 +400,7 @@ class BendOffer(PrimitiveOffer): data_at: DataCallable, *, ccw: bool, - priority_bias: float = 0.0, + cost: CostCallable | float = 1.0, bbox_for_data: BBoxForDataCallable | None = None, length_domain: tuple[float, float] = (0.0, numpy.inf), ) -> Self: @@ -386,7 +415,7 @@ class BendOffer(PrimitiveOffer): return cls( in_ptype = ptype, out_ptype = ptype, - priority_bias = priority_bias, + cost = cost, bbox_planner = bbox_planner, endpoint_planner = endpoint_planner, commit_planner = commit_planner, @@ -403,7 +432,7 @@ class BendOffer(PrimitiveOffer): data: Any, *, ccw: bool, - priority_bias: float = 0.0, + cost: CostCallable | float = 1.0, bbox_for_data: BBoxForDataCallable | None = None, ) -> Self: """ @@ -417,7 +446,7 @@ class BendOffer(PrimitiveOffer): return cls( in_ptype = in_ptype, out_ptype = out_ptype, - priority_bias = priority_bias, + cost = cost, bbox_planner = bbox_planner, endpoint_planner = endpoint_planner, commit_planner = commit_planner, @@ -446,7 +475,7 @@ class SOffer(PrimitiveOffer): endpoint_at: EndpointCallable, data_at: DataCallable, *, - priority_bias: float = 0.0, + cost: CostCallable | float = 1.0, bbox_for_data: BBoxForDataCallable | None = None, jog_domain: tuple[float, float] = (-numpy.inf, numpy.inf), ) -> Self: @@ -461,7 +490,7 @@ class SOffer(PrimitiveOffer): return cls( in_ptype = ptype, out_ptype = ptype, - priority_bias = priority_bias, + cost = cost, bbox_planner = bbox_planner, endpoint_planner = endpoint_planner, commit_planner = commit_planner, @@ -476,7 +505,7 @@ class SOffer(PrimitiveOffer): endpoint: Port, data: Any, *, - priority_bias: float = 0.0, + cost: CostCallable | float = 1.0, bbox_for_data: BBoxForDataCallable | None = None, ) -> Self: """ @@ -490,7 +519,7 @@ class SOffer(PrimitiveOffer): return cls( in_ptype = in_ptype, out_ptype = out_ptype, - priority_bias = priority_bias, + cost = cost, bbox_planner = bbox_planner, endpoint_planner = endpoint_planner, commit_planner = commit_planner, @@ -518,7 +547,7 @@ class UOffer(PrimitiveOffer): endpoint_at: EndpointCallable, data_at: DataCallable, *, - priority_bias: float = 0.0, + cost: CostCallable | float = 1.0, bbox_for_data: BBoxForDataCallable | None = None, jog_domain: tuple[float, float] = (-numpy.inf, numpy.inf), ) -> Self: @@ -533,7 +562,7 @@ class UOffer(PrimitiveOffer): return cls( in_ptype = ptype, out_ptype = ptype, - priority_bias = priority_bias, + cost = cost, bbox_planner = bbox_planner, endpoint_planner = endpoint_planner, commit_planner = commit_planner, @@ -548,7 +577,7 @@ class UOffer(PrimitiveOffer): endpoint: Port, data: Any, *, - priority_bias: float = 0.0, + cost: CostCallable | float = 1.0, bbox_for_data: BBoxForDataCallable | None = None, ) -> Self: """ @@ -562,7 +591,7 @@ class UOffer(PrimitiveOffer): return cls( in_ptype = in_ptype, out_ptype = out_ptype, - priority_bias = priority_bias, + cost = cost, bbox_planner = bbox_planner, endpoint_planner = endpoint_planner, commit_planner = commit_planner, @@ -764,12 +793,14 @@ class AutoTool(Tool): """ A routing tool assembled from reusable path primitives. - `AutoTool` chooses among prioritized straight generators, pre-rendered bends, - optional generated S-bend primitives, pre-rendered U-turns, and + `AutoTool` chooses among straight generators, pre-rendered bends, optional + generated S-bend primitives, pre-rendered U-turns, and pre-rendered transitions registered through `add_straight()`, `add_bend()`, `add_sbend()`, `add_uturn()`, and `add_transition()`. - Registration call order defines primitive priority. + Primitive selection uses each registration's explicit `cost` policy rather + than registration order. Numeric costs scale the default geometric cost; + callable costs replace it. Straight and bend offers use one straight and, if turning, one bend. `add_sbend()` exposes generated S-bend primitives. `add_uturn()` exposes @@ -972,6 +1003,7 @@ class AutoTool(Tool): in_port_name: str | None = None, *, length_range: tuple[float, float] = (0, numpy.inf), + cost: CostCallable | float = 1.0, ) -> Self: """ Register a generated straight primitive. @@ -980,8 +1012,8 @@ class AutoTool(Tool): generated and the missing metadata is inferred from an equivalent two-port straight. Metadata inference does not receive route options. """ + cost = _validated_cost(cost) ptype, in_port_name = self._infer_straight_metadata(fn, length_range, ptype, in_port_name) - priority_bias = len(self._straight_offers) * BUILTIN_PRIORITY_STEP def data_at(length: float) -> AutoTool.GeneratedData: return self.GeneratedData(fn, in_port_name, length) @@ -989,7 +1021,7 @@ class AutoTool(Tool): self._straight_offers.append(StraightOffer.generated( ptype, data_at, - priority_bias = priority_bias, + cost = cost, bbox_for_data = self._bbox_for_data, length_domain = length_range, )) @@ -1003,6 +1035,7 @@ class AutoTool(Tool): *, clockwise: bool | None = None, mirror: bool = True, + cost: CostCallable | float = 1.0, ) -> Self: """ Register a reusable L-bend primitive. @@ -1011,6 +1044,7 @@ class AutoTool(Tool): direction is inferred from the selected port orientations; `clockwise`, when provided, is checked against that inferred direction. """ + cost = _validated_cost(cost) if (in_port_name is None) != (out_port_name is None): raise BuildError('Bend port names must be provided together') if in_port_name is None or out_port_name is None: @@ -1019,7 +1053,6 @@ class AutoTool(Tool): raise BuildError(f'Bend port names are required for {len(port_names)}-port abstracts') in_port_name, out_port_name = port_names - priority_bias = len(self._bend_offers[0]) * BUILTIN_PRIORITY_STEP in_port = abstract.ports[in_port_name] out_port = abstract.ports[out_port_name] out_ptype = out_port.ptype @@ -1044,7 +1077,7 @@ class AutoTool(Tool): endpoint = endpoint, data = reusable_data, ccw = ccw, - priority_bias = priority_bias, + cost = cost, bbox_for_data = self._bbox_for_data, )) return self @@ -1058,6 +1091,7 @@ class AutoTool(Tool): *, jog_range: tuple[float, float] = (0, numpy.inf), endpoint: GeneratedEndpointFn | None = None, + cost: CostCallable | float = 1.0, ) -> Self: """ Register a generated S-bend primitive. @@ -1070,6 +1104,7 @@ class AutoTool(Tool): and the missing metadata is inferred from an equivalent two-port S-bend. Metadata inference and endpoint planning do not receive route options. """ + cost = _validated_cost(cost) ptype, in_port_name, out_port_name = self._infer_sbend_metadata( fn, jog_range, @@ -1090,8 +1125,6 @@ class AutoTool(Tool): return out_port for jog_domain in self._signed_jog_domains(jog_range): - priority_bias = len(self._s_offers) * BUILTIN_PRIORITY_STEP - def data_at(jog: float) -> AutoTool.GeneratedData: return self.GeneratedData( fn, @@ -1104,7 +1137,7 @@ class AutoTool(Tool): ptype, endpoint_at, data_at, - priority_bias = priority_bias, + cost = cost, bbox_for_data = self._bbox_for_data, jog_domain = jog_domain, )) @@ -1117,10 +1150,12 @@ class AutoTool(Tool): out_port_name: str, *, mirror: bool = True, + cost: CostCallable | float = 1.0, ) -> Self: """ Register a reusable U-turn primitive. """ + cost = _validated_cost(cost) in_port = abstract.ports[in_port_name] out_port = abstract.ports[out_port_name] dxy, angle = in_port.measure_travel(out_port) @@ -1140,7 +1175,6 @@ class AutoTool(Tool): mirrored: bool, ) -> None: reusable_data = self.ReusableData(abstract, in_port_name, mirrored) - priority_bias = len(self._u_offers) * BUILTIN_PRIORITY_STEP endpoint = Port((length, offer_jog), rotation=0, ptype=out_ptype) self._u_offers.append(UOffer.prebuilt( @@ -1148,7 +1182,7 @@ class AutoTool(Tool): out_ptype = out_ptype, endpoint = endpoint, data = reusable_data, - priority_bias = priority_bias, + cost = cost, bbox_for_data = self._bbox_for_data, )) @@ -1164,6 +1198,7 @@ class AutoTool(Tool): our_port_name: str | None = None, *, one_way: bool = False, + cost: CostCallable | float = 1.0, ) -> Self: """ Register a reusable port-type transition and expose it as router-visible adapter offers. @@ -1171,6 +1206,7 @@ class AutoTool(Tool): If the transition has exactly two ports and is bidirectional, port names may be omitted. """ + cost = _validated_cost(cost) if (their_port_name is None) != (our_port_name is None): raise BuildError('Transition port names must be provided together') if their_port_name is None or our_port_name is None: @@ -1181,9 +1217,9 @@ class AutoTool(Tool): raise BuildError(f'Transition port names are required for {len(port_names)}-port abstracts') their_port_name, our_port_name = port_names - self._add_transition_direction(abstract, their_port_name, our_port_name) + self._add_transition_direction(abstract, their_port_name, our_port_name, cost) if not one_way: - self._add_transition_direction(abstract, our_port_name, their_port_name) + self._add_transition_direction(abstract, our_port_name, their_port_name, cost) return self @staticmethod @@ -1261,6 +1297,7 @@ class AutoTool(Tool): abstract: Abstract, their_port_name: str, our_port_name: str, + cost: CostCallable | float, ) -> None: their_port = abstract.ports[their_port_name] our_port = abstract.ports[our_port_name] @@ -1286,14 +1323,13 @@ class AutoTool(Tool): endpoint = Port((dx, dy), rotation=pi, ptype=our_port.ptype) offers = self._transition_adapter_offers_by_key.setdefault((kind, in_key), []) - priority_bias = len(offers) * BUILTIN_PRIORITY_STEP if kind == 'straight': offers.append(StraightOffer.prebuilt( in_ptype = their_port.ptype, out_ptype = our_port.ptype, endpoint = endpoint, data = transition_data, - priority_bias = priority_bias, + cost = cost, bbox_for_data = self._bbox_for_data, )) return @@ -1303,7 +1339,7 @@ class AutoTool(Tool): out_ptype = our_port.ptype, endpoint = endpoint, data = transition_data, - priority_bias = priority_bias, + cost = cost, bbox_for_data = self._bbox_for_data, ))