Compare commits

..

No commits in common. "49713896c3af7d06279b1bb1556dae88f6ad0024" and "fe4e2f760bbfd4f4e132aea6c6e048ec9c5da6d3" have entirely different histories.

14 changed files with 125 additions and 1145 deletions

View file

@ -82,10 +82,6 @@ from .abstract import Abstract as Abstract
from .builder import ( from .builder import (
Tool as Tool, Tool as Tool,
Pather as Pather, Pather as Pather,
RouteError as RouteError,
RouteFailureDetails as RouteFailureDetails,
RouteFailurePolicy as RouteFailurePolicy,
MinimumStatus as MinimumStatus,
RenderStep as RenderStep, RenderStep as RenderStep,
AutoTool as AutoTool, AutoTool as AutoTool,
PathTool as PathTool, PathTool as PathTool,

View file

@ -8,23 +8,16 @@ optional footprint metadata, and a commit hook for producing tool-specific
render data after a concrete parameter has been selected. render data after a concrete parameter has been selected.
`Pather` owns user-facing route operations such as `trace()`, `jog()`, `Pather` owns user-facing route operations such as `trace()`, `jog()`,
`uturn()`, and `trace_into()`. The internal planner resolves each operation into `uturn()`, and `trace_into()`. For each operation, it asks the active `Tool` for
one or more `SolverRequest`s. This normalization is why the public routing API offers and passes those offers plus route constraints to the internal router.
can remain a convenient keyword-based interface without making the solver The router selects a sequence of internal selected primitives, each pairing an
stringly typed internally. A pure solver search selects a `Candidate`, and a offer with a concrete parameter, endpoint, and cost.
`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.
Selection is pure with respect to caller-owned Pattern, Library, and Pather Route commit is separate from route selection. Once a route is selected,
state. Tool offer discovery and endpoint/cost/bbox callbacks must likewise be `Pather` calls `offer.commit(parameter)` only for the selected primitives and
deterministic and must not mutate that state. `commit()` is the first stores the returned opaque tool payload in `RenderStep.data`. Later,
selected-offer materialization hook, but it still must not mutate the live `Pather.render()` batches compatible `RenderStep`s and calls `Tool.render()` to
layout. `Tool.render()` is the geometry mutation boundary: later, turn those committed payloads into geometry.
`Pather.render()` batches compatible `RenderStep`s and inserts the resulting
geometry into the Pattern and Library.
`PrimitiveOffer` and `RenderStep.data` are the tool-facing contract. `PrimitiveOffer` and `RenderStep.data` are the tool-facing contract.
`RenderStep` is `Pather`'s deferred-render record, and `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 - Pather applies the prepared result to ports, deferred render queues, and the
target pattern/library. 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 Code outside the builder package should prefer the exports here over importing
from `masque.builder.planner`. The planner package is intentionally available from `masque.builder.planner`. The planner package is intentionally available
for tests and internal maintenance, but it is not the compatibility boundary for tests and internal maintenance, but it is not the compatibility boundary
@ -53,13 +41,6 @@ from .pather import (
Pather as Pather, Pather as Pather,
PortPather as PortPather, 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 .utils import ell as ell
from .tools import ( from .tools import (
Tool as Tool, Tool as Tool,

View file

@ -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))

View file

@ -22,25 +22,10 @@ Routing is split into four ownership phases:
This split keeps selection failures largely transactional for live Pather This split keeps selection failures largely transactional for live Pather
state: unsupported primitive combinations can fail before the Pattern, pending state: unsupported primitive combinations can fail before the Pattern, pending
step queue, or Library are touched. Once prepared actions are applied, plug, step queue, or Library are touched. Once selected offers are committed and
rename, render, or insertion failures may leave partial output; that mutation prepared actions are applied, later commit, plug, rename, render, or insertion
boundary belongs to Pather, not to Tool implementations or the route solver. 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.
""" """
from typing import Self, Any, Literal, overload from typing import Self, Any, Literal, overload
from collections.abc import Iterator, Iterable, Mapping, MutableMapping, Sequence from collections.abc import Iterator, Iterable, Mapping, MutableMapping, Sequence
@ -70,9 +55,8 @@ from .tools import (
from .planner.interface import ( from .planner.interface import (
PreparedRouteResult, PreparedRouteResult,
RoutePortContext, RoutePortContext,
route_failure_policy, route_error_is_fatal,
) )
from .error import RouteFailurePolicy
from .planner import ( from .planner import (
RouteTieBreakStrategy, RouteTieBreakStrategy,
RoutingPlanner, RoutingPlanner,
@ -132,9 +116,6 @@ class Pather(PortList):
""" """
Tool objects used to dynamically generate new routing segments. Tool objects used to dynamically generate new routing segments.
A key of `None` indicates the default `Tool`. 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 planner: RoutingPlanner
@ -178,16 +159,6 @@ class Pather(PortList):
def _route_contexts(self, portspecs: Sequence[str]) -> tuple[RoutePortContext, ...]: def _route_contexts(self, portspecs: Sequence[str]) -> tuple[RoutePortContext, ...]:
"""Snapshot several ports in request order for bundle planning.""" """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) return tuple(self._route_context(portspec) for portspec in portspecs)
@property @property
@ -574,7 +545,7 @@ class Pather(PortList):
try: try:
result = self.planner.plan_trace_route(contexts, ccw, length, spacing=spacing, strategy=strategy, **bounds) result = self.planner.plan_trace_route(contexts, ccw, length, spacing=spacing, strategy=strategy, **bounds)
except (BuildError, NotImplementedError) as err: 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 raise
if length is not None and len(contexts) == 1: if length is not None and len(contexts) == 1:
context = contexts[0] context = contexts[0]
@ -631,11 +602,7 @@ class Pather(PortList):
try: try:
result = self.planner.plan_trace_to_route(contexts, ccw, spacing=spacing, strategy=strategy, **bounds) result = self.planner.plan_trace_to_route(contexts, ccw, spacing=spacing, strategy=strategy, **bounds)
except (BuildError, NotImplementedError) as err: except (BuildError, NotImplementedError) as err:
if ( if not self._dead or len(contexts) != 1 or route_error_is_fatal(err):
not self._dead
or len(contexts) != 1
or route_failure_policy(err) is RouteFailurePolicy.FATAL
):
raise raise
if bounds.get('length') is not None: if bounds.get('length') is not None:
length = bounds['length'] length = bounds['length']
@ -707,11 +674,7 @@ class Pather(PortList):
try: try:
result = self.planner.plan_jog_route(contexts, offset, length, spacing=spacing, strategy=strategy, **bounds) result = self.planner.plan_jog_route(contexts, offset, length, spacing=spacing, strategy=strategy, **bounds)
except (BuildError, NotImplementedError) as err: except (BuildError, NotImplementedError) as err:
if ( if not self._dead or len(contexts) != 1 or route_error_is_fatal(err):
not self._dead
or len(contexts) != 1
or route_failure_policy(err) is RouteFailurePolicy.FATAL
):
raise raise
if numpy.isclose(offset, 0): if numpy.isclose(offset, 0):
if length is None: if length is None:
@ -777,7 +740,7 @@ class Pather(PortList):
not self._dead not self._dead
or len(contexts) != 1 or len(contexts) != 1
or length is None or length is None
or route_failure_policy(err) is RouteFailurePolicy.FATAL or route_error_is_fatal(err)
): ):
raise raise
context = contexts[0] context = contexts[0]
@ -857,11 +820,6 @@ class Pather(PortList):
Consecutive compatible `RenderStep`s are batched by port and Tool, then Consecutive compatible `RenderStep`s are batched by port and Tool, then
passed to `Tool.render()`. After insertion, the rendered output port is passed to `Tool.render()`. After insertion, the rendered output port is
checked against the endpoint that planning selected. 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): with self._logger.log_operation(self, 'render', None, append=append):
tool_port_names = ('A', 'B') tool_port_names = ('A', 'B')
@ -934,7 +892,7 @@ class Pather(PortList):
batch: list[RenderStep] = [] batch: list[RenderStep] = []
for step in steps: for step in steps:
appendable = step.opcode in ('L', 'S', 'U') 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)): if batch and (not appendable or not same_tool or not batch[-1].is_continuous_with(step)):
render_batch(portspec, batch, append) render_batch(portspec, batch, append)
batch = [] batch = []
@ -1014,13 +972,7 @@ class Pather(PortList):
class PortPather: class PortPather:
""" """ Port state manager for fluent pathing. """
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.
"""
def __init__( def __init__(
self, self,
ports: str | Iterable[str], ports: str | Iterable[str],

View file

@ -10,7 +10,7 @@ from .interface import (
PreparedRouteResult as PreparedRouteResult, PreparedRouteResult as PreparedRouteResult,
RoutePlanningError as RoutePlanningError, RoutePlanningError as RoutePlanningError,
RoutePortContext as RoutePortContext, 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 RouteTieBreakStrategy as RouteTieBreakStrategy
from .planner import RoutingPlanner as RoutingPlanner from .planner import RoutingPlanner as RoutingPlanner

View file

@ -15,30 +15,21 @@ from dataclasses import dataclass
from ...error import BuildError from ...error import BuildError
from ...ports import Port from ...ports import Port
from ..tools import RenderStep, Tool from ..tools import RenderStep, Tool
from ..error import RouteError, RouteFailurePolicy
class RoutePlanningError(BuildError): class RoutePlanningError(BuildError):
"""Route-planning error with fallback policy metadata.""" """Route-planning error with fallback policy metadata."""
policy: RouteFailurePolicy fatal: bool
def __init__( def __init__(self, *args: object, fatal: bool = False) -> None:
self,
*args: object,
policy: RouteFailurePolicy = RouteFailurePolicy.RECOVERABLE,
) -> None:
super().__init__(*args) super().__init__(*args)
self.policy = policy self.fatal = fatal
def route_failure_policy(err: Exception) -> RouteFailurePolicy: def route_error_is_fatal(err: Exception) -> bool:
"""Return typed route recovery policy, defaulting generic errors to recoverable.""" """Return true when a planning error should bypass dead-Pather fallback."""
if isinstance(err, RoutePlanningError): return bool(getattr(err, 'fatal', False))
return err.policy
if isinstance(err, RouteError):
return err.policy
return RouteFailurePolicy.RECOVERABLE
@dataclass(frozen=True, slots=True) @dataclass(frozen=True, slots=True)
@ -63,10 +54,9 @@ class PreparedRouteAction:
""" """
Prepared mutation for one routed Pather port. Prepared mutation for one routed Pather port.
Pure selection has already completed, and the planner has materialized the The planner has already committed selected primitive offers into
selected primitive offers into `render_steps` and computed the final live `render_steps` and computed the final live port. `plug_into`, when set,
port. `plug_into`, when set, names the destination port to consume after the names the destination port to consume after the route endpoint is applied.
route endpoint is applied.
""" """
portspec: str portspec: str
"""Live Pather port name to update.""" """Live Pather port name to update."""
@ -83,10 +73,9 @@ class PreparedRouteResult:
""" """
Complete prepared result for one Pather routing operation. Complete prepared result for one Pather routing operation.
`actions` contain materialized, committed render data and are applied first. `actions` are applied first. `renames` are deferred until after all route
`renames` are deferred until after all route actions so trace-into/thru actions so trace-into/thru behavior can be represented without exposing the
behavior can be represented without exposing the solver's selected solver's selected primitive sequence to Pather.
primitive sequence to Pather.
""" """
actions: tuple[PreparedRouteAction, ...] actions: tuple[PreparedRouteAction, ...]
"""Prepared per-port route mutations.""" """Prepared per-port route mutations."""

View file

@ -6,19 +6,11 @@ Tool primitive offers. `Pather` passes copied `RoutePortContext` snapshots here;
the planner returns `PreparedRouteResult` records that describe pending the planner returns `PreparedRouteResult` records that describe pending
mutations without applying them to the live Pattern. mutations without applying them to the live Pattern.
Public routing modes and bounds are normalized by `bounds.py` into per-leg Public routing modes and bounds are normalized by `bounds.py` before the solver
`SolverRequest` values. `Solver` performs the pure search: it queries Tool sees them. This module plans one route intent at a time: it queries Tool offers,
offers, enumerates bounded primitive compositions, inserts ptype adapters, enumerates bounded primitive compositions, inserts ptype adapters, solves
solves primitive parameters, and returns the ranked `Candidate`. A `RouteLeg` primitive parameters, ranks candidates, and commits only the selected offers
then attaches the candidate to its copied source port and Tool. Preparation is into `RenderStep` payloads.
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.
All search is performed in Tool-local route coordinates. The active input port 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 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 # ruff: noqa: ANN401,PLR0912,PLR0913,PLR0915,TC001,TC002,TC003
from collections.abc import Iterable, Mapping, Sequence from collections.abc import Iterable, Mapping, Sequence
from dataclasses import dataclass, replace from dataclasses import dataclass
from itertools import combinations from itertools import combinations
from math import cos, isclose as math_isclose, sin from math import cos, isclose as math_isclose, sin
from typing import Any, Literal from typing import Any, Literal
@ -51,13 +43,6 @@ from ..tools import (
StraightOffer, StraightOffer,
Tool, Tool,
) )
from ..error import (
MinimumStatus,
RouteError,
RouteFailureDetails,
RouteFailurePolicy,
RouteOperation,
)
from ..utils import ell from ..utils import ell
from . import bounds as planner_bounds from . import bounds as planner_bounds
from .interface import ( from .interface import (
@ -65,16 +50,12 @@ from .interface import (
PreparedRouteResult, PreparedRouteResult,
RoutePlanningError, RoutePlanningError,
RoutePortContext, RoutePortContext,
route_failure_policy, route_error_is_fatal,
) )
RouteTieBreakStrategy = Literal['straight_first', 'turn_first'] RouteTieBreakStrategy = Literal['straight_first', 'turn_first']
class NoLegalRouteError(BuildError):
"""Internal marker for exhaustive candidate-selection failure."""
def validate_strategy(strategy: RouteTieBreakStrategy | str) -> RouteTieBreakStrategy: def validate_strategy(strategy: RouteTieBreakStrategy | str) -> RouteTieBreakStrategy:
"""Return a supported route tie-break strategy or raise a routing error.""" """Return a supported route tie-break strategy or raise a routing error."""
if strategy in ('straight_first', 'turn_first'): 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: def raise_if_fatal(err: Exception) -> None:
"""Propagate fatal planning errors while allowing normal candidate rejection.""" """Propagate fatal planning errors while allowing normal candidate rejection."""
if route_failure_policy(err) is RouteFailurePolicy.FATAL: if getattr(err, 'fatal', False):
raise err raise err
@ -245,9 +226,9 @@ class Candidate:
@dataclass(frozen=True, slots=True) @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 Public Pather calls are converted into this smaller shape before grammar
enumeration. `length`, `jog`, and `out_ptype` become endpoint constraints; 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. 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.request = request
self.eval_cache: dict[tuple[int, float, str | None, str | None, str, str, PrimitiveKind | None], SelectedPrimitive] = {} self.eval_cache: dict[tuple[int, float, str | None, str | None, str, str, PrimitiveKind | None], SelectedPrimitive] = {}
self.offer_cache: dict[ self.offer_cache: dict[
@ -410,16 +391,14 @@ class Solver:
if not candidates: if not candidates:
for err in errors: for err in errors:
if route_failure_policy(err) is RouteFailurePolicy.FATAL: if getattr(err, 'fatal', False):
raise err raise err
if errors: if errors:
last_error = errors[-1] last_error = errors[-1]
if self.request.route_name in str(last_error): if self.request.route_name in str(last_error):
raise NoLegalRouteError(str(last_error)) from last_error raise last_error
raise NoLegalRouteError( raise BuildError(f'{self.request.route_name} route is unsupported: {last_error}') from last_error
f'{self.request.route_name} route is unsupported: {last_error}' raise BuildError(f'No legal primitive offer for {self.request.route_name}')
) from last_error
raise NoLegalRouteError(f'No legal primitive offer for {self.request.route_name}')
return min( return min(
candidates, candidates,
@ -490,12 +469,12 @@ class Solver:
if not ptypes_compatible(out_port.ptype, offer.out_ptype): if not ptypes_compatible(out_port.ptype, offer.out_ptype):
raise RoutePlanningError( raise RoutePlanningError(
f'{route_name} primitive endpoint ptype does not match declared offer out_ptype', 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): if out_ptype is not None and not ptypes_compatible(out_port.ptype, out_ptype):
raise RoutePlanningError( raise RoutePlanningError(
'Requested out_ptype does not match primitive endpoint ptype', 'Requested out_ptype does not match primitive endpoint ptype',
policy=RouteFailurePolicy.FATAL, fatal=True,
) )
cost = float(offer.cost_at(selected)) cost = float(offer.cost_at(selected))
if not numpy.isfinite(cost): if not numpy.isfinite(cost):
@ -1007,7 +986,7 @@ class RoutingPlanner:
bands.append((4, 4)) bands.append((4, 4))
return tuple(bands) return tuple(bands)
def solver_request( def route_request(
self, self,
family: PrimitiveKind, family: PrimitiveKind,
context: RoutePortContext, context: RoutePortContext,
@ -1019,9 +998,9 @@ class RoutingPlanner:
max_bends: int | None = None, max_bends: int | None = None,
strategy: RouteTieBreakStrategy | str | None = None, strategy: RouteTieBreakStrategy | str | None = None,
**kwargs: Any, **kwargs: Any,
) -> SolverRequest: ) -> RouteRequest:
"""Build normalized solver input for one route leg.""" """Build normalized solver input for one route leg."""
return SolverRequest( return RouteRequest(
family=family, family=family,
tool=context.tool, tool=context.tool,
in_ptype=context.port.ptype, in_ptype=context.port.ptype,
@ -1035,7 +1014,7 @@ class RoutingPlanner:
strategy=self.resolve_strategy(strategy), 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.""" """Construct the solver for a route request."""
return Solver(request) return Solver(request)
@ -1059,9 +1038,6 @@ class RoutingPlanner:
self, self,
family: PrimitiveKind, family: PrimitiveKind,
context: RoutePortContext, context: RoutePortContext,
operation: RouteOperation | None = None,
diagnostic_request: Mapping[str, Any] | None = None,
/,
*, *,
length: float | None = None, length: float | None = None,
jog: float | None = None, jog: float | None = None,
@ -1073,32 +1049,7 @@ class RoutingPlanner:
**kwargs: Any, **kwargs: Any,
) -> RouteLeg: ) -> RouteLeg:
"""Solve one route leg and attach it to its source Pather context.""" """Solve one route leg and attach it to its source Pather context."""
if operation is None: request = self.route_request(
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(
family=family, family=family,
context=context, context=context,
length=length, length=length,
@ -1111,48 +1062,10 @@ class RoutingPlanner:
) )
try: try:
candidate = self.solver_for_request(request).solve() candidate = self.solver_for_request(request).solve()
except NoLegalRouteError as err: except BuildError as err:
cause = ( if family == 'u' and length is None and not getattr(err, 'fatal', False):
'No legal primitive offer for omitted-length U-turn' raise BuildError('No legal primitive offer for omitted-length U-turn') from err
if family == 'u' and length is None raise
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
return self.route_leg_from_candidate(context, candidate, plug_into=plug_into) return self.route_leg_from_candidate(context, candidate, plug_into=plug_into)
def prepared_route_action_from_leg( def prepared_route_action_from_leg(
@ -1216,87 +1129,25 @@ class RoutingPlanner:
) -> PreparedRouteResult: ) -> PreparedRouteResult:
"""Plan straight or single-bend traces, including `each` and bundle-bound modes.""" """Plan straight or single-bend traces, including `each` and bundle-bound modes."""
route_bounds = dict(bounds) 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) portspec = tuple(context.portspec for context in contexts)
planner_bounds.validate_trace_args(portspec, length=length, spacing=spacing, bounds=route_bounds) planner_bounds.validate_trace_args(portspec, length=length, spacing=spacing, bounds=route_bounds)
family: Literal['straight', 'bend'] = 'straight' if ccw is None else 'bend' family: Literal['straight', 'bend'] = 'straight' if ccw is None else 'bend'
if length is not None: if length is not None:
leg = self.plan_leg( leg = self.plan_leg(family, contexts[0], length=length, ccw=ccw, strategy=strategy, **route_bounds)
family,
contexts[0],
operation,
diagnostic_request,
length=length,
ccw=ccw,
strategy=strategy,
**route_bounds,
)
return self.prepared_result_from_legs((leg,)) return self.prepared_result_from_legs((leg,))
if route_bounds.get('each') is not None: if route_bounds.get('each') is not None:
each = route_bounds.pop('each') each = route_bounds.pop('each')
return PreparedRouteResult(tuple( return PreparedRouteResult(tuple(
self.prepared_route_action_from_leg( self.prepared_route_action_from_leg(
self.plan_leg( self.plan_leg(family, context, length=each, ccw=ccw, strategy=strategy, **route_bounds),
family,
context,
operation,
diagnostic_request,
length=each,
ccw=ccw,
strategy=strategy,
**route_bounds,
),
) )
for context in contexts for context in contexts
)) ))
bundle_bounds = planner_bounds.present_bundle_bounds(route_bounds) bundle_bounds = planner_bounds.present_bundle_bounds(route_bounds)
if not bundle_bounds: if not bundle_bounds:
leg = self.plan_leg( leg = self.plan_leg(family, contexts[0], length=None, ccw=ccw, strategy=strategy, **route_bounds)
family,
contexts[0],
operation,
diagnostic_request,
length=None,
ccw=ccw,
strategy=strategy,
**route_bounds,
)
return self.prepared_result_from_legs((leg,)) return self.prepared_result_from_legs((leg,))
bound_type = bundle_bounds[0] bound_type = bundle_bounds[0]
@ -1313,16 +1164,7 @@ class RoutingPlanner:
actions = [] actions = []
for port_name, route_length in extensions.items(): for port_name, route_length in extensions.items():
context = next(context for context in contexts if context.portspec == port_name) context = next(context for context in contexts if context.portspec == port_name)
leg = self.plan_leg( leg = self.plan_leg(family, context, length=route_length, ccw=ccw, strategy=strategy, **route_bounds)
family,
context,
operation,
diagnostic_request,
length=route_length,
ccw=ccw,
strategy=strategy,
**route_bounds,
)
actions.append(self.prepared_route_action_from_leg(leg)) actions.append(self.prepared_route_action_from_leg(leg))
return PreparedRouteResult(tuple(actions)) return PreparedRouteResult(tuple(actions))
@ -1337,37 +1179,6 @@ class RoutingPlanner:
) -> PreparedRouteResult: ) -> PreparedRouteResult:
"""Plan `trace_to()` by resolving positional targets or delegating to `trace()` modes.""" """Plan `trace_to()` by resolving positional targets or delegating to `trace()` modes."""
route_bounds = dict(bounds) 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: if len(contexts) == 1:
resolved = planner_bounds.resolved_position_bound(contexts[0].port, route_bounds, allow_length=False) resolved = planner_bounds.resolved_position_bound(contexts[0].port, route_bounds, allow_length=False)
else: else:
@ -1375,16 +1186,7 @@ class RoutingPlanner:
if any(route_bounds.get(key) is not None for key in planner_bounds.POSITION_KEYS): 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') raise BuildError('Position bounds only allowed with a single port')
if resolved is None: if resolved is None:
return self._plan_trace_route( return self.plan_trace_route(contexts, ccw, spacing=spacing, strategy=strategy, **route_bounds)
contexts,
ccw,
route_bounds.pop('length', None),
operation,
diagnostic_request,
spacing=spacing,
strategy=strategy,
**route_bounds,
)
planner_bounds.validate_trace_to_positional_args(spacing=spacing, bounds=route_bounds) planner_bounds.validate_trace_to_positional_args(spacing=spacing, bounds=route_bounds)
_key, _value, length = resolved _key, _value, length = resolved
@ -1394,16 +1196,7 @@ class RoutingPlanner:
if key not in planner_bounds.POSITION_KEYS and key != 'length' if key not in planner_bounds.POSITION_KEYS and key != 'length'
} }
family: Literal['straight', 'bend'] = 'straight' if ccw is None else 'bend' family: Literal['straight', 'bend'] = 'straight' if ccw is None else 'bend'
leg = self.plan_leg( leg = self.plan_leg(family, contexts[0], length=length, ccw=ccw, strategy=strategy, **other_bounds)
family,
contexts[0],
operation,
diagnostic_request,
length=length,
ccw=ccw,
strategy=strategy,
**other_bounds,
)
return self.prepared_result_from_legs((leg,)) return self.prepared_result_from_legs((leg,))
def plan_jog_route( def plan_jog_route(
@ -1417,28 +1210,8 @@ class RoutingPlanner:
**bounds: Any, **bounds: Any,
) -> PreparedRouteResult: ) -> PreparedRouteResult:
"""Plan S-bend routes for single ports or spaced bundles.""" """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): if numpy.isclose(offset, 0):
return self._plan_trace_to_route( return self.plan_trace_to_route(contexts, None, length=length, spacing=spacing, strategy=strategy, **bounds)
contexts,
None,
operation,
diagnostic_request,
length=length,
spacing=spacing,
strategy=strategy,
**bounds,
)
route_bounds = dict(bounds) route_bounds = dict(bounds)
portspec = tuple(context.portspec for context in contexts) portspec = tuple(context.portspec for context in contexts)
planner_bounds.validate_jog_args(portspec, length=length, spacing=spacing, bounds=route_bounds) planner_bounds.validate_jog_args(portspec, length=length, spacing=spacing, bounds=route_bounds)
@ -1451,28 +1224,9 @@ class RoutingPlanner:
if len(contexts) > 1: if len(contexts) > 1:
return PreparedRouteResult(tuple( return PreparedRouteResult(tuple(
self.prepared_route_action_from_leg(leg) self.prepared_route_action_from_leg(leg)
for leg in self.plan_su_bundle_routes( for leg in self.plan_su_bundle_routes('s', contexts, offset, length, spacing, strategy=strategy, **other_bounds)
's',
contexts,
offset,
length,
spacing,
operation,
diagnostic_request,
strategy=strategy,
**other_bounds,
)
)) ))
leg = self.plan_leg( leg = self.plan_leg('s', contexts[0], length=length, jog=offset, strategy=strategy, **other_bounds)
's',
contexts[0],
operation,
diagnostic_request,
length=length,
jog=offset,
strategy=strategy,
**other_bounds,
)
return self.prepared_result_from_legs((leg,)) return self.prepared_result_from_legs((leg,))
def plan_uturn_route( def plan_uturn_route(
@ -1487,44 +1241,14 @@ class RoutingPlanner:
) -> PreparedRouteResult: ) -> PreparedRouteResult:
"""Plan U-turn routes for single ports or spaced bundles.""" """Plan U-turn routes for single ports or spaced bundles."""
route_bounds = dict(bounds) 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) portspec = tuple(context.portspec for context in contexts)
planner_bounds.validate_uturn_args(portspec, spacing=spacing, bounds=route_bounds) planner_bounds.validate_uturn_args(portspec, spacing=spacing, bounds=route_bounds)
if len(contexts) > 1: if len(contexts) > 1:
return PreparedRouteResult(tuple( return PreparedRouteResult(tuple(
self.prepared_route_action_from_leg(leg) self.prepared_route_action_from_leg(leg)
for leg in self.plan_su_bundle_routes( for leg in self.plan_su_bundle_routes('u', contexts, offset, length, spacing, strategy=strategy, **route_bounds)
'u',
contexts,
offset,
length,
spacing,
operation,
diagnostic_request,
strategy=strategy,
**route_bounds,
)
)) ))
leg = self.plan_leg( leg = self.plan_leg('u', contexts[0], length=length, jog=offset, strategy=strategy, **route_bounds)
'u',
contexts[0],
operation,
diagnostic_request,
length=length,
jog=offset,
strategy=strategy,
**route_bounds,
)
return self.prepared_result_from_legs((leg,)) return self.prepared_result_from_legs((leg,))
def plan_su_bundle_routes( def plan_su_bundle_routes(
@ -1534,9 +1258,6 @@ class RoutingPlanner:
offset: float, offset: float,
length: float | None, length: float | None,
spacing: float | ArrayLike | None, spacing: float | ArrayLike | None,
operation: RouteOperation,
diagnostic_request: Mapping[str, Any],
/,
*, *,
strategy: RouteTieBreakStrategy | str | None = None, strategy: RouteTieBreakStrategy | str | None = None,
**kwargs: Any, **kwargs: Any,
@ -1549,32 +1270,14 @@ class RoutingPlanner:
and offset so all legs can be planned independently. and offset so all legs can be planned independently.
""" """
if len(contexts) == 1: if len(contexts) == 1:
return (self.plan_leg( return (self.plan_leg(kind, contexts[0], length=length, jog=offset, strategy=strategy, **kwargs),)
kind,
contexts[0],
operation,
diagnostic_request,
length=length,
jog=offset,
strategy=strategy,
**kwargs,
),)
route_name = 'jog' if kind == 's' else 'uturn' route_name = 'jog' if kind == 's' else 'uturn'
if kind == 'u' and is_close(offset, 0): if kind == 'u' and is_close(offset, 0):
raise BuildError('multi-port uturn() requires nonzero offset to determine bundle ordering') raise BuildError('multi-port uturn() requires nonzero offset to determine bundle ordering')
contexts_by_name = {context.portspec: context for context in contexts} 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) 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_portspec, _anchor_length, _anchor_offset = initial_specs[0]
anchor = self.plan_leg( anchor = self.plan_leg(kind, contexts_by_name[anchor_portspec], length=length, jog=offset, strategy=strategy, **kwargs)
kind,
contexts_by_name[anchor_portspec],
operation,
diagnostic_request,
length=length,
jog=offset,
strategy=strategy,
**kwargs,
)
base_length = anchor.candidate.public_length base_length = anchor.candidate.public_length
specs = planner_bounds.su_bundle_specs(contexts, offset, base_length, spacing, route_name=route_name) specs = planner_bounds.su_bundle_specs(contexts, offset, base_length, spacing, route_name=route_name)
first_portspec, _first_length, _first_offset = specs[0] first_portspec, _first_length, _first_offset = specs[0]
@ -1584,10 +1287,7 @@ class RoutingPlanner:
routes_by_name[spec_portspec] = self.plan_leg( routes_by_name[spec_portspec] = self.plan_leg(
'straight', 'straight',
contexts_by_name[spec_portspec], contexts_by_name[spec_portspec],
operation,
diagnostic_request,
length=spec_length, length=spec_length,
jog=spec_offset,
strategy=strategy, strategy=strategy,
**kwargs, **kwargs,
) )
@ -1595,8 +1295,6 @@ class RoutingPlanner:
routes_by_name[spec_portspec] = self.plan_leg( routes_by_name[spec_portspec] = self.plan_leg(
kind, kind,
contexts_by_name[spec_portspec], contexts_by_name[spec_portspec],
operation,
diagnostic_request,
length=spec_length, length=spec_length,
jog=spec_offset, jog=spec_offset,
strategy=strategy, strategy=strategy,
@ -1633,7 +1331,7 @@ class RoutingPlanner:
desired.rotation = port_dst.rotation - pi desired.rotation = port_dst.rotation - pi
desired.ptype = out_ptype desired.ptype = out_ptype
family, length, jog, ccw = self.trace_into_spec(context_src.port, desired) family, length, jog, ccw = self.trace_into_spec(context_src.port, desired)
request = self.solver_request( request = self.route_request(
family, family,
context_src, context_src,
length=length, length=length,
@ -1652,7 +1350,7 @@ class RoutingPlanner:
candidate = solver.solve(min_bends=min_bends, max_bends=max_bends) candidate = solver.solve(min_bends=min_bends, max_bends=max_bends)
break break
except (BuildError, NotImplementedError) as err: except (BuildError, NotImplementedError) as err:
if route_failure_policy(err) is RouteFailurePolicy.FATAL: if route_error_is_fatal(err):
raise raise
last_error = err last_error = err
if candidate is None: if candidate is None:

View file

@ -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 that same local frame. The planner transforms selected endpoints into layout
coordinates only after a complete route has been chosen. 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 Tool authors should treat offer planning callbacks as pure descriptions. The
solver may call `endpoint_at()` and `cost_at()` many times while enumerating solver may call `endpoint_at()`, `cost_at()`, and `bbox_at()` many times while
candidate compositions, ptype adapters, and parameter solutions. Those enumerating candidate compositions, ptype adapters, and parameter solutions.
callbacks should be deterministic and should not mutate a Library, Pattern, or Those callbacks should be deterministic and should not mutate a Library,
live Pather. `bbox_at()` follows the same purity contract, but footprint data is Pattern, or live Pather. `commit()` is the first selected-offer hook: it runs
currently reserved for future footprint-aware planning and is not consumed by only for primitives in the chosen route and returns the opaque value stored in
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
`RenderStep.data`. `RenderStep.data`.
`render()` is the geometry-construction hook. Pather calls it later with a `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)` Parameter domains are half-open `[min, max)` ranges, except `(value, value)`
is a closed singleton for fixed-size primitives. `None` and `"unk"` ptypes is a closed singleton for fixed-size primitives. `None` and `"unk"` ptypes
are wildcards; incompatible concrete ptypes are rejected by `Pather`. 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 in_ptype: str | None
out_ptype: str | None out_ptype: str | None
priority_bias: float = 0.0 priority_bias: float = 0.0
bbox_planner: BBoxCallable | None = None bbox_planner: BBoxCallable | None = None
parameterized_bbox: Any | None = None parameterized_bbox: Any | None = None
"""Reserved footprint metadata; the current solver does not inspect it."""
endpoint_planner: EndpointCallable | None = None endpoint_planner: EndpointCallable | None = None
commit_planner: CommitCallable | None = None commit_planner: CommitCallable | None = None
@ -245,11 +226,10 @@ class PrimitiveOffer(ABC):
def bbox_at(self, parameter: float) -> NDArray[numpy.float64]: 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 Tools may omit this hook by raising `NotImplementedError`; when present
it must return a finite `(2, 2)` min/max array in local coordinates. 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: if self.bbox_planner is None:
raise NotImplementedError raise NotImplementedError
@ -270,11 +250,9 @@ class PrimitiveOffer(ABC):
""" """
Produce opaque render data for a selected primitive. Produce opaque render data for a selected primitive.
Routing preparation calls this only for selected primitives while `Pather` calls this only for selected primitives while preparing
building `RenderStep.data`. Unselected candidates are evaluated by `RenderStep.data`. Unselected candidates are evaluated by endpoint/cost
endpoint/cost only and should not need commit-side work. This is a only and should not need commit-side work.
materialization hook, not the live-layout mutation boundary: it must not
mutate the caller's Pather, Pattern, or Library.
""" """
selected = self.canonicalize_parameter(parameter) selected = self.canonicalize_parameter(parameter)
if self.commit_planner is not None: if self.commit_planner is not None:

View file

@ -1,5 +1,5 @@
from typing import TYPE_CHECKING from typing import SupportsFloat, cast, TYPE_CHECKING
from collections.abc import Mapping from collections.abc import Mapping, Sequence
from pprint import pformat from pprint import pformat
import numpy import numpy
@ -13,17 +13,6 @@ if TYPE_CHECKING:
from ..ports import Port 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( def ell(
ports: Mapping[str, 'Port'], ports: Mapping[str, 'Port'],
ccw: SupportsBool | None, ccw: SupportsBool | None,
@ -93,18 +82,6 @@ def ell(
""" """
if not ports: if not ports:
raise BuildError('Empty port list passed to `ell()`') 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 ccw is None:
if spacing is not None and not numpy.allclose(spacing, 0): 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) travel = d_to_align - (ch_offsets.max() - ch_offsets)
offsets = travel - travel.min().clip(max=0) offsets = travel - travel.min().clip(max=0)
if bound_type in _EXTENSION_BOUND_TYPES: rot_bound: SupportsFloat
if numpy.any(bound_values < 0): if bound_type in ('emin', 'min_extension',
raise BuildError(f'Got negative bound for extension: {bound_values}') 'emax', 'max_extension',
'min_past_furthest',):
if bound_values.size == 2: if numpy.size(bound) == 2:
horizontal_weight = abs(float(numpy.cos(direction))) bound = cast('Sequence[float]', bound)
vertical_weight = abs(float(numpy.sin(direction))) rot_bound = (rot_matrix @ ((bound[0], 0),
use_x = horizontal_weight > vertical_weight or numpy.isclose(horizontal_weight, vertical_weight) (0, bound[1])))[0, :]
rot_bound = float(bound_values[0 if use_x else 1])
else: 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'): if bound_type in ('emin', 'min_extension', 'min_past_furthest'):
offsets += rot_bound offsets += rot_bound.max()
elif bound_type in ('emax', 'max_extension'): elif bound_type in ('emax', 'max_extension'):
offsets += rot_bound - offsets.max() offsets += rot_bound.min() - offsets.max()
else: else:
if bound_values.size == 2: if numpy.size(bound) == 2:
rot_bound = float((rot_matrix @ bound_values)[0]) bound = cast('Sequence[float]', bound)
rot_bound = (rot_matrix @ bound)[0]
else: else:
bound = cast('float', bound)
neg = (direction + pi / 4) % (2 * pi) > pi neg = (direction + pi / 4) % (2 * pi) > pi
bound_scalar = float(bound_values[0]) rot_bound = -bound if neg else bound
rot_bound = -bound_scalar if neg else bound_scalar
min_possible = x_start + offsets 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() extension = rot_bound - min_possible.max()
else: elif bound_type in ('pmin', 'min_position', 'xmin', 'ymin'):
extension = rot_bound - min_possible.min() extension = rot_bound - min_possible.min()
offsets += extension offsets += extension

View file

@ -3,7 +3,7 @@ import pytest
from numpy.testing import assert_equal, assert_allclose from numpy.testing import assert_equal, assert_allclose
from numpy import pi from numpy import pi
from ..builder import MinimumStatus, Pather, RouteFailureDetails from ..builder import Pather
from ..builder.utils import ell from ..builder.utils import ell
from ..error import BuildError from ..error import BuildError
from ..library import Library from ..library import Library
@ -14,50 +14,11 @@ from ..ports import Port
def test_builder_public_imports() -> None: def test_builder_public_imports() -> None:
from masque import PortPather as TopPortPather from masque import PortPather as TopPortPather
from masque import RenderStep as TopRenderStep 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 PortPather as BuilderPortPather
from masque.builder import RenderStep as BuilderRenderStep 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 TopPortPather is BuilderPortPather
assert TopRenderStep is BuilderRenderStep 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: 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'): with pytest.raises(BuildError, match='Spacing must be 0 or None'):
ell(ports, None, 'min_extension', 5, spacing=numpy.array([1, 0])) 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)

View file

@ -5,10 +5,9 @@ import pytest
import numpy import numpy
from numpy import pi 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 import RoutePortContext, RoutingPlanner
from masque.builder.planner.planner import NoLegalRouteError from masque.builder.tools import BendOffer, PathTool, RenderStep, StraightOffer, Tool
from masque.builder.tools import BendOffer, PathTool, RenderStep, StraightOffer, Tool, UOffer
from masque.error import BuildError from masque.error import BuildError
from masque.library import ILibrary from masque.library import ILibrary
@ -115,347 +114,15 @@ class CountingPathTool(PathTool):
return super().render(batch, port_names=port_names, **kwargs) 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: def test_pather_jog_failed_two_bend_route_is_atomic() -> None:
lib = Library() lib = Library()
tool = PathTool(layer='M1', width=2, ptype='wire') tool = PathTool(layer='M1', width=2, ptype='wire')
p = Pather(lib, tools=tool, render='immediate') p = Pather(lib, tools=tool, render='immediate')
p.pattern.ports['A'] = Port((0, 0), rotation=0, ptype='wire') 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) 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 numpy.allclose(p.pattern.ports['A'].offset, (0, 0))
assert p.pattern.ports['A'].rotation == 0 assert p.pattern.ports['A'].rotation == 0
assert len(p._paths['A']) == 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 = Pather(lib, tools=tool)
p.pattern.ports['A'] = Port((0, 0), rotation=0, ptype='wire') 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) 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 numpy.allclose(p.pattern.ports['A'].offset, (0, 0))
assert p.pattern.ports['A'].rotation == 0 assert p.pattern.ports['A'].rotation == 0
assert len(p._paths['A']) == 0 assert len(p._paths['A']) == 0

View file

@ -5,7 +5,7 @@ import numpy
from numpy import pi from numpy import pi
from numpy.testing import assert_allclose, assert_equal 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 import PathTool, PrimitiveOffer, StraightOffer
from masque.builder.planner import RoutingPlanner from masque.builder.planner import RoutingPlanner
from masque.error import BuildError, PortError 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 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: def test_pather_straight(pather_setup: tuple[Pather, PathTool, Library]) -> None:
p, tool, lib = pather_setup p, tool, lib = pather_setup
p.straight("start", 10) 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 = Pather(lib, ports={"in": Port((0, 0), 0)}, tools=tool)
p.set_dead() 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) 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() 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.pattern.ports['A'] = Port((0, 0), rotation=0, ptype='wire')
p.set_dead() 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 p.pattern.ports['A'].ptype == 'other'
assert len(p._paths['A']) == 0 assert len(p._paths['A']) == 0

View file

@ -6,7 +6,7 @@ import numpy
from numpy import pi from numpy import pi
from numpy.testing import assert_allclose from numpy.testing import assert_allclose
from ..builder import Pather, RouteError from ..builder import Pather
from ..builder.tools import PathTool, RenderStep, StraightOffer, Tool from ..builder.tools import PathTool, RenderStep, StraightOffer, Tool
from ..error import BuildError from ..error import BuildError
from ..library import Library 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[(1, 0)]) == 1
assert len(rp.pattern.shapes[(2, 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: def test_portpather_translate_only_affects_future_steps(deferred_render_setup: tuple[Pather, PathTool, Library]) -> None:
rp, tool, lib = deferred_render_setup rp, tool, lib = deferred_render_setup
pp = rp.at("start") 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 = Pather(lib, ports={"in": Port((0, 0), 0)}, tools=tool, render='deferred')
rp.set_dead() 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 assert len(rp._paths["in"]) == 0

View file

@ -5,9 +5,9 @@ import pytest
from numpy import pi from numpy import pi
from numpy.testing import assert_equal 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 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.builder.tools import BendOffer, PrimitiveOffer, StraightOffer, Tool
from masque.error import BuildError, PortError from masque.error import BuildError, PortError
@ -280,7 +280,7 @@ class TraceIntoBudgetSolver:
band = (min_bends, max_bends) band = (min_bends, max_bends)
self.attempts.append(band) self.attempts.append(band)
if band in self.fatal_at: if band in self.fatal_at:
raise RoutePlanningError('fatal', policy=RouteFailurePolicy.FATAL) raise RoutePlanningError('fatal', fatal=True)
if band not in self.successes: if band not in self.successes:
raise BuildError('try next budget') raise BuildError('try next budget')
return Candidate((), Port((0, 0), rotation=0, ptype='wire'), 0.0, 0, 0.0) 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 = TraceIntoBudgetSolver(successes, fatal_at=fatal_at)
self.solver_requests = 0 self.solver_requests = 0
def solver_for_request(self, request: SolverRequest) -> Any: def solver_for_request(self, request: RouteRequest) -> Any:
_ = request _ = request
self.solver_requests += 1 self.solver_requests += 1
return self.solver return self.solver