Compare commits

...

4 commits

14 changed files with 1145 additions and 125 deletions

View file

@ -82,6 +82,10 @@ 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,16 +8,23 @@ 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()`. For each operation, it asks the active `Tool` for `uturn()`, and `trace_into()`. The internal planner resolves each operation into
offers and passes those offers plus route constraints to the internal router. one or more `SolverRequest`s. This normalization is why the public routing API
The router selects a sequence of internal selected primitives, each pairing an can remain a convenient keyword-based interface without making the solver
offer with a concrete parameter, endpoint, and cost. 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.
Route commit is separate from route selection. Once a route is selected, Selection is pure with respect to caller-owned Pattern, Library, and Pather
`Pather` calls `offer.commit(parameter)` only for the selected primitives and state. Tool offer discovery and endpoint/cost/bbox callbacks must likewise be
stores the returned opaque tool payload in `RenderStep.data`. Later, deterministic and must not mutate that state. `commit()` is the first
`Pather.render()` batches compatible `RenderStep`s and calls `Tool.render()` to selected-offer materialization hook, but it still must not mutate the live
turn those committed payloads into geometry. 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.
`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
@ -31,6 +38,11 @@ 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
@ -41,6 +53,13 @@ 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,

102
masque/builder/error.py Normal file
View file

@ -0,0 +1,102 @@
"""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,10 +22,25 @@ 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 selected offers are committed and step queue, or Library are touched. Once prepared actions are applied, plug,
prepared actions are applied, later commit, plug, rename, render, or insertion rename, render, or insertion failures may leave partial output; that mutation
failures may leave partial output; that mutation boundary belongs to Pather, boundary belongs to Pather, not to Tool implementations or the route solver.
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
@ -55,8 +70,9 @@ from .tools import (
from .planner.interface import ( from .planner.interface import (
PreparedRouteResult, PreparedRouteResult,
RoutePortContext, RoutePortContext,
route_error_is_fatal, route_failure_policy,
) )
from .error import RouteFailurePolicy
from .planner import ( from .planner import (
RouteTieBreakStrategy, RouteTieBreakStrategy,
RoutingPlanner, RoutingPlanner,
@ -116,6 +132,9 @@ 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
@ -159,6 +178,16 @@ 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
@ -545,7 +574,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_error_is_fatal(err): if not self._dead or route_failure_policy(err) is RouteFailurePolicy.FATAL:
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]
@ -602,7 +631,11 @@ 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 not self._dead or len(contexts) != 1 or route_error_is_fatal(err): if (
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']
@ -674,7 +707,11 @@ 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 not self._dead or len(contexts) != 1 or route_error_is_fatal(err): if (
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:
@ -740,7 +777,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_error_is_fatal(err) or route_failure_policy(err) is RouteFailurePolicy.FATAL
): ):
raise raise
context = contexts[0] context = contexts[0]
@ -820,6 +857,11 @@ 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')
@ -892,7 +934,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 == batch[0].tool same_tool = batch and step.tool is 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 = []
@ -972,7 +1014,13 @@ 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_error_is_fatal as route_error_is_fatal, route_failure_policy as route_failure_policy,
) )
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,21 +15,30 @@ 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."""
fatal: bool policy: RouteFailurePolicy
def __init__(self, *args: object, fatal: bool = False) -> None: def __init__(
self,
*args: object,
policy: RouteFailurePolicy = RouteFailurePolicy.RECOVERABLE,
) -> None:
super().__init__(*args) super().__init__(*args)
self.fatal = fatal self.policy = policy
def route_error_is_fatal(err: Exception) -> bool: def route_failure_policy(err: Exception) -> RouteFailurePolicy:
"""Return true when a planning error should bypass dead-Pather fallback.""" """Return typed route recovery policy, defaulting generic errors to recoverable."""
return bool(getattr(err, 'fatal', False)) if isinstance(err, RoutePlanningError):
return err.policy
if isinstance(err, RouteError):
return err.policy
return RouteFailurePolicy.RECOVERABLE
@dataclass(frozen=True, slots=True) @dataclass(frozen=True, slots=True)
@ -54,9 +63,10 @@ class PreparedRouteAction:
""" """
Prepared mutation for one routed Pather port. Prepared mutation for one routed Pather port.
The planner has already committed selected primitive offers into Pure selection has already completed, and the planner has materialized the
`render_steps` and computed the final live port. `plug_into`, when set, selected primitive offers into `render_steps` and computed the final live
names the destination port to consume after the route endpoint is applied. port. `plug_into`, when set, names the destination port to consume after the
route endpoint is applied.
""" """
portspec: str portspec: str
"""Live Pather port name to update.""" """Live Pather port name to update."""
@ -73,9 +83,10 @@ class PreparedRouteResult:
""" """
Complete prepared result for one Pather routing operation. Complete prepared result for one Pather routing operation.
`actions` are applied first. `renames` are deferred until after all route `actions` contain materialized, committed render data and are applied first.
actions so trace-into/thru behavior can be represented without exposing the `renames` are deferred until after all route actions so trace-into/thru
solver's selected primitive sequence to Pather. behavior can be represented without exposing the solver's selected
primitive sequence to Pather.
""" """
actions: tuple[PreparedRouteAction, ...] actions: tuple[PreparedRouteAction, ...]
"""Prepared per-port route mutations.""" """Prepared per-port route mutations."""

View file

@ -6,11 +6,19 @@ 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` before the solver Public routing modes and bounds are normalized by `bounds.py` into per-leg
sees them. This module plans one route intent at a time: it queries Tool offers, `SolverRequest` values. `Solver` performs the pure search: it queries Tool
enumerates bounded primitive compositions, inserts ptype adapters, solves offers, enumerates bounded primitive compositions, inserts ptype adapters,
primitive parameters, ranks candidates, and commits only the selected offers solves primitive parameters, and returns the ranked `Candidate`. A `RouteLeg`
into `RenderStep` payloads. 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.
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
@ -22,7 +30,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 from dataclasses import dataclass, replace
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
@ -43,6 +51,13 @@ 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 (
@ -50,12 +65,16 @@ from .interface import (
PreparedRouteResult, PreparedRouteResult,
RoutePlanningError, RoutePlanningError,
RoutePortContext, RoutePortContext,
route_error_is_fatal, route_failure_policy,
) )
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'):
@ -144,7 +163,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 getattr(err, 'fatal', False): if route_failure_policy(err) is RouteFailurePolicy.FATAL:
raise err raise err
@ -226,9 +245,9 @@ class Candidate:
@dataclass(frozen=True, slots=True) @dataclass(frozen=True, slots=True)
class RouteRequest: class SolverRequest:
""" """
Normalized solver input for one route leg. Normalized input for one `Solver` invocation and 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;
@ -297,7 +316,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: RouteRequest) -> None: def __init__(self, request: SolverRequest) -> 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[
@ -391,14 +410,16 @@ class Solver:
if not candidates: if not candidates:
for err in errors: for err in errors:
if getattr(err, 'fatal', False): if route_failure_policy(err) is RouteFailurePolicy.FATAL:
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 last_error raise NoLegalRouteError(str(last_error)) from last_error
raise BuildError(f'{self.request.route_name} route is unsupported: {last_error}') from last_error raise NoLegalRouteError(
raise BuildError(f'No legal primitive offer for {self.request.route_name}') 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}')
return min( return min(
candidates, candidates,
@ -469,12 +490,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',
fatal=True, policy=RouteFailurePolicy.FATAL,
) )
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',
fatal=True, policy=RouteFailurePolicy.FATAL,
) )
cost = float(offer.cost_at(selected)) cost = float(offer.cost_at(selected))
if not numpy.isfinite(cost): if not numpy.isfinite(cost):
@ -986,7 +1007,7 @@ class RoutingPlanner:
bands.append((4, 4)) bands.append((4, 4))
return tuple(bands) return tuple(bands)
def route_request( def solver_request(
self, self,
family: PrimitiveKind, family: PrimitiveKind,
context: RoutePortContext, context: RoutePortContext,
@ -998,9 +1019,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,
) -> RouteRequest: ) -> SolverRequest:
"""Build normalized solver input for one route leg.""" """Build normalized solver input for one route leg."""
return RouteRequest( return SolverRequest(
family=family, family=family,
tool=context.tool, tool=context.tool,
in_ptype=context.port.ptype, in_ptype=context.port.ptype,
@ -1014,7 +1035,7 @@ class RoutingPlanner:
strategy=self.resolve_strategy(strategy), strategy=self.resolve_strategy(strategy),
) )
def solver_for_request(self, request: RouteRequest) -> Solver: def solver_for_request(self, request: SolverRequest) -> Solver:
"""Construct the solver for a route request.""" """Construct the solver for a route request."""
return Solver(request) return Solver(request)
@ -1038,6 +1059,9 @@ 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,
@ -1049,7 +1073,32 @@ 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."""
request = self.route_request( 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(
family=family, family=family,
context=context, context=context,
length=length, length=length,
@ -1062,10 +1111,48 @@ class RoutingPlanner:
) )
try: try:
candidate = self.solver_for_request(request).solve() candidate = self.solver_for_request(request).solve()
except BuildError as err: except NoLegalRouteError as err:
if family == 'u' and length is None and not getattr(err, 'fatal', False): cause = (
raise BuildError('No legal primitive offer for omitted-length U-turn') from err '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 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(
@ -1129,25 +1216,87 @@ 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(family, contexts[0], length=length, ccw=ccw, strategy=strategy, **route_bounds) leg = self.plan_leg(
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(family, context, length=each, ccw=ccw, strategy=strategy, **route_bounds), self.plan_leg(
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(family, contexts[0], length=None, ccw=ccw, strategy=strategy, **route_bounds) leg = self.plan_leg(
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]
@ -1164,7 +1313,16 @@ 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(family, context, length=route_length, ccw=ccw, strategy=strategy, **route_bounds) leg = self.plan_leg(
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))
@ -1179,6 +1337,37 @@ 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:
@ -1186,7 +1375,16 @@ 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(contexts, ccw, spacing=spacing, strategy=strategy, **route_bounds) return self._plan_trace_route(
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
@ -1196,7 +1394,16 @@ 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(family, contexts[0], length=length, ccw=ccw, strategy=strategy, **other_bounds) leg = self.plan_leg(
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(
@ -1210,8 +1417,28 @@ 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(contexts, None, length=length, spacing=spacing, strategy=strategy, **bounds) return self._plan_trace_to_route(
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)
@ -1224,9 +1451,28 @@ 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('s', contexts, offset, length, spacing, strategy=strategy, **other_bounds) for leg in self.plan_su_bundle_routes(
's',
contexts,
offset,
length,
spacing,
operation,
diagnostic_request,
strategy=strategy,
**other_bounds,
)
)) ))
leg = self.plan_leg('s', contexts[0], length=length, jog=offset, strategy=strategy, **other_bounds) leg = self.plan_leg(
'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(
@ -1241,14 +1487,44 @@ 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('u', contexts, offset, length, spacing, strategy=strategy, **route_bounds) for leg in self.plan_su_bundle_routes(
'u',
contexts,
offset,
length,
spacing,
operation,
diagnostic_request,
strategy=strategy,
**route_bounds,
)
)) ))
leg = self.plan_leg('u', contexts[0], length=length, jog=offset, strategy=strategy, **route_bounds) leg = self.plan_leg(
'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(
@ -1258,6 +1534,9 @@ 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,
@ -1270,14 +1549,32 @@ 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(kind, contexts[0], length=length, jog=offset, strategy=strategy, **kwargs),) return (self.plan_leg(
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(kind, contexts_by_name[anchor_portspec], length=length, jog=offset, strategy=strategy, **kwargs) anchor = self.plan_leg(
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]
@ -1287,7 +1584,10 @@ 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,
) )
@ -1295,6 +1595,8 @@ 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,
@ -1331,7 +1633,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.route_request( request = self.solver_request(
family, family,
context_src, context_src,
length=length, length=length,
@ -1350,7 +1652,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_error_is_fatal(err): if route_failure_policy(err) is RouteFailurePolicy.FATAL:
raise raise
last_error = err last_error = err
if candidate is None: if candidate is None:

View file

@ -15,12 +15,23 @@ 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()`, `cost_at()`, and `bbox_at()` many times while solver may call `endpoint_at()` and `cost_at()` many times while enumerating
enumerating candidate compositions, ptype adapters, and parameter solutions. candidate compositions, ptype adapters, and parameter solutions. Those
Those callbacks should be deterministic and should not mutate a Library, callbacks should be deterministic and should not mutate a Library, Pattern, or
Pattern, or live Pather. `commit()` is the first selected-offer hook: it runs live Pather. `bbox_at()` follows the same purity contract, but footprint data is
only for primitives in the chosen route and returns the opaque value stored in 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
`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
@ -166,12 +177,20 @@ 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
@ -226,10 +245,11 @@ 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 footprint-aware planning. Return local primitive bounds for future 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
@ -250,9 +270,11 @@ class PrimitiveOffer(ABC):
""" """
Produce opaque render data for a selected primitive. Produce opaque render data for a selected primitive.
`Pather` calls this only for selected primitives while preparing Routing preparation calls this only for selected primitives while
`RenderStep.data`. Unselected candidates are evaluated by endpoint/cost building `RenderStep.data`. Unselected candidates are evaluated by
only and should not need commit-side work. 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.
""" """
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 SupportsFloat, cast, TYPE_CHECKING from typing import TYPE_CHECKING
from collections.abc import Mapping, Sequence from collections.abc import Mapping
from pprint import pformat from pprint import pformat
import numpy import numpy
@ -13,6 +13,17 @@ 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,
@ -82,6 +93,18 @@ 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):
@ -173,38 +196,34 @@ 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)
rot_bound: SupportsFloat if bound_type in _EXTENSION_BOUND_TYPES:
if bound_type in ('emin', 'min_extension', if numpy.any(bound_values < 0):
'emax', 'max_extension', raise BuildError(f'Got negative bound for extension: {bound_values}')
'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:
bound = cast('float', bound)
rot_bound = numpy.array(bound)
if rot_bound < 0: if bound_values.size == 2:
raise BuildError(f'Got negative bound for extension: {rot_bound}') 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])
else:
rot_bound = float(bound_values[0])
if bound_type in ('emin', 'min_extension', 'min_past_furthest'): if bound_type in ('emin', 'min_extension', 'min_past_furthest'):
offsets += rot_bound.max() offsets += rot_bound
elif bound_type in ('emax', 'max_extension'): elif bound_type in ('emax', 'max_extension'):
offsets += rot_bound.min() - offsets.max() offsets += rot_bound - offsets.max()
else: else:
if numpy.size(bound) == 2: if bound_values.size == 2:
bound = cast('Sequence[float]', bound) rot_bound = float((rot_matrix @ bound_values)[0])
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
rot_bound = -bound if neg else bound bound_scalar = float(bound_values[0])
rot_bound = -bound_scalar if neg else bound_scalar
min_possible = x_start + offsets min_possible = x_start + offsets
if bound_type in ('pmax', 'max_position', 'xmax', 'ymax'): if bound_type in _POSITION_MAX_BOUND_TYPES:
extension = rot_bound - min_possible.max() extension = rot_bound - min_possible.max()
elif bound_type in ('pmin', 'min_position', 'xmin', 'ymin'): else:
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 Pather from ..builder import MinimumStatus, Pather, RouteFailureDetails
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,11 +14,50 @@ 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:
@ -171,3 +210,41 @@ 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,9 +5,10 @@ import pytest
import numpy import numpy
from numpy import pi from numpy import pi
from masque import Pather, Library, Port from masque import MinimumStatus, Pather, Library, Port, RouteError, RouteFailurePolicy
from masque.builder.planner import RoutePortContext, RoutingPlanner from masque.builder.planner import RoutePortContext, RoutingPlanner
from masque.builder.tools import BendOffer, PathTool, RenderStep, StraightOffer, Tool from masque.builder.planner.planner import NoLegalRouteError
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
@ -114,15 +115,347 @@ 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(BuildError, match='S-bend'): with pytest.raises(RouteError, match='S-bend') as exc_info:
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
@ -593,9 +926,12 @@ 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(BuildError, match='U-turn'): with pytest.raises(RouteError, match='U-turn') as exc_info:
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 from masque import Pather, Library, Pattern, Port, RouteError
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,6 +93,30 @@ 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)
@ -136,12 +160,13 @@ 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, [10, 0], atol=1e-10) assert_allclose(p.ports["in"].offset, [0, 0], atol=1e-10)
p.straight("in", 20) p.straight("in", 20)
assert_allclose(p.ports["in"].offset, [-10, 0], atol=1e-10) assert_allclose(p.ports["in"].offset, [-20, 0], atol=1e-10)
assert not p.pattern.has_shapes() assert not p.pattern.has_shapes()
@ -394,9 +419,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 from ..builder import Pather, RouteError
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,6 +101,60 @@ 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")
@ -122,9 +176,10 @@ 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, [10, 0], atol=1e-10) assert_allclose(rp.ports["in"].offset, [0, 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 from masque import Library, PathTool, Port, Pather, RouteFailurePolicy
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, RouteRequest from masque.builder.planner.planner import Candidate, SolverRequest
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', fatal=True) raise RoutePlanningError('fatal', policy=RouteFailurePolicy.FATAL)
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: RouteRequest) -> Any: def solver_for_request(self, request: SolverRequest) -> Any:
_ = request _ = request
self.solver_requests += 1 self.solver_requests += 1
return self.solver return self.solver