diff --git a/masque/__init__.py b/masque/__init__.py index defd1ef..b5dddde 100644 --- a/masque/__init__.py +++ b/masque/__init__.py @@ -82,10 +82,6 @@ from .abstract import Abstract as Abstract from .builder import ( Tool as Tool, Pather as Pather, - RouteError as RouteError, - RouteFailureDetails as RouteFailureDetails, - RouteFailurePolicy as RouteFailurePolicy, - MinimumStatus as MinimumStatus, RenderStep as RenderStep, AutoTool as AutoTool, PathTool as PathTool, diff --git a/masque/builder/__init__.py b/masque/builder/__init__.py index eba23ba..5bd3914 100644 --- a/masque/builder/__init__.py +++ b/masque/builder/__init__.py @@ -8,23 +8,16 @@ optional footprint metadata, and a commit hook for producing tool-specific render data after a concrete parameter has been selected. `Pather` owns user-facing route operations such as `trace()`, `jog()`, -`uturn()`, and `trace_into()`. The internal planner resolves each operation into -one or more `SolverRequest`s. This normalization is why the public routing API -can remain a convenient keyword-based interface without making the solver -stringly typed internally. A pure solver search selects a `Candidate`, and a -`RouteLeg` attaches that candidate to its copied source port and Tool. Only -after selection succeeds are the chosen offers materialized through -`offer.commit(parameter)` into a `PreparedRouteResult` containing -`RenderStep.data`. `Pather` then applies that prepared result to its live ports -and pending render queue. +`uturn()`, and `trace_into()`. For each operation, it asks the active `Tool` for +offers and passes those offers plus route constraints to the internal router. +The router selects a sequence of internal selected primitives, each pairing an +offer with a concrete parameter, endpoint, and cost. -Selection is pure with respect to caller-owned Pattern, Library, and Pather -state. Tool offer discovery and endpoint/cost/bbox callbacks must likewise be -deterministic and must not mutate that state. `commit()` is the first -selected-offer materialization hook, but it still must not mutate the live -layout. `Tool.render()` is the geometry mutation boundary: later, -`Pather.render()` batches compatible `RenderStep`s and inserts the resulting -geometry into the Pattern and Library. +Route commit is separate from route selection. Once a route is selected, +`Pather` calls `offer.commit(parameter)` only for the selected primitives and +stores the returned opaque tool payload in `RenderStep.data`. Later, +`Pather.render()` batches compatible `RenderStep`s and calls `Tool.render()` to +turn those committed payloads into geometry. `PrimitiveOffer` and `RenderStep.data` are the tool-facing contract. `RenderStep` is `Pather`'s deferred-render record, and @@ -38,11 +31,6 @@ The practical layering is: - Pather applies the prepared result to ports, deferred render queues, and the target pattern/library. -`Pather` intentionally remains a Pattern-oriented facade rather than exposing -separate assembly and routing objects: its user model is a working Pattern with -routing tools attached. The ownership phases above are internal boundaries, -not additional objects callers must coordinate. - Code outside the builder package should prefer the exports here over importing from `masque.builder.planner`. The planner package is intentionally available for tests and internal maintenance, but it is not the compatibility boundary @@ -53,13 +41,6 @@ from .pather import ( Pather as Pather, PortPather as PortPather, ) -from .error import ( - RouteError as RouteError, - RouteFailureDetails as RouteFailureDetails, - RouteOperation as RouteOperation, - RouteFailurePolicy as RouteFailurePolicy, - MinimumStatus as MinimumStatus, -) from .utils import ell as ell from .tools import ( Tool as Tool, diff --git a/masque/builder/error.py b/masque/builder/error.py deleted file mode 100644 index 6a647ce..0000000 --- a/masque/builder/error.py +++ /dev/null @@ -1,102 +0,0 @@ -"""Public routing failure diagnostics.""" -from typing import Any, Literal -from collections.abc import Mapping -from dataclasses import dataclass -from enum import Enum, auto -from pprint import pformat -from types import MappingProxyType - -from ..error import BuildError - - -RouteOperation = Literal['trace', 'trace_to', 'jog', 'uturn'] - - -class RouteFailurePolicy(Enum): - """Whether route failure may be recovered through alternate/dead planning. - - `RECOVERABLE` means a caller-controlled fallback may try another planning - branch. `FATAL` marks an invalid request or broken planning contract that - must be reported directly. - """ - - RECOVERABLE = auto() - FATAL = auto() - - -class MinimumStatus(Enum): - """Outcome of preferred-minimum-length diagnosis. - - `NOT_EVALUATED` is used when diagnosis is inapplicable, notably for an - invalid resolved length. `FOUND` carries `minimum_length`; `NO_ROUTE` means - exhaustive planning found no legal unconstrained route; `FAILED` means the - secondary diagnostic calculation itself raised a recoverable error. - """ - - NOT_EVALUATED = auto() - FOUND = auto() - NO_ROUTE = auto() - FAILED = auto() - - -@dataclass(frozen=True, slots=True) -class RouteFailureDetails: - """Structured context for a failed Pather routing request.""" - - operation: RouteOperation - portspec: str - in_ptype: str | None - out_ptype: str | None - request: Mapping[str, Any] - resolved_length: float | None - resolved_jog: float | None - minimum_length: float | None - minimum_status: MinimumStatus - cause: str - minimum_cause: str | None = None - - def __post_init__(self) -> None: - if self.minimum_status is MinimumStatus.FOUND and self.minimum_length is None: - raise BuildError('MinimumStatus.FOUND requires minimum_length') - if self.minimum_status is not MinimumStatus.FOUND and self.minimum_length is not None: - raise BuildError(f'{self.minimum_status} requires minimum_length=None') - object.__setattr__(self, 'request', MappingProxyType(dict(self.request))) - - -class RouteError(BuildError): - """A route-selection failure with structured request diagnostics.""" - - details: RouteFailureDetails - policy: RouteFailurePolicy - - def __init__( - self, - details: RouteFailureDetails, - *, - policy: RouteFailurePolicy = RouteFailurePolicy.RECOVERABLE, - ) -> None: - self.details = details - self.policy = policy - if details.minimum_status is MinimumStatus.NOT_EVALUATED: - minimum = 'not evaluated' - elif details.minimum_status is MinimumStatus.FOUND: - assert details.minimum_length is not None - minimum = f'{details.minimum_length:g}' - elif details.minimum_status is MinimumStatus.NO_ROUTE: - minimum = 'unavailable (no legal route exists at any length)' - else: - minimum = 'unavailable (minimum-length calculation failed)' - - lines = [ - f'Unable to plan {details.operation} route for port {details.portspec!r}:', - f' in_ptype: {details.in_ptype!r}', - f' out_ptype: {details.out_ptype!r}', - f' request: {pformat(dict(details.request), compact=True)}', - f' resolved_length: {details.resolved_length!r}', - f' resolved_jog: {details.resolved_jog!r}', - f' preferred_minimum_length: {minimum}', - f' cause: {details.cause}', - ] - if details.minimum_cause is not None: - lines.append(f' minimum_failure: {details.minimum_cause}') - super().__init__('\n'.join(lines)) diff --git a/masque/builder/pather.py b/masque/builder/pather.py index 9d6b46b..cbf87f4 100644 --- a/masque/builder/pather.py +++ b/masque/builder/pather.py @@ -22,25 +22,10 @@ Routing is split into four ownership phases: This split keeps selection failures largely transactional for live Pather state: unsupported primitive combinations can fail before the Pattern, pending -step queue, or Library are touched. Once prepared actions are applied, plug, -rename, render, or insertion failures may leave partial output; that mutation -boundary belongs to Pather, not to Tool implementations or the route solver. - -Routing policy follows port names. Port-specific Tool assignments and -`PortPather` selections remain attached to their names rather than following a -physical port through arbitrary renames. Pending `RenderStep`s are different: -they snapshot their geometric endpoints when planned. The `_paths` keys are -rendering buckets used for ordering and batching those historical steps, not -identities that can retarget their saved geometry when a name is deleted or -reused. - -While pending steps exist, mutate ports through Pather's methods. Directly -editing `pather.pattern.ports` bypasses the bookkeeping that maintains render -buckets and is unsupported. - -Rendering is not transactional across Pattern and Library mutations. A render -exception is terminal for that Pather: callers may catch it for reporting or -cleanup, but must not retry rendering or continue routing with the same object. +step queue, or Library are touched. Once selected offers are committed and +prepared actions are applied, later commit, plug, rename, render, or insertion +failures may leave partial output; that mutation boundary belongs to Pather, +not to Tool implementations or the route solver. """ from typing import Self, Any, Literal, overload from collections.abc import Iterator, Iterable, Mapping, MutableMapping, Sequence @@ -70,9 +55,8 @@ from .tools import ( from .planner.interface import ( PreparedRouteResult, RoutePortContext, - route_failure_policy, + route_error_is_fatal, ) -from .error import RouteFailurePolicy from .planner import ( RouteTieBreakStrategy, RoutingPlanner, @@ -132,9 +116,6 @@ class Pather(PortList): """ Tool objects used to dynamically generate new routing segments. A key of `None` indicates the default `Tool`. - - Non-`None` keys are policies attached to port names. Renaming a port does - not transfer its Tool assignment to the new name. """ planner: RoutingPlanner @@ -178,16 +159,6 @@ class Pather(PortList): def _route_contexts(self, portspecs: Sequence[str]) -> tuple[RoutePortContext, ...]: """Snapshot several ports in request order for bundle planning.""" - if not portspecs: - raise BuildError('Routing requires at least one port') - seen: set[str] = set() - duplicates: set[str] = set() - for portspec in portspecs: - if portspec in seen: - duplicates.add(portspec) - seen.add(portspec) - if duplicates: - raise BuildError(f'Routing port names must be unique; got duplicates: {sorted(duplicates)}') return tuple(self._route_context(portspec) for portspec in portspecs) @property @@ -574,7 +545,7 @@ class Pather(PortList): try: 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_failure_policy(err) is RouteFailurePolicy.FATAL: + if not self._dead or route_error_is_fatal(err): raise if length is not None and len(contexts) == 1: context = contexts[0] @@ -631,11 +602,7 @@ class Pather(PortList): try: 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_failure_policy(err) is RouteFailurePolicy.FATAL - ): + if not self._dead or len(contexts) != 1 or route_error_is_fatal(err): raise if bounds.get('length') is not None: length = bounds['length'] @@ -707,11 +674,7 @@ class Pather(PortList): try: 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_failure_policy(err) is RouteFailurePolicy.FATAL - ): + if not self._dead or len(contexts) != 1 or route_error_is_fatal(err): raise if numpy.isclose(offset, 0): if length is None: @@ -777,7 +740,7 @@ class Pather(PortList): not self._dead or len(contexts) != 1 or length is None - or route_failure_policy(err) is RouteFailurePolicy.FATAL + or route_error_is_fatal(err) ): raise context = contexts[0] @@ -857,11 +820,6 @@ class Pather(PortList): Consecutive compatible `RenderStep`s are batched by port and Tool, then passed to `Tool.render()`. After insertion, the rendered output port is checked against the endpoint that planning selected. - - Rendering may modify the Library before every batch has completed and - does not provide rollback. If this method raises, the Pather must be - treated as unusable; retrying or continuing to route with it is - unsupported. """ with self._logger.log_operation(self, 'render', None, append=append): tool_port_names = ('A', 'B') @@ -934,7 +892,7 @@ class Pather(PortList): batch: list[RenderStep] = [] for step in steps: appendable = step.opcode in ('L', 'S', 'U') - same_tool = batch and step.tool is batch[0].tool + same_tool = batch and step.tool == batch[0].tool if batch and (not appendable or not same_tool or not batch[-1].is_continuous_with(step)): render_batch(portspec, batch, append) batch = [] @@ -1014,13 +972,7 @@ class Pather(PortList): class PortPather: - """ - Port-name selection for fluent pathing. - - The selection stores names, not stable physical port identities. Its own - rename/delete helpers update the selection, but unrelated changes made - through the parent Pather do not retarget it. - """ + """ Port state manager for fluent pathing. """ def __init__( self, ports: str | Iterable[str], diff --git a/masque/builder/planner/__init__.py b/masque/builder/planner/__init__.py index 41c53d6..f2df9e6 100644 --- a/masque/builder/planner/__init__.py +++ b/masque/builder/planner/__init__.py @@ -10,7 +10,7 @@ from .interface import ( PreparedRouteResult as PreparedRouteResult, RoutePlanningError as RoutePlanningError, RoutePortContext as RoutePortContext, - route_failure_policy as route_failure_policy, + 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/interface.py b/masque/builder/planner/interface.py index 514779a..58028df 100644 --- a/masque/builder/planner/interface.py +++ b/masque/builder/planner/interface.py @@ -15,30 +15,21 @@ from dataclasses import dataclass from ...error import BuildError from ...ports import Port from ..tools import RenderStep, Tool -from ..error import RouteError, RouteFailurePolicy class RoutePlanningError(BuildError): """Route-planning error with fallback policy metadata.""" - policy: RouteFailurePolicy + fatal: bool - def __init__( - self, - *args: object, - policy: RouteFailurePolicy = RouteFailurePolicy.RECOVERABLE, - ) -> None: + def __init__(self, *args: object, fatal: bool = False) -> None: super().__init__(*args) - self.policy = policy + self.fatal = fatal -def route_failure_policy(err: Exception) -> RouteFailurePolicy: - """Return typed route recovery policy, defaulting generic errors to recoverable.""" - if isinstance(err, RoutePlanningError): - return err.policy - if isinstance(err, RouteError): - return err.policy - return RouteFailurePolicy.RECOVERABLE +def route_error_is_fatal(err: Exception) -> bool: + """Return true when a planning error should bypass dead-Pather fallback.""" + return bool(getattr(err, 'fatal', False)) @dataclass(frozen=True, slots=True) @@ -63,10 +54,9 @@ class PreparedRouteAction: """ Prepared mutation for one routed Pather port. - Pure selection has already completed, and the planner has materialized the - selected primitive offers into `render_steps` and computed the final live - port. `plug_into`, when set, names the destination port to consume after the - route endpoint is applied. + The planner has already committed selected primitive offers into + `render_steps` and computed the final live port. `plug_into`, when set, + names the destination port to consume after the route endpoint is applied. """ portspec: str """Live Pather port name to update.""" @@ -83,10 +73,9 @@ class PreparedRouteResult: """ Complete prepared result for one Pather routing operation. - `actions` contain materialized, committed render data and are applied first. - `renames` are deferred until after all route actions so trace-into/thru - behavior can be represented without exposing the solver's selected - primitive sequence to Pather. + `actions` are applied first. `renames` are deferred until after all route + actions so trace-into/thru behavior can be represented without exposing the + solver's selected primitive sequence to Pather. """ actions: tuple[PreparedRouteAction, ...] """Prepared per-port route mutations.""" diff --git a/masque/builder/planner/planner.py b/masque/builder/planner/planner.py index 6e6a3c0..913758e 100644 --- a/masque/builder/planner/planner.py +++ b/masque/builder/planner/planner.py @@ -6,19 +6,11 @@ Tool primitive offers. `Pather` passes copied `RoutePortContext` snapshots here; the planner returns `PreparedRouteResult` records that describe pending mutations without applying them to the live Pattern. -Public routing modes and bounds are normalized by `bounds.py` into per-leg -`SolverRequest` values. `Solver` performs the pure search: it queries Tool -offers, enumerates bounded primitive compositions, inserts ptype adapters, -solves primitive parameters, and returns the ranked `Candidate`. A `RouteLeg` -then attaches the candidate to its copied source port and Tool. Preparation is -the layer above selection: only chosen offers are committed into `RenderStep` -payloads and returned in a `PreparedRouteResult` for Pather to apply. - -These named stages are architectural boundaries even though most of their -implementation is currently co-located in this module. File layout is an -internal organization choice; the distinction between normalized request, -pure selection, prepared result, and live application is the meaningful -mutation boundary. +Public routing modes and bounds are normalized by `bounds.py` before the solver +sees them. This module plans one route intent at a time: it queries Tool offers, +enumerates bounded primitive compositions, inserts ptype adapters, solves +primitive parameters, ranks candidates, and commits only the selected offers +into `RenderStep` payloads. All search is performed in Tool-local route coordinates. The active input port is at the origin, travel is along +x, and positive jog is to the left. After a @@ -30,7 +22,7 @@ from __future__ import annotations # ruff: noqa: ANN401,PLR0912,PLR0913,PLR0915,TC001,TC002,TC003 from collections.abc import Iterable, Mapping, Sequence -from dataclasses import dataclass, replace +from dataclasses import dataclass from itertools import combinations from math import cos, isclose as math_isclose, sin from typing import Any, Literal @@ -51,13 +43,6 @@ from ..tools import ( StraightOffer, Tool, ) -from ..error import ( - MinimumStatus, - RouteError, - RouteFailureDetails, - RouteFailurePolicy, - RouteOperation, - ) from ..utils import ell from . import bounds as planner_bounds from .interface import ( @@ -65,16 +50,12 @@ from .interface import ( PreparedRouteResult, RoutePlanningError, RoutePortContext, - route_failure_policy, + route_error_is_fatal, ) RouteTieBreakStrategy = Literal['straight_first', 'turn_first'] -class NoLegalRouteError(BuildError): - """Internal marker for exhaustive candidate-selection failure.""" - - 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'): @@ -163,7 +144,7 @@ def is_adapter_offer(offer: PrimitiveOffer) -> bool: def raise_if_fatal(err: Exception) -> None: """Propagate fatal planning errors while allowing normal candidate rejection.""" - if route_failure_policy(err) is RouteFailurePolicy.FATAL: + if getattr(err, 'fatal', False): raise err @@ -245,9 +226,9 @@ class Candidate: @dataclass(frozen=True, slots=True) -class SolverRequest: +class RouteRequest: """ - Normalized input for one `Solver` invocation and one route leg. + Normalized solver input for one route leg. Public Pather calls are converted into this smaller shape before grammar enumeration. `length`, `jog`, and `out_ptype` become endpoint constraints; @@ -316,7 +297,7 @@ class Solver: enumerated so fixed and adjustable offers share the same path. """ - def __init__(self, request: SolverRequest) -> None: + def __init__(self, request: RouteRequest) -> None: self.request = request self.eval_cache: dict[tuple[int, float, str | None, str | None, str, str, PrimitiveKind | None], SelectedPrimitive] = {} self.offer_cache: dict[ @@ -410,16 +391,14 @@ class Solver: if not candidates: for err in errors: - if route_failure_policy(err) is RouteFailurePolicy.FATAL: + if getattr(err, 'fatal', False): raise err if errors: last_error = errors[-1] if self.request.route_name in str(last_error): - raise NoLegalRouteError(str(last_error)) from last_error - raise NoLegalRouteError( - f'{self.request.route_name} route is unsupported: {last_error}' - ) from last_error - raise NoLegalRouteError(f'No legal primitive offer for {self.request.route_name}') + raise last_error + raise BuildError(f'{self.request.route_name} route is unsupported: {last_error}') from last_error + raise BuildError(f'No legal primitive offer for {self.request.route_name}') return min( candidates, @@ -490,12 +469,12 @@ class Solver: if not ptypes_compatible(out_port.ptype, offer.out_ptype): raise RoutePlanningError( f'{route_name} primitive endpoint ptype does not match declared offer out_ptype', - policy=RouteFailurePolicy.FATAL, + fatal=True, ) if out_ptype is not None and not ptypes_compatible(out_port.ptype, out_ptype): raise RoutePlanningError( 'Requested out_ptype does not match primitive endpoint ptype', - policy=RouteFailurePolicy.FATAL, + fatal=True, ) cost = float(offer.cost_at(selected)) if not numpy.isfinite(cost): @@ -1007,7 +986,7 @@ class RoutingPlanner: bands.append((4, 4)) return tuple(bands) - def solver_request( + def route_request( self, family: PrimitiveKind, context: RoutePortContext, @@ -1019,9 +998,9 @@ class RoutingPlanner: max_bends: int | None = None, strategy: RouteTieBreakStrategy | str | None = None, **kwargs: Any, - ) -> SolverRequest: + ) -> RouteRequest: """Build normalized solver input for one route leg.""" - return SolverRequest( + return RouteRequest( family=family, tool=context.tool, in_ptype=context.port.ptype, @@ -1035,7 +1014,7 @@ class RoutingPlanner: strategy=self.resolve_strategy(strategy), ) - def solver_for_request(self, request: SolverRequest) -> Solver: + def solver_for_request(self, request: RouteRequest) -> Solver: """Construct the solver for a route request.""" return Solver(request) @@ -1059,9 +1038,6 @@ class RoutingPlanner: self, family: PrimitiveKind, context: RoutePortContext, - operation: RouteOperation | None = None, - diagnostic_request: Mapping[str, Any] | None = None, - /, *, length: float | None = None, jog: float | None = None, @@ -1073,32 +1049,7 @@ class RoutingPlanner: **kwargs: Any, ) -> RouteLeg: """Solve one route leg and attach it to its source Pather context.""" - if operation is None: - if family in ('straight', 'bend'): - operation = 'trace' - elif family == 's': - operation = 'jog' - else: - operation = 'uturn' - if diagnostic_request is None: - diagnostic_request = {} - - if length is not None and (not numpy.isfinite(length) or length < 0): - details = RouteFailureDetails( - operation=operation, - portspec=context.portspec, - in_ptype=context.port.ptype, - out_ptype=kwargs.get('out_ptype'), - request=diagnostic_request, - resolved_length=float(length), - resolved_jog=None if jog is None else float(jog), - minimum_length=None, - minimum_status=MinimumStatus.NOT_EVALUATED, - cause='Resolved route length must be finite and nonnegative', - ) - raise RouteError(details, policy=RouteFailurePolicy.FATAL) - - request = self.solver_request( + request = self.route_request( family=family, context=context, length=length, @@ -1111,48 +1062,10 @@ class RoutingPlanner: ) try: candidate = self.solver_for_request(request).solve() - except NoLegalRouteError as err: - cause = ( - 'No legal primitive offer for omitted-length U-turn' - if family == 'u' and length is None - else str(err) - ) - - minimum_length: float | None = None - minimum_cause: str | None = None - minimum_status: MinimumStatus - if length is None: - minimum_cause = cause - minimum_status = MinimumStatus.NO_ROUTE - else: - minimum_request = replace(request, length=None) - try: - minimum_candidate = self.solver_for_request(minimum_request).solve() - minimum_length = minimum_candidate.public_length - minimum_status = MinimumStatus.FOUND - except NoLegalRouteError as minimum_err: - minimum_cause = str(minimum_err) - minimum_status = MinimumStatus.NO_ROUTE - except Exception as minimum_err: - if route_failure_policy(minimum_err) is RouteFailurePolicy.FATAL: - raise - minimum_cause = str(minimum_err) - minimum_status = MinimumStatus.FAILED - - details = RouteFailureDetails( - operation=operation, - portspec=context.portspec, - in_ptype=context.port.ptype, - out_ptype=request.out_ptype, - request=diagnostic_request, - resolved_length=None if length is None else float(length), - resolved_jog=None if jog is None else float(jog), - minimum_length=minimum_length, - minimum_status=minimum_status, - cause=cause, - minimum_cause=minimum_cause, - ) - raise RouteError(details) from err + except BuildError as err: + if family == 'u' and length is None and not getattr(err, 'fatal', False): + raise BuildError('No legal primitive offer for omitted-length U-turn') from err + raise return self.route_leg_from_candidate(context, candidate, plug_into=plug_into) def prepared_route_action_from_leg( @@ -1216,87 +1129,25 @@ class RoutingPlanner: ) -> PreparedRouteResult: """Plan straight or single-bend traces, including `each` and bundle-bound modes.""" route_bounds = dict(bounds) - request_details = {'ccw': ccw, **{ - key: value for key, value in route_bounds.items() if value is not None - }} - if length is not None: - request_details['length'] = length - if spacing is not None: - request_details['spacing'] = spacing - if strategy is not None: - request_details['strategy'] = strategy - operation: RouteOperation = 'trace' - diagnostic_request = request_details - return self._plan_trace_route( - contexts, - ccw, - length, - operation, - diagnostic_request, - spacing=spacing, - strategy=strategy, - **route_bounds, - ) - - def _plan_trace_route( - self, - contexts: Sequence[RoutePortContext], - ccw: SupportsBool | None, - length: float | None, - operation: RouteOperation, - diagnostic_request: Mapping[str, Any], - /, - *, - spacing: float | ArrayLike | None, - strategy: RouteTieBreakStrategy | str | None, - **route_bounds: Any, - ) -> PreparedRouteResult: portspec = tuple(context.portspec for context in contexts) 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], - operation, - diagnostic_request, - length=length, - ccw=ccw, - strategy=strategy, - **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, - operation, - diagnostic_request, - length=each, - ccw=ccw, - strategy=strategy, - **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], - operation, - diagnostic_request, - length=None, - ccw=ccw, - strategy=strategy, - **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] @@ -1313,16 +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, - operation, - diagnostic_request, - length=route_length, - ccw=ccw, - strategy=strategy, - **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)) @@ -1337,37 +1179,6 @@ class RoutingPlanner: ) -> PreparedRouteResult: """Plan `trace_to()` by resolving positional targets or delegating to `trace()` modes.""" route_bounds = dict(bounds) - request_details = {'ccw': ccw, **{ - key: value for key, value in route_bounds.items() if value is not None - }} - if spacing is not None: - request_details['spacing'] = spacing - if strategy is not None: - request_details['strategy'] = strategy - operation: RouteOperation = 'trace_to' - diagnostic_request = request_details - return self._plan_trace_to_route( - contexts, - ccw, - operation, - diagnostic_request, - spacing=spacing, - strategy=strategy, - **route_bounds, - ) - - def _plan_trace_to_route( - self, - contexts: Sequence[RoutePortContext], - ccw: SupportsBool | None, - operation: RouteOperation, - diagnostic_request: Mapping[str, Any], - /, - *, - spacing: float | ArrayLike | None, - strategy: RouteTieBreakStrategy | str | None, - **route_bounds: Any, - ) -> PreparedRouteResult: if len(contexts) == 1: resolved = planner_bounds.resolved_position_bound(contexts[0].port, route_bounds, allow_length=False) else: @@ -1375,16 +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, - route_bounds.pop('length', None), - operation, - diagnostic_request, - spacing=spacing, - strategy=strategy, - **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 @@ -1394,16 +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], - operation, - diagnostic_request, - length=length, - ccw=ccw, - strategy=strategy, - **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( @@ -1417,28 +1210,8 @@ class RoutingPlanner: **bounds: Any, ) -> PreparedRouteResult: """Plan S-bend routes for single ports or spaced bundles.""" - request_details = {'offset': offset, **{ - key: value for key, value in bounds.items() if value is not None - }} - if length is not None: - request_details['length'] = length - if spacing is not None: - request_details['spacing'] = spacing - if strategy is not None: - request_details['strategy'] = strategy - operation: RouteOperation = 'jog' - diagnostic_request = request_details if numpy.isclose(offset, 0): - return self._plan_trace_to_route( - contexts, - None, - operation, - diagnostic_request, - length=length, - spacing=spacing, - strategy=strategy, - **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) @@ -1451,28 +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, - operation, - diagnostic_request, - strategy=strategy, - **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], - operation, - diagnostic_request, - length=length, - jog=offset, - strategy=strategy, - **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( @@ -1487,44 +1241,14 @@ class RoutingPlanner: ) -> PreparedRouteResult: """Plan U-turn routes for single ports or spaced bundles.""" route_bounds = dict(bounds) - request_details = {'offset': offset, **{ - key: value for key, value in route_bounds.items() if value is not None - }} - if length is not None: - request_details['length'] = length - if spacing is not None: - request_details['spacing'] = spacing - if strategy is not None: - request_details['strategy'] = strategy - operation: RouteOperation = 'uturn' - diagnostic_request = request_details portspec = tuple(context.portspec for context in contexts) planner_bounds.validate_uturn_args(portspec, spacing=spacing, bounds=route_bounds) 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, - operation, - diagnostic_request, - strategy=strategy, - **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], - operation, - diagnostic_request, - length=length, - jog=offset, - strategy=strategy, - **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( @@ -1534,9 +1258,6 @@ class RoutingPlanner: offset: float, length: float | None, spacing: float | ArrayLike | None, - operation: RouteOperation, - diagnostic_request: Mapping[str, Any], - /, *, strategy: RouteTieBreakStrategy | str | None = None, **kwargs: Any, @@ -1549,32 +1270,14 @@ class RoutingPlanner: and offset so all legs can be planned independently. """ if len(contexts) == 1: - return (self.plan_leg( - kind, - contexts[0], - operation, - diagnostic_request, - length=length, - jog=offset, - strategy=strategy, - **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], - operation, - diagnostic_request, - length=length, - jog=offset, - strategy=strategy, - **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] @@ -1584,10 +1287,7 @@ class RoutingPlanner: routes_by_name[spec_portspec] = self.plan_leg( 'straight', contexts_by_name[spec_portspec], - operation, - diagnostic_request, length=spec_length, - jog=spec_offset, strategy=strategy, **kwargs, ) @@ -1595,8 +1295,6 @@ class RoutingPlanner: routes_by_name[spec_portspec] = self.plan_leg( kind, contexts_by_name[spec_portspec], - operation, - diagnostic_request, length=spec_length, jog=spec_offset, strategy=strategy, @@ -1633,7 +1331,7 @@ class RoutingPlanner: desired.rotation = port_dst.rotation - pi desired.ptype = out_ptype family, length, jog, ccw = self.trace_into_spec(context_src.port, desired) - request = self.solver_request( + request = self.route_request( family, context_src, length=length, @@ -1652,7 +1350,7 @@ class RoutingPlanner: candidate = solver.solve(min_bends=min_bends, max_bends=max_bends) break except (BuildError, NotImplementedError) as err: - if route_failure_policy(err) is RouteFailurePolicy.FATAL: + if route_error_is_fatal(err): raise last_error = err if candidate is None: diff --git a/masque/builder/tools.py b/masque/builder/tools.py index 9db7645..40097a0 100644 --- a/masque/builder/tools.py +++ b/masque/builder/tools.py @@ -15,23 +15,12 @@ jog is left of travel; returned endpoint ports describe the primitive output in that same local frame. The planner transforms selected endpoints into layout coordinates only after a complete route has been chosen. -Primitive parameters are also the basis for route-failure diagnostics. Tools -must keep endpoint topology stable across each offer domain: length-like -Straight/Bend offers use a finite, attained, nonnegative minimum and advance -their local x coordinate by the selected length; S/U offers move their local y -coordinate by the selected jog. Endpoint rotation and output ptype must not -vary with the parameter and must agree with the concrete offer kind and its -declared `out_ptype`. These are Tool contract requirements rather than -exhaustively runtime-checked properties. - Tool authors should treat offer planning callbacks as pure descriptions. The -solver may call `endpoint_at()` and `cost_at()` many times while enumerating -candidate compositions, ptype adapters, and parameter solutions. Those -callbacks should be deterministic and should not mutate a Library, Pattern, or -live Pather. `bbox_at()` follows the same purity contract, but footprint data is -currently reserved for future footprint-aware planning and is not consumed by -the solver. `commit()` is the first selected-offer hook: it runs only for -primitives in the chosen route and returns the opaque value stored in +solver may call `endpoint_at()`, `cost_at()`, and `bbox_at()` many times while +enumerating candidate compositions, ptype adapters, and parameter solutions. +Those callbacks should be deterministic and should not mutate a Library, +Pattern, or live Pather. `commit()` is the first selected-offer hook: it runs +only for primitives in the chosen route and returns the opaque value stored in `RenderStep.data`. `render()` is the geometry-construction hook. Pather calls it later with a @@ -177,20 +166,12 @@ class PrimitiveOffer(ABC): Parameter domains are half-open `[min, max)` ranges, except `(value, value)` is a closed singleton for fixed-size primitives. `None` and `"unk"` ptypes are wildcards; incompatible concrete ptypes are rejected by `Pather`. - - Custom offers must have stable endpoint ptype and rotation throughout their - domain. Straight/Bend length domains must have a finite, attained, - nonnegative minimum and produce local `x == parameter`; S/U offers produce - local `y == parameter`. The planner relies on these invariants when it - diagnoses whether a failed constrained route has a preferred minimum-length - alternative or is unsupported at every legal length. """ in_ptype: str | None out_ptype: str | None priority_bias: float = 0.0 bbox_planner: BBoxCallable | None = None parameterized_bbox: Any | None = None - """Reserved footprint metadata; the current solver does not inspect it.""" endpoint_planner: EndpointCallable | None = None commit_planner: CommitCallable | None = None @@ -245,11 +226,10 @@ class PrimitiveOffer(ABC): def bbox_at(self, parameter: float) -> NDArray[numpy.float64]: """ - Return local primitive bounds for future footprint-aware planning. + Return local primitive bounds for footprint-aware planning. Tools may omit this hook by raising `NotImplementedError`; when present it must return a finite `(2, 2)` min/max array in local coordinates. - The current route solver does not call this method. """ if self.bbox_planner is None: raise NotImplementedError @@ -270,11 +250,9 @@ class PrimitiveOffer(ABC): """ Produce opaque render data for a selected primitive. - Routing preparation calls this only for selected primitives while - building `RenderStep.data`. Unselected candidates are evaluated by - endpoint/cost only and should not need commit-side work. This is a - materialization hook, not the live-layout mutation boundary: it must not - mutate the caller's Pather, Pattern, or Library. + `Pather` calls this only for selected primitives while preparing + `RenderStep.data`. Unselected candidates are evaluated by endpoint/cost + only and should not need commit-side work. """ selected = self.canonicalize_parameter(parameter) if self.commit_planner is not None: diff --git a/masque/builder/utils.py b/masque/builder/utils.py index ceea550..ca36fff 100644 --- a/masque/builder/utils.py +++ b/masque/builder/utils.py @@ -1,5 +1,5 @@ -from typing import TYPE_CHECKING -from collections.abc import Mapping +from typing import SupportsFloat, cast, TYPE_CHECKING +from collections.abc import Mapping, Sequence from pprint import pformat import numpy @@ -13,17 +13,6 @@ if TYPE_CHECKING: from ..ports import Port -_EXTENSION_BOUND_TYPES = ( - 'emin', 'min_extension', - 'emax', 'max_extension', - 'min_past_furthest', - ) -_POSITION_MIN_BOUND_TYPES = ('pmin', 'min_position', 'xmin', 'ymin') -_POSITION_MAX_BOUND_TYPES = ('pmax', 'max_position', 'xmax', 'ymax') -_POSITION_BOUND_TYPES = _POSITION_MIN_BOUND_TYPES + _POSITION_MAX_BOUND_TYPES -_BOUND_TYPES = _EXTENSION_BOUND_TYPES + _POSITION_BOUND_TYPES - - def ell( ports: Mapping[str, 'Port'], ccw: SupportsBool | None, @@ -93,18 +82,6 @@ def ell( """ if not ports: raise BuildError('Empty port list passed to `ell()`') - if bound_type not in _BOUND_TYPES: - raise BuildError(f'Invalid bound type {bound_type!r}; expected one of {_BOUND_TYPES}') - - try: - bound_arr = numpy.asarray(bound, dtype=float) - except (TypeError, ValueError) as err: - raise BuildError('bound must be a numeric scalar or length-2 vector') from err - if bound_arr.size not in (1, 2): - raise BuildError(f'bound must be scalar or have length 2; got {bound_arr.size} values') - if not numpy.all(numpy.isfinite(bound_arr)): - raise BuildError('bound must contain only finite values') - bound_values = bound_arr.reshape(-1) if ccw is None: if spacing is not None and not numpy.allclose(spacing, 0): @@ -196,34 +173,38 @@ def ell( travel = d_to_align - (ch_offsets.max() - ch_offsets) offsets = travel - travel.min().clip(max=0) - if bound_type in _EXTENSION_BOUND_TYPES: - if numpy.any(bound_values < 0): - raise BuildError(f'Got negative bound for extension: {bound_values}') - - if bound_values.size == 2: - horizontal_weight = abs(float(numpy.cos(direction))) - vertical_weight = abs(float(numpy.sin(direction))) - use_x = horizontal_weight > vertical_weight or numpy.isclose(horizontal_weight, vertical_weight) - rot_bound = float(bound_values[0 if use_x else 1]) + rot_bound: SupportsFloat + if bound_type in ('emin', 'min_extension', + 'emax', 'max_extension', + 'min_past_furthest',): + if numpy.size(bound) == 2: + bound = cast('Sequence[float]', bound) + rot_bound = (rot_matrix @ ((bound[0], 0), + (0, bound[1])))[0, :] else: - rot_bound = float(bound_values[0]) + bound = cast('float', bound) + rot_bound = numpy.array(bound) + + if rot_bound < 0: + raise BuildError(f'Got negative bound for extension: {rot_bound}') if bound_type in ('emin', 'min_extension', 'min_past_furthest'): - offsets += rot_bound + offsets += rot_bound.max() elif bound_type in ('emax', 'max_extension'): - offsets += rot_bound - offsets.max() + offsets += rot_bound.min() - offsets.max() else: - if bound_values.size == 2: - rot_bound = float((rot_matrix @ bound_values)[0]) + if numpy.size(bound) == 2: + bound = cast('Sequence[float]', bound) + rot_bound = (rot_matrix @ bound)[0] else: + bound = cast('float', bound) neg = (direction + pi / 4) % (2 * pi) > pi - bound_scalar = float(bound_values[0]) - rot_bound = -bound_scalar if neg else bound_scalar + rot_bound = -bound if neg else bound min_possible = x_start + offsets - if bound_type in _POSITION_MAX_BOUND_TYPES: + if bound_type in ('pmax', 'max_position', 'xmax', 'ymax'): extension = rot_bound - min_possible.max() - else: + elif bound_type in ('pmin', 'min_position', 'xmin', 'ymin'): extension = rot_bound - min_possible.min() offsets += extension diff --git a/masque/test/test_builder.py b/masque/test/test_builder.py index 4bf5c53..53c0a8c 100644 --- a/masque/test/test_builder.py +++ b/masque/test/test_builder.py @@ -3,7 +3,7 @@ import pytest from numpy.testing import assert_equal, assert_allclose from numpy import pi -from ..builder import MinimumStatus, Pather, RouteFailureDetails +from ..builder import Pather from ..builder.utils import ell from ..error import BuildError from ..library import Library @@ -14,50 +14,11 @@ from ..ports import Port def test_builder_public_imports() -> None: from masque import PortPather as TopPortPather from masque import RenderStep as TopRenderStep - from masque import RouteError as TopRouteError - from masque import RouteFailureDetails as TopRouteFailureDetails - from masque import RouteFailurePolicy as TopRouteFailurePolicy - from masque import MinimumStatus as TopMinimumStatus from masque.builder import PortPather as BuilderPortPather from masque.builder import RenderStep as BuilderRenderStep - from masque.builder import RouteError as BuilderRouteError - from masque.builder import RouteFailureDetails as BuilderRouteFailureDetails - from masque.builder import RouteFailurePolicy as BuilderRouteFailurePolicy - from masque.builder import MinimumStatus as BuilderMinimumStatus assert TopPortPather is BuilderPortPather assert TopRenderStep is BuilderRenderStep - assert TopRouteError is BuilderRouteError - assert TopRouteFailureDetails is BuilderRouteFailureDetails - assert TopRouteFailurePolicy is BuilderRouteFailurePolicy - assert TopMinimumStatus is BuilderMinimumStatus - - -def test_route_failure_details_enforces_minimum_status_invariants() -> None: - common = { - 'operation': 'trace', - 'portspec': 'A', - 'in_ptype': 'wire', - 'out_ptype': 'wide', - 'request': {}, - 'resolved_length': 1, - 'resolved_jog': None, - 'cause': 'no route', - } - - with pytest.raises(BuildError, match='FOUND requires minimum_length'): - RouteFailureDetails( - **common, - minimum_length=None, - minimum_status=MinimumStatus.FOUND, - ) - - with pytest.raises(BuildError, match='requires minimum_length=None'): - RouteFailureDetails( - **common, - minimum_length=2, - minimum_status=MinimumStatus.NO_ROUTE, - ) def test_builder_init() -> None: @@ -210,41 +171,3 @@ def test_ell_handles_array_spacing_when_ccw_none() -> None: with pytest.raises(BuildError, match='Spacing must be 0 or None'): ell(ports, None, 'min_extension', 5, spacing=numpy.array([1, 0])) - - -@pytest.mark.parametrize('bound_type', ['emin', 'emax', 'min_past_furthest']) -@pytest.mark.parametrize( - ('rotation', 'expected'), - [ - (0, 5), - (pi / 2, 7), - (pi / 6, 5), - (pi / 3, 7), - (pi / 4, 5), - ], - ) -def test_ell_extension_vector_selects_dominant_route_axis( - bound_type: str, - rotation: float, - expected: float, - ) -> None: - result = ell({'A': Port((0, 0), rotation)}, None, bound_type, (5, 7), spacing=0) - - assert_allclose(result['A'], expected) - - -@pytest.mark.parametrize('bound', [(-1, 2), (1, -2), (numpy.nan, 2), (1, numpy.inf), (1, 2, 3)]) -def test_ell_rejects_invalid_extension_vector(bound: tuple[float, ...]) -> None: - with pytest.raises(BuildError, match='bound|negative'): - ell({'A': Port((0, 0), 0)}, None, 'emin', bound, spacing=0) - - -def test_ell_position_vector_still_projects_onto_route_direction() -> None: - result = ell({'A': Port((0, 0), pi)}, None, 'pmax', (5, 7), spacing=0) - - assert_allclose(result['A'], 5) - - -def test_ell_rejects_invalid_bound_type() -> None: - with pytest.raises(BuildError, match='Invalid bound type'): - ell({'A': Port((0, 0), 0)}, None, 'nearest', 5, spacing=0) diff --git a/masque/test/test_pather_constraints.py b/masque/test/test_pather_constraints.py index e1c0744..fab1c68 100644 --- a/masque/test/test_pather_constraints.py +++ b/masque/test/test_pather_constraints.py @@ -5,10 +5,9 @@ import pytest import numpy from numpy import pi -from masque import MinimumStatus, Pather, Library, Port, RouteError, RouteFailurePolicy +from masque import Pather, Library, Port from masque.builder.planner import RoutePortContext, RoutingPlanner -from masque.builder.planner.planner import NoLegalRouteError -from masque.builder.tools import BendOffer, PathTool, RenderStep, StraightOffer, Tool, UOffer +from masque.builder.tools import BendOffer, PathTool, RenderStep, StraightOffer, Tool from masque.error import BuildError from masque.library import ILibrary @@ -115,347 +114,15 @@ class CountingPathTool(PathTool): return super().render(batch, port_names=port_names, **kwargs) -class RequestCountingTool(PlanningOnlyTool): - def __init__(self) -> None: - self.offer_calls = 0 - - def primitive_offers( - self, - kind: Literal['straight', 'bend', 's', 'u'], - *, - in_ptype: str | None = None, - out_ptype: str | None = None, - **kwargs: Any, - ) -> tuple[Any, ...]: - self.offer_calls += 1 - return super().primitive_offers( - kind, - in_ptype=in_ptype, - out_ptype=out_ptype, - **kwargs, - ) - - -class PreferredMinimumTool(PlanningOnlyTool): - def __init__(self) -> None: - self.commit_calls = 0 - self.diagnostic_context_kwargs: list[Any] = [] - - def _commit(self, parameter: float) -> dict[str, float]: - self.commit_calls += 1 - return {'parameter': parameter} - - def primitive_offers( - self, - kind: Literal['straight', 'bend', 's', 'u'], - *, - in_ptype: str | None = None, - out_ptype: str | None = None, - **kwargs: Any, - ) -> tuple[Any, ...]: - _ = out_ptype - self.diagnostic_context_kwargs.append(kwargs.get('diagnostic_context')) - if kind == 'bend': - ccw = bool(kwargs['ccw']) - rotation = -pi / 2 if ccw else pi / 2 - jog = 1 if ccw else -1 - return ( - BendOffer( - in_ptype=in_ptype, - out_ptype='wide', - priority_bias=100, - ccw=ccw, - length_domain=(2, 2), - endpoint_planner=lambda _length: Port((2, jog), rotation, ptype='wide'), - commit_planner=self._commit, - ), - BendOffer( - in_ptype=in_ptype, - out_ptype='wide', - ccw=ccw, - length_domain=(5, 5), - endpoint_planner=lambda _length: Port((5, jog), rotation, ptype='wide'), - commit_planner=self._commit, - ), - ) - if kind == 'u': - return (UOffer( - in_ptype=in_ptype, - out_ptype='wide', - jog_domain=(4, 4), - endpoint_planner=lambda _jog: Port((5, 4), 0, ptype='wide'), - commit_planner=self._commit, - ),) - return () - - -def test_route_error_reports_preferred_minimum_and_request_details() -> None: - tool = PreferredMinimumTool() - p = Pather( - Library(), - ports={'A': Port((0, 0), rotation=0, ptype='wire')}, - tools=tool, - render='deferred', - ) - - with pytest.raises(RouteError) as exc_info: - p.ccw('A', 1, out_ptype='wide', diagnostic_context='tool-value') - - details = exc_info.value.details - assert isinstance(exc_info.value, BuildError) - assert details.operation == 'trace_to' - assert details.portspec == 'A' - assert details.in_ptype == 'wire' - assert details.out_ptype == 'wide' - assert details.request == { - 'ccw': True, - 'length': 1, - 'out_ptype': 'wide', - 'diagnostic_context': 'tool-value', - } - assert details.resolved_length == 1 - assert details.resolved_jog is None - # The length-2 route exists, but normal cost ranking prefers length 5. - assert details.minimum_length == 5 - assert details.minimum_status is MinimumStatus.FOUND - assert exc_info.value.policy is RouteFailurePolicy.RECOVERABLE - assert details.minimum_cause is None - assert 'preferred_minimum_length: 5' in str(exc_info.value) - assert tool.commit_calls == 0 - assert tool.diagnostic_context_kwargs - assert set(tool.diagnostic_context_kwargs) == {'tool-value'} - assert not p._paths - with pytest.raises(TypeError): - details.request['new'] = 'value' # type: ignore[index] - - -def test_route_error_reports_uturn_preferred_minimum() -> None: - tool = PreferredMinimumTool() - p = Pather( - Library(), - ports={'A': Port((0, 0), rotation=0, ptype='wire')}, - tools=tool, - render='deferred', - ) - - with pytest.raises(RouteError) as exc_info: - p.uturn('A', 4, length=1, out_ptype='wide') - - details = exc_info.value.details - assert details.operation == 'uturn' - assert details.resolved_length == 1 - assert details.resolved_jog == 4 - assert details.minimum_length == 5 - assert tool.commit_calls == 0 - - -def test_route_error_reports_no_route_at_any_length() -> None: - p = Pather( - Library(), - ports={'A': Port((0, 0), rotation=0, ptype='wire')}, - tools=PathTool(layer='M1', width=2, ptype='wire'), - render='deferred', - ) - - with pytest.raises(RouteError) as exc_info: - p.ccw('A', 10, out_ptype='optical') - - details = exc_info.value.details - assert details.minimum_status is MinimumStatus.NO_ROUTE - assert details.minimum_length is None - assert details.minimum_cause is not None - assert 'no legal route exists at any length' in str(exc_info.value) - - -def test_route_error_reports_failed_minimum_diagnosis(monkeypatch: pytest.MonkeyPatch) -> None: - planner = RoutingPlanner() - context = RoutePortContext( - 'A', - Port((0, 0), rotation=0, ptype='wire'), - PlanningOnlyTool(), - ) - solve_count = 0 - - def solver_for_request(_request: Any) -> Any: - nonlocal solve_count - solve_count += 1 - current_solve = solve_count - - class FailingSolver: - def solve(self) -> Never: - if current_solve == 1: - raise NoLegalRouteError('requested length has no route') - raise RuntimeError('minimum diagnosis failed') - - return FailingSolver() - - monkeypatch.setattr(planner, 'solver_for_request', solver_for_request) - - with pytest.raises(RouteError) as exc_info: - planner.plan_leg( - 'bend', - context, - 'trace_to', - {'ccw': True, 'length': 1}, - length=1, - ccw=True, - ) - - details = exc_info.value.details - assert details.minimum_status is MinimumStatus.FAILED - assert details.minimum_length is None - assert details.minimum_cause == 'minimum diagnosis failed' - assert 'minimum-length calculation failed' in str(exc_info.value) - - -@pytest.mark.parametrize('length', [-1, numpy.nan, numpy.inf]) -def test_invalid_route_length_fails_before_offer_query_or_dead_fallback(length: float) -> None: - tool = RequestCountingTool() - p = Pather( - Library(), - ports={'A': Port((0, 0), rotation=0, ptype='wire')}, - tools=tool, - render='deferred', - ).set_dead() - - with pytest.raises(RouteError) as exc_info: - p.ccw('A', length) - - details = exc_info.value.details - assert details.minimum_status is MinimumStatus.NOT_EVALUATED - assert exc_info.value.policy is RouteFailurePolicy.FATAL - assert details.minimum_length is None - assert tool.offer_calls == 0 - assert numpy.allclose(p.ports['A'].offset, (0, 0)) - - -def test_negative_position_length_reports_bound_without_offer_query() -> None: - tool = RequestCountingTool() - p = Pather( - Library(), - ports={'A': Port((0, 0), rotation=0, ptype='wire')}, - tools=tool, - render='deferred', - ) - - with pytest.raises(RouteError) as exc_info: - p.ccw('A', x=1) - - details = exc_info.value.details - assert details.request == {'ccw': True, 'x': 1} - assert details.resolved_length == -1 - assert details.minimum_status is MinimumStatus.NOT_EVALUATED - assert tool.offer_calls == 0 - - -def test_negative_bundle_length_reports_failing_port_and_bound() -> None: - tool = RequestCountingTool() - p = Pather( - Library(), - ports={ - 'A': Port((0, 0), rotation=0, ptype='wire'), - 'B': Port((0, 4), rotation=0, ptype='wire'), - }, - tools=tool, - render='deferred', - ) - - with pytest.raises(RouteError) as exc_info: - p.trace(['A', 'B'], True, emax=0, spacing=2) - - details = exc_info.value.details - assert details.portspec == 'A' - assert details.request == {'ccw': True, 'emax': 0, 'spacing': 2} - assert details.resolved_length == -2 - assert details.minimum_status is MinimumStatus.NOT_EVALUATED - assert tool.offer_calls == 0 - - -def test_negative_each_length_reports_failing_port_without_offer_query() -> None: - tool = RequestCountingTool() - p = Pather( - Library(), - ports={ - 'A': Port((0, 0), rotation=0, ptype='wire'), - 'B': Port((0, 4), rotation=0, ptype='wire'), - }, - tools=tool, - render='deferred', - ) - - with pytest.raises(RouteError) as exc_info: - p.trace(['A', 'B'], None, each=-2) - - details = exc_info.value.details - assert details.portspec == 'A' - assert details.request == {'ccw': None, 'each': -2} - assert details.resolved_length == -2 - assert details.minimum_status is MinimumStatus.NOT_EVALUATED - assert tool.offer_calls == 0 - - -@pytest.mark.parametrize( - 'operation', - [ - lambda p: p.trace([], None, length=1), - lambda p: p.trace_to([], None, length=1), - lambda p: p.jog([], 2, length=3), - lambda p: p.uturn([], 2, length=3), - ], - ids=['trace', 'trace_to', 'jog', 'uturn'], - ) -def test_pather_rejects_empty_route_selection_before_planning(operation: Any) -> None: - tool = RequestCountingTool() - p = Pather(Library(), tools=tool, render='deferred') - - with pytest.raises(BuildError, match='at least one port'): - operation(p) - - assert tool.offer_calls == 0 - assert not p.ports - assert not p._paths - - -@pytest.mark.parametrize( - 'operation', - [ - lambda p: p.trace(['A', 'A'], None, each=1), - lambda p: p.trace_to(['A', 'A'], None, xmin=-10), - lambda p: p.jog(['A', 'A'], 2, length=3, spacing=1), - lambda p: p.uturn(['A', 'A'], 2, length=3, spacing=1), - lambda p: p.at(['A', 'A']).straight(1), - ], - ids=['trace', 'trace_to', 'jog', 'uturn', 'port_pather'], - ) -def test_pather_rejects_duplicate_route_selection_before_planning(operation: Any) -> None: - tool = RequestCountingTool() - p = Pather( - Library(), - ports={'A': Port((0, 0), rotation=0, ptype='wire')}, - tools=tool, - render='deferred', - ) - - with pytest.raises(BuildError, match=r"duplicates: \['A'\]"): - operation(p) - - assert tool.offer_calls == 0 - assert numpy.allclose(p.ports['A'].offset, (0, 0)) - assert not p._paths - - def test_pather_jog_failed_two_bend_route_is_atomic() -> None: lib = Library() tool = PathTool(layer='M1', width=2, ptype='wire') p = Pather(lib, tools=tool, render='immediate') p.pattern.ports['A'] = Port((0, 0), rotation=0, ptype='wire') - with pytest.raises(RouteError, match='S-bend') as exc_info: + with pytest.raises(BuildError, match='S-bend'): p.jog('A', 1.5, length=1.5) - assert exc_info.value.details.minimum_length == 2 - assert exc_info.value.details.resolved_jog == 1.5 - assert numpy.allclose(p.pattern.ports['A'].offset, (0, 0)) assert p.pattern.ports['A'].rotation == 0 assert len(p._paths['A']) == 0 @@ -926,12 +593,9 @@ def test_pather_uturn_failed_two_bend_route_is_atomic() -> None: p = Pather(lib, tools=tool) p.pattern.ports['A'] = Port((0, 0), rotation=0, ptype='wire') - with pytest.raises(RouteError, match='U-turn') as exc_info: + with pytest.raises(BuildError, match='U-turn'): p.uturn('A', 1.5, length=0) - assert exc_info.value.details.minimum_status is MinimumStatus.NO_ROUTE - assert exc_info.value.details.minimum_length is None - assert numpy.allclose(p.pattern.ports['A'].offset, (0, 0)) assert p.pattern.ports['A'].rotation == 0 assert len(p._paths['A']) == 0 diff --git a/masque/test/test_pather_core.py b/masque/test/test_pather_core.py index fca27c8..ddf869b 100644 --- a/masque/test/test_pather_core.py +++ b/masque/test/test_pather_core.py @@ -5,7 +5,7 @@ import numpy from numpy import pi from numpy.testing import assert_allclose, assert_equal -from masque import Pather, Library, Pattern, Port, RouteError +from masque import Pather, Library, Pattern, Port from masque.builder import PathTool, PrimitiveOffer, StraightOffer from masque.builder.planner import RoutingPlanner from masque.error import BuildError, PortError @@ -93,30 +93,6 @@ def test_pather_accepts_and_reuses_planner_instance() -> None: assert planner.trace_to_calls == 2 -def test_port_tool_policy_and_portpather_selection_follow_names() -> None: - default_tool = PathTool(layer=(1, 0), width=1, ptype='wire') - named_tool = PathTool(layer=(2, 0), width=1, ptype='wire') - p = Pather( - Library(), - ports={'A': Port((0, 0), rotation=0, ptype='wire')}, - tools={None: default_tool, 'A': named_tool}, - render='deferred', - ) - selected = p.at('A') - - p.rename_ports({'A': 'B'}) - - assert selected.ports == ['A'] - assert p.tools['A'] is named_tool - assert 'B' not in p.tools - p.straight('B', 1) - assert p._paths['B'][0].tool is default_tool - - p.mkport('A', Port((10, 0), rotation=0, ptype='wire')) - p.straight('A', 1) - assert p._paths['A'][0].tool is named_tool - - def test_pather_straight(pather_setup: tuple[Pather, PathTool, Library]) -> None: p, tool, lib = pather_setup p.straight("start", 10) @@ -160,13 +136,12 @@ def test_pather_dead_ports() -> None: p = Pather(lib, ports={"in": Port((0, 0), 0)}, tools=tool) p.set_dead() - with pytest.raises(RouteError, match='finite and nonnegative'): - p.straight("in", -10) + p.straight("in", -10) - assert_allclose(p.ports["in"].offset, [0, 0], atol=1e-10) + assert_allclose(p.ports["in"].offset, [10, 0], atol=1e-10) p.straight("in", 20) - assert_allclose(p.ports["in"].offset, [-20, 0], atol=1e-10) + assert_allclose(p.ports["in"].offset, [-10, 0], atol=1e-10) assert not p.pattern.has_shapes() @@ -419,9 +394,9 @@ def test_pather_dead_fallback_preserves_out_ptype() -> None: p.pattern.ports['A'] = Port((0, 0), rotation=0, ptype='wire') p.set_dead() - p.straight('A', 1000, out_ptype='other') + p.straight('A', -1000, out_ptype='other') - assert numpy.allclose(p.pattern.ports['A'].offset, (-1000, 0)) + assert numpy.allclose(p.pattern.ports['A'].offset, (1000, 0)) assert p.pattern.ports['A'].ptype == 'other' assert len(p._paths['A']) == 0 diff --git a/masque/test/test_pather_rendering.py b/masque/test/test_pather_rendering.py index 3c50e3b..186faf4 100644 --- a/masque/test/test_pather_rendering.py +++ b/masque/test/test_pather_rendering.py @@ -6,7 +6,7 @@ import numpy from numpy import pi from numpy.testing import assert_allclose -from ..builder import Pather, RouteError +from ..builder import Pather from ..builder.tools import PathTool, RenderStep, StraightOffer, Tool from ..error import BuildError from ..library import Library @@ -101,60 +101,6 @@ def test_deferred_render_retool(deferred_render_setup: tuple[Pather, PathTool, L assert len(rp.pattern.shapes[(1, 0)]) == 1 assert len(rp.pattern.shapes[(2, 0)]) == 1 - -def test_deferred_render_batches_tools_by_identity() -> None: - class CountingTool(PathTool): - def __init__(self, *args, **kwargs) -> None: # noqa: ANN002,ANN003 - super().__init__(*args, **kwargs) - self.render_calls = 0 - - def render(self, *args, **kwargs): # noqa: ANN002,ANN003,ANN202 - self.render_calls += 1 - return super().render(*args, **kwargs) - - lib = Library() - tool1 = CountingTool(layer=(1, 0), width=2, ptype='wire') - tool2 = CountingTool(layer=(1, 0), width=2, ptype='wire') - assert tool1 == tool2 - assert tool1 is not tool2 - p = Pather( - lib, - ports={'A': Port((0, 0), rotation=0, ptype='wire')}, - tools=tool1, - render='deferred', - ) - - p.straight('A', 5) - p.retool(tool2, 'A') - p.straight('A', 5) - p.render() - - assert tool1.render_calls == 1 - assert tool2.render_calls == 1 - assert len(p.pattern.shapes[(1, 0)]) == 2 - - -def test_deleted_name_reuse_does_not_retarget_pending_render_steps() -> None: - lib = Library() - tool = PathTool(layer=(1, 0), width=2, ptype='wire') - p = Pather( - lib, - ports={'A': Port((0, 0), rotation=0, ptype='wire')}, - tools=tool, - render='deferred', - ) - - p.straight('A', 5) - original_step = p._paths['A'][0] - p.rename_ports({'A': None}) - p.mkport('A', Port((100, 0), rotation=0, ptype='wire')) - p.straight('A', 5) - - assert len(p._paths['A']) == 2 - assert_allclose(original_step.start_port.offset, (0, 0)) - assert_allclose(original_step.end_port.offset, (-5, 0)) - assert_allclose(p._paths['A'][1].start_port.offset, (100, 0)) - def test_portpather_translate_only_affects_future_steps(deferred_render_setup: tuple[Pather, PathTool, Library]) -> None: rp, tool, lib = deferred_render_setup pp = rp.at("start") @@ -176,10 +122,9 @@ def test_deferred_render_dead_ports() -> None: rp = Pather(lib, ports={"in": Port((0, 0), 0)}, tools=tool, render='deferred') rp.set_dead() - with pytest.raises(RouteError, match='finite and nonnegative'): - rp.straight("in", -10) + rp.straight("in", -10) - assert_allclose(rp.ports["in"].offset, [0, 0], atol=1e-10) + assert_allclose(rp.ports["in"].offset, [10, 0], atol=1e-10) assert len(rp._paths["in"]) == 0 diff --git a/masque/test/test_pather_trace_into.py b/masque/test/test_pather_trace_into.py index f79ecc1..ed967dd 100644 --- a/masque/test/test_pather_trace_into.py +++ b/masque/test/test_pather_trace_into.py @@ -5,9 +5,9 @@ import pytest from numpy import pi from numpy.testing import assert_equal -from masque import Library, PathTool, Port, Pather, RouteFailurePolicy +from masque import Library, PathTool, Port, Pather from masque.builder.planner import PreparedRouteResult, RoutePlanningError, RoutePortContext, RoutingPlanner -from masque.builder.planner.planner import Candidate, SolverRequest +from masque.builder.planner.planner import Candidate, RouteRequest from masque.builder.tools import BendOffer, PrimitiveOffer, StraightOffer, Tool from masque.error import BuildError, PortError @@ -280,7 +280,7 @@ class TraceIntoBudgetSolver: band = (min_bends, max_bends) self.attempts.append(band) if band in self.fatal_at: - raise RoutePlanningError('fatal', policy=RouteFailurePolicy.FATAL) + raise RoutePlanningError('fatal', fatal=True) if band not in self.successes: raise BuildError('try next budget') return Candidate((), Port((0, 0), rotation=0, ptype='wire'), 0.0, 0, 0.0) @@ -291,7 +291,7 @@ class TraceIntoBudgetPlanner(RoutingPlanner): self.solver = TraceIntoBudgetSolver(successes, fatal_at=fatal_at) self.solver_requests = 0 - def solver_for_request(self, request: SolverRequest) -> Any: + def solver_for_request(self, request: RouteRequest) -> Any: _ = request self.solver_requests += 1 return self.solver