[Pather/Planner] Improve error reporting

This commit is contained in:
Jan Petykiewicz 2026-07-12 15:27:14 -07:00
commit 3a3cd854a5
9 changed files with 777 additions and 58 deletions

View file

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

View file

@ -41,6 +41,11 @@ from .pather import (
Pather as Pather,
PortPather as PortPather,
)
from .error import (
RouteError as RouteError,
RouteFailureDetails as RouteFailureDetails,
RouteOperation as RouteOperation,
)
from .utils import ell as ell
from .tools import (
Tool as Tool,

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

@ -0,0 +1,65 @@
"""Public routing failure diagnostics."""
from typing import Any, Literal
from collections.abc import Mapping
from dataclasses import dataclass
from pprint import pformat
from types import MappingProxyType
from ..error import BuildError
RouteOperation = Literal['trace', 'trace_to', 'jog', 'uturn']
@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_attempted: bool
minimum_exhausted: bool
cause: str
minimum_cause: str | None = None
def __post_init__(self) -> None:
object.__setattr__(self, 'request', MappingProxyType(dict(self.request)))
class RouteError(BuildError):
"""A route-selection failure with structured request diagnostics."""
details: RouteFailureDetails
fatal: bool
def __init__(self, details: RouteFailureDetails, *, fatal: bool = False) -> None:
self.details = details
self.fatal = fatal
if not details.minimum_attempted:
minimum = 'not evaluated'
elif details.minimum_length is not None:
minimum = f'{details.minimum_length:g}'
elif details.minimum_exhausted:
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,7 +22,7 @@ from __future__ import annotations
# ruff: noqa: ANN401,PLR0912,PLR0913,PLR0915,TC001,TC002,TC003
from collections.abc import Iterable, Mapping, Sequence
from dataclasses import dataclass
from dataclasses import dataclass, replace
from itertools import combinations
from math import cos, isclose as math_isclose, sin
from typing import Any, Literal
@ -43,6 +43,7 @@ from ..tools import (
StraightOffer,
Tool,
)
from ..error import RouteError, RouteFailureDetails, RouteOperation
from ..utils import ell
from . import bounds as planner_bounds
from .interface import (
@ -56,6 +57,10 @@ from .interface import (
RouteTieBreakStrategy = Literal['straight_first', 'turn_first']
class NoLegalRouteError(BuildError):
"""Internal marker for exhaustive candidate-selection failure."""
def validate_strategy(strategy: RouteTieBreakStrategy | str) -> RouteTieBreakStrategy:
"""Return a supported route tie-break strategy or raise a routing error."""
if strategy in ('straight_first', 'turn_first'):
@ -396,9 +401,11 @@ class Solver:
if errors:
last_error = errors[-1]
if self.request.route_name in str(last_error):
raise last_error
raise BuildError(f'{self.request.route_name} route is unsupported: {last_error}') from last_error
raise BuildError(f'No legal primitive offer for {self.request.route_name}')
raise NoLegalRouteError(str(last_error)) from last_error
raise NoLegalRouteError(
f'{self.request.route_name} route is unsupported: {last_error}'
) from last_error
raise NoLegalRouteError(f'No legal primitive offer for {self.request.route_name}')
return min(
candidates,
@ -1038,6 +1045,9 @@ class RoutingPlanner:
self,
family: PrimitiveKind,
context: RoutePortContext,
operation: RouteOperation | None = None,
diagnostic_request: Mapping[str, Any] | None = None,
/,
*,
length: float | None = None,
jog: float | None = None,
@ -1049,6 +1059,32 @@ class RoutingPlanner:
**kwargs: Any,
) -> RouteLeg:
"""Solve one route leg and attach it to its source Pather context."""
if operation is None:
if family in ('straight', 'bend'):
operation = 'trace'
elif family == 's':
operation = 'jog'
else:
operation = 'uturn'
if diagnostic_request is None:
diagnostic_request = {}
if length is not None and (not numpy.isfinite(length) or length < 0):
details = RouteFailureDetails(
operation=operation,
portspec=context.portspec,
in_ptype=context.port.ptype,
out_ptype=kwargs.get('out_ptype'),
request=diagnostic_request,
resolved_length=float(length),
resolved_jog=None if jog is None else float(jog),
minimum_length=None,
minimum_attempted=False,
minimum_exhausted=False,
cause='Resolved route length must be finite and nonnegative',
)
raise RouteError(details, fatal=True)
request = self.route_request(
family=family,
context=context,
@ -1062,10 +1098,47 @@ class RoutingPlanner:
)
try:
candidate = self.solver_for_request(request).solve()
except BuildError as err:
if family == 'u' and length is None and not getattr(err, 'fatal', False):
raise BuildError('No legal primitive offer for omitted-length U-turn') from err
except NoLegalRouteError as err:
cause = (
'No legal primitive offer for omitted-length U-turn'
if family == 'u' and length is None
else str(err)
)
minimum_length: float | None = None
minimum_cause: str | None = None
minimum_exhausted = False
if length is None:
minimum_cause = cause
minimum_exhausted = True
else:
minimum_request = replace(request, length=None)
try:
minimum_candidate = self.solver_for_request(minimum_request).solve()
minimum_length = minimum_candidate.public_length
except NoLegalRouteError as minimum_err:
minimum_cause = str(minimum_err)
minimum_exhausted = True
except Exception as minimum_err:
if route_error_is_fatal(minimum_err):
raise
minimum_cause = str(minimum_err)
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_attempted=True,
minimum_exhausted=minimum_exhausted,
cause=cause,
minimum_cause=minimum_cause,
)
raise RouteError(details) from err
return self.route_leg_from_candidate(context, candidate, plug_into=plug_into)
def prepared_route_action_from_leg(
@ -1129,25 +1202,87 @@ class RoutingPlanner:
) -> PreparedRouteResult:
"""Plan straight or single-bend traces, including `each` and bundle-bound modes."""
route_bounds = dict(bounds)
request_details = {'ccw': ccw, **{
key: value for key, value in route_bounds.items() if value is not None
}}
if length is not None:
request_details['length'] = length
if spacing is not None:
request_details['spacing'] = spacing
if strategy is not None:
request_details['strategy'] = strategy
operation: RouteOperation = 'trace'
diagnostic_request = request_details
return self._plan_trace_route(
contexts,
ccw,
length,
operation,
diagnostic_request,
spacing=spacing,
strategy=strategy,
**route_bounds,
)
def _plan_trace_route(
self,
contexts: Sequence[RoutePortContext],
ccw: SupportsBool | None,
length: float | None,
operation: RouteOperation,
diagnostic_request: Mapping[str, Any],
/,
*,
spacing: float | ArrayLike | None,
strategy: RouteTieBreakStrategy | str | None,
**route_bounds: Any,
) -> PreparedRouteResult:
portspec = tuple(context.portspec for context in contexts)
planner_bounds.validate_trace_args(portspec, length=length, spacing=spacing, bounds=route_bounds)
family: Literal['straight', 'bend'] = 'straight' if ccw is None else 'bend'
if length is not None:
leg = self.plan_leg(family, contexts[0], 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,))
if route_bounds.get('each') is not None:
each = route_bounds.pop('each')
return PreparedRouteResult(tuple(
self.prepared_route_action_from_leg(
self.plan_leg(family, context, length=each, ccw=ccw, strategy=strategy, **route_bounds),
self.plan_leg(
family,
context,
operation,
diagnostic_request,
length=each,
ccw=ccw,
strategy=strategy,
**route_bounds,
),
)
for context in contexts
))
bundle_bounds = planner_bounds.present_bundle_bounds(route_bounds)
if not bundle_bounds:
leg = self.plan_leg(family, contexts[0], length=None, ccw=ccw, 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,))
bound_type = bundle_bounds[0]
@ -1164,7 +1299,16 @@ class RoutingPlanner:
actions = []
for port_name, route_length in extensions.items():
context = next(context for context in contexts if context.portspec == port_name)
leg = self.plan_leg(family, context, length=route_length, ccw=ccw, 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))
return PreparedRouteResult(tuple(actions))
@ -1179,6 +1323,37 @@ class RoutingPlanner:
) -> PreparedRouteResult:
"""Plan `trace_to()` by resolving positional targets or delegating to `trace()` modes."""
route_bounds = dict(bounds)
request_details = {'ccw': ccw, **{
key: value for key, value in route_bounds.items() if value is not None
}}
if spacing is not None:
request_details['spacing'] = spacing
if strategy is not None:
request_details['strategy'] = strategy
operation: RouteOperation = 'trace_to'
diagnostic_request = request_details
return self._plan_trace_to_route(
contexts,
ccw,
operation,
diagnostic_request,
spacing=spacing,
strategy=strategy,
**route_bounds,
)
def _plan_trace_to_route(
self,
contexts: Sequence[RoutePortContext],
ccw: SupportsBool | None,
operation: RouteOperation,
diagnostic_request: Mapping[str, Any],
/,
*,
spacing: float | ArrayLike | None,
strategy: RouteTieBreakStrategy | str | None,
**route_bounds: Any,
) -> PreparedRouteResult:
if len(contexts) == 1:
resolved = planner_bounds.resolved_position_bound(contexts[0].port, route_bounds, allow_length=False)
else:
@ -1186,7 +1361,16 @@ class RoutingPlanner:
if any(route_bounds.get(key) is not None for key in planner_bounds.POSITION_KEYS):
raise BuildError('Position bounds only allowed with a single port')
if resolved is None:
return self.plan_trace_route(contexts, ccw, spacing=spacing, 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)
_key, _value, length = resolved
@ -1196,7 +1380,16 @@ class RoutingPlanner:
if key not in planner_bounds.POSITION_KEYS and key != 'length'
}
family: Literal['straight', 'bend'] = 'straight' if ccw is None else 'bend'
leg = self.plan_leg(family, contexts[0], length=length, ccw=ccw, 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,))
def plan_jog_route(
@ -1210,8 +1403,28 @@ class RoutingPlanner:
**bounds: Any,
) -> PreparedRouteResult:
"""Plan S-bend routes for single ports or spaced bundles."""
request_details = {'offset': offset, **{
key: value for key, value in bounds.items() if value is not None
}}
if length is not None:
request_details['length'] = length
if spacing is not None:
request_details['spacing'] = spacing
if strategy is not None:
request_details['strategy'] = strategy
operation: RouteOperation = 'jog'
diagnostic_request = request_details
if numpy.isclose(offset, 0):
return self.plan_trace_to_route(contexts, None, 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)
portspec = tuple(context.portspec for context in contexts)
planner_bounds.validate_jog_args(portspec, length=length, spacing=spacing, bounds=route_bounds)
@ -1224,9 +1437,28 @@ class RoutingPlanner:
if len(contexts) > 1:
return PreparedRouteResult(tuple(
self.prepared_route_action_from_leg(leg)
for leg in self.plan_su_bundle_routes('s', contexts, offset, length, spacing, 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,))
def plan_uturn_route(
@ -1241,14 +1473,44 @@ class RoutingPlanner:
) -> PreparedRouteResult:
"""Plan U-turn routes for single ports or spaced bundles."""
route_bounds = dict(bounds)
request_details = {'offset': offset, **{
key: value for key, value in route_bounds.items() if value is not None
}}
if length is not None:
request_details['length'] = length
if spacing is not None:
request_details['spacing'] = spacing
if strategy is not None:
request_details['strategy'] = strategy
operation: RouteOperation = 'uturn'
diagnostic_request = request_details
portspec = tuple(context.portspec for context in contexts)
planner_bounds.validate_uturn_args(portspec, spacing=spacing, bounds=route_bounds)
if len(contexts) > 1:
return PreparedRouteResult(tuple(
self.prepared_route_action_from_leg(leg)
for leg in self.plan_su_bundle_routes('u', contexts, offset, length, spacing, 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,))
def plan_su_bundle_routes(
@ -1258,6 +1520,9 @@ class RoutingPlanner:
offset: float,
length: float | None,
spacing: float | ArrayLike | None,
operation: RouteOperation,
diagnostic_request: Mapping[str, Any],
/,
*,
strategy: RouteTieBreakStrategy | str | None = None,
**kwargs: Any,
@ -1270,14 +1535,32 @@ class RoutingPlanner:
and offset so all legs can be planned independently.
"""
if len(contexts) == 1:
return (self.plan_leg(kind, contexts[0], length=length, jog=offset, 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'
if kind == 'u' and is_close(offset, 0):
raise BuildError('multi-port uturn() requires nonzero offset to determine bundle ordering')
contexts_by_name = {context.portspec: context for context in contexts}
initial_specs = planner_bounds.su_bundle_specs(contexts, offset, 0, spacing, route_name=route_name)
anchor_portspec, _anchor_length, _anchor_offset = initial_specs[0]
anchor = self.plan_leg(kind, contexts_by_name[anchor_portspec], length=length, jog=offset, 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
specs = planner_bounds.su_bundle_specs(contexts, offset, base_length, spacing, route_name=route_name)
first_portspec, _first_length, _first_offset = specs[0]
@ -1287,7 +1570,10 @@ class RoutingPlanner:
routes_by_name[spec_portspec] = self.plan_leg(
'straight',
contexts_by_name[spec_portspec],
operation,
diagnostic_request,
length=spec_length,
jog=spec_offset,
strategy=strategy,
**kwargs,
)
@ -1295,6 +1581,8 @@ class RoutingPlanner:
routes_by_name[spec_portspec] = self.plan_leg(
kind,
contexts_by_name[spec_portspec],
operation,
diagnostic_request,
length=spec_length,
jog=spec_offset,
strategy=strategy,

View file

@ -1,5 +1,5 @@
from typing import SupportsFloat, cast, TYPE_CHECKING
from collections.abc import Mapping, Sequence
from typing import TYPE_CHECKING
from collections.abc import Mapping
from pprint import pformat
import numpy
@ -13,6 +13,17 @@ if TYPE_CHECKING:
from ..ports import Port
_EXTENSION_BOUND_TYPES = (
'emin', 'min_extension',
'emax', 'max_extension',
'min_past_furthest',
)
_POSITION_MIN_BOUND_TYPES = ('pmin', 'min_position', 'xmin', 'ymin')
_POSITION_MAX_BOUND_TYPES = ('pmax', 'max_position', 'xmax', 'ymax')
_POSITION_BOUND_TYPES = _POSITION_MIN_BOUND_TYPES + _POSITION_MAX_BOUND_TYPES
_BOUND_TYPES = _EXTENSION_BOUND_TYPES + _POSITION_BOUND_TYPES
def ell(
ports: Mapping[str, 'Port'],
ccw: SupportsBool | None,
@ -82,6 +93,18 @@ def ell(
"""
if not ports:
raise BuildError('Empty port list passed to `ell()`')
if bound_type not in _BOUND_TYPES:
raise BuildError(f'Invalid bound type {bound_type!r}; expected one of {_BOUND_TYPES}')
try:
bound_arr = numpy.asarray(bound, dtype=float)
except (TypeError, ValueError) as err:
raise BuildError('bound must be a numeric scalar or length-2 vector') from err
if bound_arr.size not in (1, 2):
raise BuildError(f'bound must be scalar or have length 2; got {bound_arr.size} values')
if not numpy.all(numpy.isfinite(bound_arr)):
raise BuildError('bound must contain only finite values')
bound_values = bound_arr.reshape(-1)
if ccw is None:
if spacing is not None and not numpy.allclose(spacing, 0):
@ -173,38 +196,34 @@ def ell(
travel = d_to_align - (ch_offsets.max() - ch_offsets)
offsets = travel - travel.min().clip(max=0)
rot_bound: SupportsFloat
if bound_type in ('emin', 'min_extension',
'emax', 'max_extension',
'min_past_furthest',):
if numpy.size(bound) == 2:
bound = cast('Sequence[float]', bound)
rot_bound = (rot_matrix @ ((bound[0], 0),
(0, bound[1])))[0, :]
else:
bound = cast('float', bound)
rot_bound = numpy.array(bound)
if bound_type in _EXTENSION_BOUND_TYPES:
if numpy.any(bound_values < 0):
raise BuildError(f'Got negative bound for extension: {bound_values}')
if rot_bound < 0:
raise BuildError(f'Got negative bound for extension: {rot_bound}')
if bound_values.size == 2:
horizontal_weight = abs(float(numpy.cos(direction)))
vertical_weight = abs(float(numpy.sin(direction)))
use_x = horizontal_weight > vertical_weight or numpy.isclose(horizontal_weight, vertical_weight)
rot_bound = float(bound_values[0 if use_x else 1])
else:
rot_bound = float(bound_values[0])
if bound_type in ('emin', 'min_extension', 'min_past_furthest'):
offsets += rot_bound.max()
offsets += rot_bound
elif bound_type in ('emax', 'max_extension'):
offsets += rot_bound.min() - offsets.max()
offsets += rot_bound - offsets.max()
else:
if numpy.size(bound) == 2:
bound = cast('Sequence[float]', bound)
rot_bound = (rot_matrix @ bound)[0]
if bound_values.size == 2:
rot_bound = float((rot_matrix @ bound_values)[0])
else:
bound = cast('float', bound)
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
if bound_type in ('pmax', 'max_position', 'xmax', 'ymax'):
if bound_type in _POSITION_MAX_BOUND_TYPES:
extension = rot_bound - min_possible.max()
elif bound_type in ('pmin', 'min_position', 'xmin', 'ymin'):
else:
extension = rot_bound - min_possible.min()
offsets += extension

View file

@ -14,11 +14,17 @@ from ..ports import Port
def test_builder_public_imports() -> None:
from masque import PortPather as TopPortPather
from masque import RenderStep as TopRenderStep
from masque import RouteError as TopRouteError
from masque import RouteFailureDetails as TopRouteFailureDetails
from masque.builder import PortPather as BuilderPortPather
from masque.builder import RenderStep as BuilderRenderStep
from masque.builder import RouteError as BuilderRouteError
from masque.builder import RouteFailureDetails as BuilderRouteFailureDetails
assert TopPortPather is BuilderPortPather
assert TopRenderStep is BuilderRenderStep
assert TopRouteError is BuilderRouteError
assert TopRouteFailureDetails is BuilderRouteFailureDetails
def test_builder_init() -> None:
@ -171,3 +177,41 @@ def test_ell_handles_array_spacing_when_ccw_none() -> None:
with pytest.raises(BuildError, match='Spacing must be 0 or None'):
ell(ports, None, 'min_extension', 5, spacing=numpy.array([1, 0]))
@pytest.mark.parametrize('bound_type', ['emin', 'emax', 'min_past_furthest'])
@pytest.mark.parametrize(
('rotation', 'expected'),
[
(0, 5),
(pi / 2, 7),
(pi / 6, 5),
(pi / 3, 7),
(pi / 4, 5),
],
)
def test_ell_extension_vector_selects_dominant_route_axis(
bound_type: str,
rotation: float,
expected: float,
) -> None:
result = ell({'A': Port((0, 0), rotation)}, None, bound_type, (5, 7), spacing=0)
assert_allclose(result['A'], expected)
@pytest.mark.parametrize('bound', [(-1, 2), (1, -2), (numpy.nan, 2), (1, numpy.inf), (1, 2, 3)])
def test_ell_rejects_invalid_extension_vector(bound: tuple[float, ...]) -> None:
with pytest.raises(BuildError, match='bound|negative'):
ell({'A': Port((0, 0), 0)}, None, 'emin', bound, spacing=0)
def test_ell_position_vector_still_projects_onto_route_direction() -> None:
result = ell({'A': Port((0, 0), pi)}, None, 'pmax', (5, 7), spacing=0)
assert_allclose(result['A'], 5)
def test_ell_rejects_invalid_bound_type() -> None:
with pytest.raises(BuildError, match='Invalid bound type'):
ell({'A': Port((0, 0), 0)}, None, 'nearest', 5, spacing=0)

View file

@ -5,9 +5,9 @@ import pytest
import numpy
from numpy import pi
from masque import Pather, Library, Port
from masque import Pather, Library, Port, RouteError
from masque.builder.planner import RoutePortContext, RoutingPlanner
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.library import ILibrary
@ -114,15 +114,306 @@ class CountingPathTool(PathTool):
return super().render(batch, port_names=port_names, **kwargs)
class RequestCountingTool(PlanningOnlyTool):
def __init__(self) -> None:
self.offer_calls = 0
def primitive_offers(
self,
kind: Literal['straight', 'bend', 's', 'u'],
*,
in_ptype: str | None = None,
out_ptype: str | None = None,
**kwargs: Any,
) -> tuple[Any, ...]:
self.offer_calls += 1
return super().primitive_offers(
kind,
in_ptype=in_ptype,
out_ptype=out_ptype,
**kwargs,
)
class PreferredMinimumTool(PlanningOnlyTool):
def __init__(self) -> None:
self.commit_calls = 0
self.diagnostic_context_kwargs: list[Any] = []
def _commit(self, parameter: float) -> dict[str, float]:
self.commit_calls += 1
return {'parameter': parameter}
def primitive_offers(
self,
kind: Literal['straight', 'bend', 's', 'u'],
*,
in_ptype: str | None = None,
out_ptype: str | None = None,
**kwargs: Any,
) -> tuple[Any, ...]:
_ = out_ptype
self.diagnostic_context_kwargs.append(kwargs.get('diagnostic_context'))
if kind == 'bend':
ccw = bool(kwargs['ccw'])
rotation = -pi / 2 if ccw else pi / 2
jog = 1 if ccw else -1
return (
BendOffer(
in_ptype=in_ptype,
out_ptype='wide',
priority_bias=100,
ccw=ccw,
length_domain=(2, 2),
endpoint_planner=lambda _length: Port((2, jog), rotation, ptype='wide'),
commit_planner=self._commit,
),
BendOffer(
in_ptype=in_ptype,
out_ptype='wide',
ccw=ccw,
length_domain=(5, 5),
endpoint_planner=lambda _length: Port((5, jog), rotation, ptype='wide'),
commit_planner=self._commit,
),
)
if kind == 'u':
return (UOffer(
in_ptype=in_ptype,
out_ptype='wide',
jog_domain=(4, 4),
endpoint_planner=lambda _jog: Port((5, 4), 0, ptype='wide'),
commit_planner=self._commit,
),)
return ()
def test_route_error_reports_preferred_minimum_and_request_details() -> None:
tool = PreferredMinimumTool()
p = Pather(
Library(),
ports={'A': Port((0, 0), rotation=0, ptype='wire')},
tools=tool,
render='deferred',
)
with pytest.raises(RouteError) as exc_info:
p.ccw('A', 1, out_ptype='wide', diagnostic_context='tool-value')
details = exc_info.value.details
assert isinstance(exc_info.value, BuildError)
assert details.operation == 'trace_to'
assert details.portspec == 'A'
assert details.in_ptype == 'wire'
assert details.out_ptype == 'wide'
assert details.request == {
'ccw': True,
'length': 1,
'out_ptype': 'wide',
'diagnostic_context': 'tool-value',
}
assert details.resolved_length == 1
assert details.resolved_jog is None
# The length-2 route exists, but normal cost ranking prefers length 5.
assert details.minimum_length == 5
assert details.minimum_attempted
assert not details.minimum_exhausted
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_attempted
assert details.minimum_length is None
assert details.minimum_exhausted
assert details.minimum_cause is not None
assert 'no legal route exists at any length' 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 not details.minimum_attempted
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 not details.minimum_attempted
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 not details.minimum_attempted
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 not details.minimum_attempted
assert tool.offer_calls == 0
@pytest.mark.parametrize(
'operation',
[
lambda p: p.trace([], None, length=1),
lambda p: p.trace_to([], None, length=1),
lambda p: p.jog([], 2, length=3),
lambda p: p.uturn([], 2, length=3),
],
ids=['trace', 'trace_to', 'jog', 'uturn'],
)
def test_pather_rejects_empty_route_selection_before_planning(operation: Any) -> None:
tool = RequestCountingTool()
p = Pather(Library(), tools=tool, render='deferred')
with pytest.raises(BuildError, match='at least one port'):
operation(p)
assert tool.offer_calls == 0
assert not p.ports
assert not p._paths
@pytest.mark.parametrize(
'operation',
[
lambda p: p.trace(['A', 'A'], None, each=1),
lambda p: p.trace_to(['A', 'A'], None, xmin=-10),
lambda p: p.jog(['A', 'A'], 2, length=3, spacing=1),
lambda p: p.uturn(['A', 'A'], 2, length=3, spacing=1),
lambda p: p.at(['A', 'A']).straight(1),
],
ids=['trace', 'trace_to', 'jog', 'uturn', 'port_pather'],
)
def test_pather_rejects_duplicate_route_selection_before_planning(operation: Any) -> None:
tool = RequestCountingTool()
p = Pather(
Library(),
ports={'A': Port((0, 0), rotation=0, ptype='wire')},
tools=tool,
render='deferred',
)
with pytest.raises(BuildError, match=r"duplicates: \['A'\]"):
operation(p)
assert tool.offer_calls == 0
assert numpy.allclose(p.ports['A'].offset, (0, 0))
assert not p._paths
def test_pather_jog_failed_two_bend_route_is_atomic() -> None:
lib = Library()
tool = PathTool(layer='M1', width=2, ptype='wire')
p = Pather(lib, tools=tool, render='immediate')
p.pattern.ports['A'] = Port((0, 0), rotation=0, ptype='wire')
with pytest.raises(BuildError, match='S-bend'):
with pytest.raises(RouteError, match='S-bend') as exc_info:
p.jog('A', 1.5, length=1.5)
assert exc_info.value.details.minimum_length == 2
assert exc_info.value.details.resolved_jog == 1.5
assert numpy.allclose(p.pattern.ports['A'].offset, (0, 0))
assert p.pattern.ports['A'].rotation == 0
assert len(p._paths['A']) == 0
@ -593,9 +884,12 @@ def test_pather_uturn_failed_two_bend_route_is_atomic() -> None:
p = Pather(lib, tools=tool)
p.pattern.ports['A'] = Port((0, 0), rotation=0, ptype='wire')
with pytest.raises(BuildError, match='U-turn'):
with pytest.raises(RouteError, match='U-turn') as exc_info:
p.uturn('A', 1.5, length=0)
assert exc_info.value.details.minimum_attempted
assert exc_info.value.details.minimum_length is None
assert numpy.allclose(p.pattern.ports['A'].offset, (0, 0))
assert p.pattern.ports['A'].rotation == 0
assert len(p._paths['A']) == 0

View file

@ -5,7 +5,7 @@ import numpy
from numpy import pi
from numpy.testing import assert_allclose, assert_equal
from masque import Pather, Library, Pattern, Port
from masque import Pather, Library, Pattern, Port, RouteError
from masque.builder import PathTool, PrimitiveOffer, StraightOffer
from masque.builder.planner import RoutingPlanner
from masque.error import BuildError, PortError
@ -136,12 +136,13 @@ def test_pather_dead_ports() -> None:
p = Pather(lib, ports={"in": Port((0, 0), 0)}, tools=tool)
p.set_dead()
with pytest.raises(RouteError, match='finite and nonnegative'):
p.straight("in", -10)
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)
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()
@ -394,9 +395,9 @@ def test_pather_dead_fallback_preserves_out_ptype() -> None:
p.pattern.ports['A'] = Port((0, 0), rotation=0, ptype='wire')
p.set_dead()
p.straight('A', -1000, out_ptype='other')
p.straight('A', 1000, out_ptype='other')
assert numpy.allclose(p.pattern.ports['A'].offset, (1000, 0))
assert numpy.allclose(p.pattern.ports['A'].offset, (-1000, 0))
assert p.pattern.ports['A'].ptype == 'other'
assert len(p._paths['A']) == 0

View file

@ -6,7 +6,7 @@ import numpy
from numpy import pi
from numpy.testing import assert_allclose
from ..builder import Pather
from ..builder import Pather, RouteError
from ..builder.tools import PathTool, RenderStep, StraightOffer, Tool
from ..error import BuildError
from ..library import Library
@ -122,9 +122,10 @@ def test_deferred_render_dead_ports() -> None:
rp = Pather(lib, ports={"in": Port((0, 0), 0)}, tools=tool, render='deferred')
rp.set_dead()
with pytest.raises(RouteError, match='finite and nonnegative'):
rp.straight("in", -10)
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