masque/masque/builder/error.py

109 lines
3.8 KiB
Python

"""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, format_stacktrace
RouteOperation = Literal['trace', 'trace_to', 'jog', 'uturn']
class ToolContractError(BuildError):
"""A Tool returned data inconsistent with its routing contract."""
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 and call-site 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}')
stacktrace = format_stacktrace().rstrip()
if stacktrace:
lines.extend(('', 'Stack trace:', stacktrace))
super().__init__('\n'.join(lines))