From 3a3cd854a5a136d01bd9215daa6dfd4bedad4172 Mon Sep 17 00:00:00 2001 From: Jan Petykiewicz Date: Sun, 12 Jul 2026 15:27:14 -0700 Subject: [PATCH] [Pather/Planner] Improve error reporting --- masque/__init__.py | 2 + masque/builder/__init__.py | 5 + masque/builder/error.py | 65 +++++ masque/builder/planner/planner.py | 330 +++++++++++++++++++++++-- masque/builder/utils.py | 67 +++-- masque/test/test_builder.py | 44 ++++ masque/test/test_pather_constraints.py | 302 +++++++++++++++++++++- masque/test/test_pather_core.py | 13 +- masque/test/test_pather_rendering.py | 7 +- 9 files changed, 777 insertions(+), 58 deletions(-) create mode 100644 masque/builder/error.py diff --git a/masque/__init__.py b/masque/__init__.py index b5dddde..6598e2f 100644 --- a/masque/__init__.py +++ b/masque/__init__.py @@ -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, diff --git a/masque/builder/__init__.py b/masque/builder/__init__.py index 5bd3914..76fd1f8 100644 --- a/masque/builder/__init__.py +++ b/masque/builder/__init__.py @@ -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, diff --git a/masque/builder/error.py b/masque/builder/error.py new file mode 100644 index 0000000..e8f00fd --- /dev/null +++ b/masque/builder/error.py @@ -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)) diff --git a/masque/builder/planner/planner.py b/masque/builder/planner/planner.py index 913758e..548d429 100644 --- a/masque/builder/planner/planner.py +++ b/masque/builder/planner/planner.py @@ -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 - raise + 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, diff --git a/masque/builder/utils.py b/masque/builder/utils.py index ca36fff..ceea550 100644 --- a/masque/builder/utils.py +++ b/masque/builder/utils.py @@ -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 diff --git a/masque/test/test_builder.py b/masque/test/test_builder.py index 53c0a8c..5251622 100644 --- a/masque/test/test_builder.py +++ b/masque/test/test_builder.py @@ -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) diff --git a/masque/test/test_pather_constraints.py b/masque/test/test_pather_constraints.py index fab1c68..f695acd 100644 --- a/masque/test/test_pather_constraints.py +++ b/masque/test/test_pather_constraints.py @@ -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 diff --git a/masque/test/test_pather_core.py b/masque/test/test_pather_core.py index ddf869b..cfc426b 100644 --- a/masque/test/test_pather_core.py +++ b/masque/test/test_pather_core.py @@ -5,7 +5,7 @@ import numpy from numpy import pi from numpy.testing import assert_allclose, assert_equal -from masque import Pather, Library, Pattern, Port +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() - p.straight("in", -10) + 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 diff --git a/masque/test/test_pather_rendering.py b/masque/test/test_pather_rendering.py index 186faf4..677fa12 100644 --- a/masque/test/test_pather_rendering.py +++ b/masque/test/test_pather_rendering.py @@ -6,7 +6,7 @@ import numpy from numpy import pi from numpy.testing import assert_allclose -from ..builder import Pather +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() - rp.straight("in", -10) + 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