1730 lines
68 KiB
Python
1730 lines
68 KiB
Python
"""
|
|
Primitive-offer route selection for `Pather`.
|
|
|
|
`RoutingPlanner` is the stateless boundary between `Pather` routing calls and
|
|
Tool primitive offers. `Pather` passes copied `RoutePortContext` snapshots here;
|
|
the planner returns `PreparedRouteResult` records that describe pending
|
|
mutations without applying them to the live Pattern.
|
|
|
|
Public routing modes and bounds are normalized by `bounds.py` into per-leg
|
|
`SolverRequest` values. `Solver` performs the pure search: it queries Tool
|
|
offers, enumerates bounded primitive compositions, inserts ptype adapters,
|
|
solves primitive parameters, and returns the ranked `Candidate`. A `RouteLeg`
|
|
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
|
|
is at the origin, travel is along +x, and positive jog is to the left. After a
|
|
candidate is selected, committed steps are transformed back into layout-space
|
|
using the copied starting port.
|
|
"""
|
|
from __future__ import annotations
|
|
|
|
# ruff: noqa: ANN401,PLR0912,PLR0913,PLR0915,TC001,TC002,TC003
|
|
|
|
from collections.abc import Iterable, Mapping, Sequence
|
|
from dataclasses import dataclass, replace
|
|
from itertools import combinations
|
|
from math import cos, isclose as math_isclose, sin
|
|
from typing import Any, Literal
|
|
|
|
import numpy
|
|
from numpy import pi
|
|
from numpy.typing import ArrayLike
|
|
|
|
from ...error import BuildError, PortError
|
|
from ...ports import Port
|
|
from ...utils import PTypeMatch, SupportsBool, ptype_match, ptypes_compatible
|
|
from ..tools import (
|
|
BendOffer,
|
|
PrimitiveKind,
|
|
PrimitiveOffer,
|
|
RenderStep,
|
|
SOffer,
|
|
StraightOffer,
|
|
Tool,
|
|
)
|
|
from ..error import (
|
|
MinimumStatus,
|
|
RouteError,
|
|
RouteFailureDetails,
|
|
RouteFailurePolicy,
|
|
RouteOperation,
|
|
)
|
|
from ..utils import ell
|
|
from . import bounds as planner_bounds
|
|
from .interface import (
|
|
PreparedRouteAction,
|
|
PreparedRouteResult,
|
|
RoutePlanningError,
|
|
RoutePortContext,
|
|
route_failure_policy,
|
|
)
|
|
|
|
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'):
|
|
return strategy
|
|
raise BuildError(f'Invalid route strategy {strategy!r}; expected straight_first or turn_first')
|
|
|
|
|
|
def is_close(a: float, b: float) -> bool:
|
|
"""Compare route-solver scalars with the planner tolerance."""
|
|
return math_isclose(float(a), float(b), rel_tol=1e-5, abs_tol=1e-8)
|
|
|
|
|
|
def clean_parameter(value: float) -> float:
|
|
"""Snap tiny solver noise in primitive parameters before domain checks."""
|
|
rounded = round(float(value))
|
|
if abs(float(value) - rounded) <= 1e-8:
|
|
return float(rounded)
|
|
if abs(float(value)) <= 1e-10:
|
|
return 0.0
|
|
return float(value)
|
|
|
|
|
|
def minimum_parameter(offer: PrimitiveOffer, route_name: str) -> float:
|
|
"""Return an offer's deterministic minimum legal parameter."""
|
|
lower, upper = offer.parameter_domain
|
|
if lower != upper and lower >= upper:
|
|
raise BuildError(f'{route_name} primitive has an invalid parameter domain {offer.parameter_domain}')
|
|
if not numpy.isfinite(lower):
|
|
raise BuildError(f'{route_name} primitive has no finite minimum parameter')
|
|
return offer.canonicalize_parameter(lower)
|
|
|
|
|
|
def minimum_nonzero_parameters(offer: PrimitiveOffer) -> tuple[float, ...]:
|
|
"""Return deterministic nonzero endpoint parameters near an offer domain edge."""
|
|
lower, upper = offer.parameter_domain
|
|
if lower == upper:
|
|
try:
|
|
value = offer.canonicalize_parameter(lower)
|
|
except BuildError:
|
|
return ()
|
|
return () if is_close(value, 0) else (value,)
|
|
|
|
candidates: list[float] = []
|
|
if lower > 0 and numpy.isfinite(lower):
|
|
candidates.append(float(lower))
|
|
if upper < 0 and numpy.isfinite(upper):
|
|
candidates.append(float(numpy.nextafter(upper, -numpy.inf)))
|
|
|
|
selected: list[float] = []
|
|
for value in candidates:
|
|
try:
|
|
parameter = offer.canonicalize_parameter(value)
|
|
except BuildError:
|
|
continue
|
|
if not is_close(parameter, 0) and not any(is_close(parameter, prev) for prev in selected):
|
|
selected.append(parameter)
|
|
return tuple(selected)
|
|
|
|
|
|
def adapter_s_parameter(offer: PrimitiveOffer, residual_jog: float) -> float:
|
|
"""Choose a deterministic S-adapter parameter, preferring residual-jog direction."""
|
|
candidates = minimum_nonzero_parameters(offer)
|
|
if not candidates:
|
|
raise BuildError('S adapter has no finite deterministic parameter')
|
|
|
|
residual_sign = numpy.sign(residual_jog)
|
|
|
|
def key(item: tuple[int, float]) -> tuple[float, int, int]:
|
|
index, value = item
|
|
sign = numpy.sign(value)
|
|
sign_rank = 0 if is_close(residual_jog, 0) or sign == residual_sign else 1
|
|
return round(abs(value), 9), sign_rank, index
|
|
|
|
return min(enumerate(candidates), key=key)[1]
|
|
|
|
|
|
def is_adapter_offer(offer: PrimitiveOffer) -> bool:
|
|
"""Return true for straight/S offers that intentionally change concrete ptype."""
|
|
return (
|
|
isinstance(offer, StraightOffer | SOffer)
|
|
and ptype_match(offer.in_ptype, offer.in_ptype) is PTypeMatch.EXACT
|
|
and ptype_match(offer.out_ptype, offer.out_ptype) is PTypeMatch.EXACT
|
|
and ptype_match(offer.in_ptype, offer.out_ptype) is PTypeMatch.MISMATCH
|
|
)
|
|
|
|
|
|
def raise_if_fatal(err: Exception) -> None:
|
|
"""Propagate fatal planning errors while allowing normal candidate rejection."""
|
|
if route_failure_policy(err) is RouteFailurePolicy.FATAL:
|
|
raise err
|
|
|
|
|
|
def solve_small_lstsq(
|
|
matrix: Sequence[Sequence[float]],
|
|
residual: Sequence[float],
|
|
) -> tuple[float, ...] | None:
|
|
"""
|
|
Solve tiny least-squares systems without always paying NumPy setup cost.
|
|
|
|
Route parameter solving only uses one or two constraints and one or two
|
|
adjustable primitive parameters. Closed forms keep common cases simple;
|
|
NumPy remains the fallback for degenerate or future larger systems.
|
|
"""
|
|
rows = len(matrix)
|
|
cols = len(matrix[0]) if rows else 0
|
|
if rows == 1 and cols == 1:
|
|
a = matrix[0][0]
|
|
return None if a == 0 else (residual[0] / a,)
|
|
if rows == 1 and cols == 2:
|
|
a, b = matrix[0]
|
|
denom = a * a + b * b
|
|
return None if denom == 0 else (residual[0] * a / denom, residual[0] * b / denom)
|
|
if rows == 2 and cols == 1:
|
|
a = matrix[0][0]
|
|
b = matrix[1][0]
|
|
denom = a * a + b * b
|
|
return None if denom == 0 else ((a * residual[0] + b * residual[1]) / denom,)
|
|
if rows == 2 and cols == 2:
|
|
a, b = matrix[0]
|
|
c, d = matrix[1]
|
|
determinant = a * d - b * c
|
|
if determinant != 0:
|
|
return (
|
|
(d * residual[0] - b * residual[1]) / determinant,
|
|
(-c * residual[0] + a * residual[1]) / determinant,
|
|
)
|
|
matrix_array = numpy.array(matrix)
|
|
deltas, _residuals, _rank, _singular = numpy.linalg.lstsq(matrix_array, numpy.array(residual), rcond=None)
|
|
return tuple(float(delta) for delta in deltas)
|
|
|
|
|
|
@dataclass(frozen=True, slots=True)
|
|
class SelectedPrimitive:
|
|
"""
|
|
One evaluated primitive offer in a candidate route.
|
|
|
|
`out_port` is still in route-local coordinates. `role` distinguishes
|
|
primitives that satisfy the requested route shape from ptype adapters that
|
|
the grammar may insert around those primitives.
|
|
"""
|
|
offer: PrimitiveOffer
|
|
"""Offer selected for this primitive step."""
|
|
parameter: float
|
|
"""Canonicalized offer parameter used for endpoint, cost, and commit."""
|
|
out_port: Port
|
|
"""Route-local endpoint produced by the selected offer."""
|
|
cost: float
|
|
"""Finite additive planning cost reported by the offer."""
|
|
role: Literal['main', 'adapter'] = 'main'
|
|
"""Whether this step satisfies route geometry or adapts ptype."""
|
|
route_kind: PrimitiveKind | None = None
|
|
"""Primitive kind used when querying the Tool for this step."""
|
|
|
|
|
|
@dataclass(frozen=True, slots=True)
|
|
class Candidate:
|
|
"""A fully solved primitive sequence with its composed local endpoint."""
|
|
steps: tuple[SelectedPrimitive, ...]
|
|
"""Ordered primitive sequence selected by the grammar."""
|
|
end_port: Port
|
|
"""Composed route-local endpoint."""
|
|
cost: float
|
|
"""Sum of primitive costs."""
|
|
order: int
|
|
"""Deterministic discovery order used as the final tie-breaker."""
|
|
public_length: float
|
|
"""Length reported back to bundle planning for omitted-length anchors."""
|
|
|
|
|
|
@dataclass(frozen=True, slots=True)
|
|
class SolverRequest:
|
|
"""
|
|
Normalized input for one `Solver` invocation and one route leg.
|
|
|
|
Public Pather calls are converted into this smaller shape before grammar
|
|
enumeration. `length`, `jog`, and `out_ptype` become endpoint constraints;
|
|
`tool_options` are forwarded to Tool primitive-offer generation.
|
|
"""
|
|
family: PrimitiveKind
|
|
"""High-level route family being solved."""
|
|
tool: Tool
|
|
"""Tool queried for primitive offers."""
|
|
in_ptype: str | None
|
|
"""Input ptype at the start of the route."""
|
|
tool_options: Mapping[str, Any]
|
|
"""Custom Tool kwargs forwarded to primitive-offer generation."""
|
|
length: float | None = None
|
|
"""Requested local x displacement, when constrained."""
|
|
jog: float | None = None
|
|
"""Requested local y displacement, when constrained."""
|
|
ccw: SupportsBool | None = None
|
|
"""Requested bend direction for single-bend routes."""
|
|
out_ptype: str | None = None
|
|
"""Requested final endpoint ptype."""
|
|
constrain_jog: bool = False
|
|
"""Whether bend-family trace_into routes must also match `jog`."""
|
|
max_bends: int | None = None
|
|
"""Optional override for grammar bend budget."""
|
|
strategy: RouteTieBreakStrategy = 'straight_first'
|
|
"""Discovery-order tie-break strategy for straight-vs-turn placement."""
|
|
|
|
@property
|
|
def route_name(self) -> str:
|
|
if self.family in ('straight', 'bend'):
|
|
return 'trace'
|
|
if self.family == 's':
|
|
return 'S-bend'
|
|
return 'U-turn'
|
|
|
|
@property
|
|
def out_rotation(self) -> float:
|
|
if self.family == 'straight':
|
|
return pi
|
|
if self.family == 'bend':
|
|
return -pi / 2 if bool(self.ccw) else pi / 2
|
|
if self.family == 's':
|
|
return pi
|
|
return 0.0
|
|
|
|
@property
|
|
def bend_budget(self) -> int:
|
|
if self.max_bends is not None:
|
|
return self.max_bends
|
|
if self.family == 'straight':
|
|
return 0
|
|
if self.family == 'bend':
|
|
return 1
|
|
return 2
|
|
|
|
|
|
class Solver:
|
|
"""
|
|
Bounded grammar solver for composed primitive routes.
|
|
|
|
The grammar is `A? (N A? (B|S|U) A?)* N A?`, where `A` is a ptype adapter,
|
|
`N` is a normal straight-like primitive, and the middle term is either a
|
|
bend primitive, a Tool-provided S/U primitive, or a composed S/U route made
|
|
from bend primitives. Parameter solving happens after a sequence is
|
|
enumerated so fixed and adjustable offers share the same path.
|
|
"""
|
|
|
|
def __init__(self, request: SolverRequest) -> None:
|
|
self.request = request
|
|
self.eval_cache: dict[tuple[int, float, str | None, str | None, str, str, PrimitiveKind | None], SelectedPrimitive] = {}
|
|
self.offer_cache: dict[
|
|
tuple[PrimitiveKind, str | None, str | None, tuple[tuple[str, Any], ...]],
|
|
tuple[PrimitiveOffer, ...],
|
|
] = {}
|
|
self.seen_candidate_keys: set[tuple[Any, ...]] = set()
|
|
self.order = 0
|
|
|
|
@staticmethod
|
|
def route_bend_count(steps: Sequence[SelectedPrimitive]) -> int:
|
|
"""Return the route bend budget consumed by non-adapter primitives."""
|
|
count = 0
|
|
for step in steps:
|
|
if step.role == 'adapter':
|
|
continue
|
|
if step.route_kind == 'bend':
|
|
count += 1
|
|
elif step.route_kind in ('s', 'u'):
|
|
count += 2
|
|
return count
|
|
|
|
def candidate_key(self, candidate: Candidate) -> tuple[Any, ...]:
|
|
"""Return a deterministic key for duplicate solved candidates."""
|
|
def endpoint_key(port: Port) -> tuple[float, float, float | None, str | None]:
|
|
return (
|
|
round(float(port.x), 9),
|
|
round(float(port.y), 9),
|
|
None if port.rotation is None else round(float(port.rotation), 9),
|
|
port.ptype,
|
|
)
|
|
|
|
def offer_key(offer: PrimitiveOffer) -> tuple[Any, ...]:
|
|
return (
|
|
type(offer).__qualname__,
|
|
offer.in_ptype,
|
|
offer.out_ptype,
|
|
round(float(offer.priority_bias), 9),
|
|
tuple(round(float(value), 9) for value in offer.parameter_domain),
|
|
getattr(offer, 'ccw', None),
|
|
id(offer.endpoint_planner),
|
|
id(offer.commit_planner),
|
|
)
|
|
|
|
return (
|
|
endpoint_key(candidate.end_port),
|
|
tuple((
|
|
offer_key(step.offer),
|
|
step.role,
|
|
step.route_kind,
|
|
round(float(step.parameter), 9),
|
|
endpoint_key(step.out_port),
|
|
) for step in candidate.steps),
|
|
)
|
|
|
|
def solve(
|
|
self,
|
|
*,
|
|
min_bends: int = 0,
|
|
max_bends: int | None = None,
|
|
) -> Candidate:
|
|
"""
|
|
Enumerate, finalize, deduplicate, and rank legal candidates.
|
|
|
|
Non-fatal candidate errors are accumulated so the failure message can
|
|
preserve useful Tool feedback. Fatal offer-contract errors stop the
|
|
solve immediately.
|
|
"""
|
|
if max_bends is None:
|
|
max_bends = self.request.bend_budget
|
|
candidates: list[Candidate] = []
|
|
errors: list[Exception] = []
|
|
for steps in self.enumerate_grammar(max_bends):
|
|
if not steps:
|
|
continue
|
|
if self.route_bend_count(steps) < min_bends:
|
|
continue
|
|
if any(first.role == 'adapter' and second.role == 'adapter' for first, second in zip(steps, steps[1:], strict=False)):
|
|
continue
|
|
try:
|
|
candidate = self.finalize(steps)
|
|
except (BuildError, NotImplementedError, PortError) as err:
|
|
raise_if_fatal(err)
|
|
errors.append(err)
|
|
continue
|
|
key = self.candidate_key(candidate)
|
|
if key in self.seen_candidate_keys:
|
|
continue
|
|
self.seen_candidate_keys.add(key)
|
|
candidates.append(candidate)
|
|
|
|
if not candidates:
|
|
for err in errors:
|
|
if route_failure_policy(err) is RouteFailurePolicy.FATAL:
|
|
raise err
|
|
if errors:
|
|
last_error = errors[-1]
|
|
if self.request.route_name in str(last_error):
|
|
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,
|
|
key=lambda candidate: (
|
|
round(float(candidate.cost), 9),
|
|
sum(step.role == 'adapter' for step in candidate.steps),
|
|
len(candidate.steps),
|
|
candidate.order,
|
|
),
|
|
)
|
|
|
|
def primitive_offers(
|
|
self,
|
|
kind: PrimitiveKind,
|
|
in_ptype: str | None,
|
|
*,
|
|
out_ptype: str | None = None,
|
|
extra: Mapping[str, Any] | None = None,
|
|
) -> tuple[PrimitiveOffer, ...]:
|
|
"""Query the active Tool with custom planning options and internal overrides."""
|
|
kwargs = dict(self.request.tool_options)
|
|
if extra:
|
|
kwargs.update(extra)
|
|
try:
|
|
cache_key = (kind, in_ptype, out_ptype, tuple(sorted(kwargs.items())))
|
|
hash(cache_key)
|
|
except TypeError:
|
|
return self.request.tool.primitive_offers(kind, in_ptype=in_ptype, out_ptype=out_ptype, **kwargs)
|
|
|
|
cached = self.offer_cache.get(cache_key)
|
|
if cached is not None:
|
|
return cached
|
|
offers = self.request.tool.primitive_offers(kind, in_ptype=in_ptype, out_ptype=out_ptype, **kwargs)
|
|
self.offer_cache[cache_key] = offers
|
|
return offers
|
|
|
|
def evaluate(
|
|
self,
|
|
offer: PrimitiveOffer,
|
|
parameter: float,
|
|
in_ptype: str | None,
|
|
*,
|
|
out_ptype: str | None,
|
|
role: Literal['main', 'adapter'],
|
|
route_kind: PrimitiveKind | None,
|
|
route_name: str | None = None,
|
|
) -> SelectedPrimitive:
|
|
"""
|
|
Canonicalize and validate one offer evaluation.
|
|
|
|
This is the single point where the solver checks ptype compatibility,
|
|
endpoint declarations, selected endpoint ptype, finite cost, and
|
|
zero-jog S rejection.
|
|
"""
|
|
route_name = self.request.route_name if route_name is None else route_name
|
|
selected = offer.canonicalize_parameter(clean_parameter(parameter))
|
|
key = (id(offer), round(float(selected), 12), in_ptype, out_ptype, role, route_name, route_kind)
|
|
cached = self.eval_cache.get(key)
|
|
if cached is not None:
|
|
return cached
|
|
|
|
if not ptypes_compatible(in_ptype, offer.in_ptype):
|
|
raise BuildError('primitive input ptype is incompatible')
|
|
if isinstance(offer, SOffer) and is_close(selected, 0):
|
|
raise BuildError('zero-jog S primitive candidates are not allowed')
|
|
out_port = offer.endpoint_at(selected)
|
|
if not ptypes_compatible(out_port.ptype, offer.out_ptype):
|
|
raise RoutePlanningError(
|
|
f'{route_name} primitive endpoint ptype does not match declared offer out_ptype',
|
|
policy=RouteFailurePolicy.FATAL,
|
|
)
|
|
if out_ptype is not None and not ptypes_compatible(out_port.ptype, out_ptype):
|
|
raise RoutePlanningError(
|
|
'Requested out_ptype does not match primitive endpoint ptype',
|
|
policy=RouteFailurePolicy.FATAL,
|
|
)
|
|
cost = float(offer.cost_at(selected))
|
|
if not numpy.isfinite(cost):
|
|
raise BuildError(f'{route_name} primitive returned non-finite cost')
|
|
if cost < 0:
|
|
raise BuildError(f'{route_name} primitive returned negative cost')
|
|
primitive = SelectedPrimitive(
|
|
offer,
|
|
selected,
|
|
out_port,
|
|
cost,
|
|
role=role,
|
|
route_kind=route_kind,
|
|
)
|
|
self.eval_cache[key] = primitive
|
|
return primitive
|
|
|
|
def compose_endpoint(self, steps: Sequence[SelectedPrimitive]) -> Port:
|
|
"""
|
|
Compose local primitive endpoints into one local route endpoint.
|
|
|
|
Primitive output rotations follow Masque's port convention: the port
|
|
points back into the primitive, so each step advances orientation by
|
|
the primitive output rotation plus pi.
|
|
"""
|
|
x = 0.0
|
|
y = 0.0
|
|
angle = 0.0
|
|
ptype: str | None = None
|
|
for step in steps:
|
|
out_port = step.out_port
|
|
if out_port.rotation is None:
|
|
raise BuildError('Primitive endpoints must have rotation')
|
|
angle_cos = cos(angle)
|
|
angle_sin = sin(angle)
|
|
x += angle_cos * float(out_port.x) - angle_sin * float(out_port.y)
|
|
y += angle_sin * float(out_port.x) + angle_cos * float(out_port.y)
|
|
angle += out_port.rotation + pi
|
|
ptype = out_port.ptype
|
|
return Port((x, y), rotation=angle - pi, ptype=ptype)
|
|
|
|
def current_ptype(self, steps: Sequence[SelectedPrimitive]) -> str | None:
|
|
return self.request.in_ptype if not steps else self.compose_endpoint(steps).ptype
|
|
|
|
def adapter_options(
|
|
self,
|
|
steps: Sequence[SelectedPrimitive],
|
|
*,
|
|
residual_jog: float,
|
|
) -> tuple[tuple[SelectedPrimitive, ...], ...]:
|
|
"""Return no-adapter plus single straight/S ptype adapter options."""
|
|
current_ptype = self.current_ptype(steps)
|
|
options: list[tuple[SelectedPrimitive, ...]] = [()]
|
|
for kind in ('straight', 's'):
|
|
try:
|
|
offers = self.primitive_offers(kind, current_ptype, out_ptype=None)
|
|
except NotImplementedError:
|
|
continue
|
|
for offer in offers:
|
|
if not is_adapter_offer(offer):
|
|
continue
|
|
try:
|
|
parameter = (
|
|
minimum_parameter(offer, 'straight adapter')
|
|
if kind == 'straight'
|
|
else adapter_s_parameter(offer, residual_jog)
|
|
)
|
|
selected = self.evaluate(
|
|
offer,
|
|
parameter,
|
|
current_ptype,
|
|
out_ptype=None,
|
|
role='adapter',
|
|
route_kind=kind,
|
|
route_name=f'{kind} adapter',
|
|
)
|
|
except BuildError as err:
|
|
raise_if_fatal(err)
|
|
continue
|
|
except NotImplementedError:
|
|
continue
|
|
options.append((selected,))
|
|
return tuple(options)
|
|
|
|
def straight_options(
|
|
self,
|
|
steps: Sequence[SelectedPrimitive],
|
|
) -> tuple[tuple[SelectedPrimitive, ...], ...]:
|
|
"""Return no-straight plus minimum-parameter non-adapter straight options."""
|
|
current_ptype = self.current_ptype(steps)
|
|
options: list[tuple[SelectedPrimitive, ...]] = [()]
|
|
try:
|
|
offers = self.primitive_offers('straight', current_ptype, out_ptype=None)
|
|
except NotImplementedError:
|
|
return tuple(options)
|
|
for offer in offers:
|
|
if is_adapter_offer(offer):
|
|
continue
|
|
try:
|
|
parameter = minimum_parameter(offer, 'trace')
|
|
selected = self.evaluate(
|
|
offer,
|
|
parameter,
|
|
current_ptype,
|
|
out_ptype=None,
|
|
role='main',
|
|
route_kind='straight',
|
|
route_name='trace',
|
|
)
|
|
except BuildError as err:
|
|
raise_if_fatal(err)
|
|
continue
|
|
except NotImplementedError:
|
|
continue
|
|
options.append((selected,))
|
|
return tuple(options)
|
|
|
|
def bend_options(
|
|
self,
|
|
steps: Sequence[SelectedPrimitive],
|
|
ccw: SupportsBool,
|
|
) -> tuple[tuple[SelectedPrimitive, ...], ...]:
|
|
"""Return legal fixed-direction bend options for the current ptype."""
|
|
current_ptype = self.current_ptype(steps)
|
|
options: list[tuple[SelectedPrimitive, ...]] = []
|
|
try:
|
|
offers = self.primitive_offers('bend', current_ptype, out_ptype=None, extra={'ccw': ccw})
|
|
except NotImplementedError:
|
|
return ()
|
|
for offer in offers:
|
|
if not isinstance(offer, BendOffer):
|
|
continue
|
|
if bool(offer.ccw) != bool(ccw):
|
|
continue
|
|
try:
|
|
parameter = minimum_parameter(offer, 'trace')
|
|
selected = self.evaluate(
|
|
offer,
|
|
parameter,
|
|
current_ptype,
|
|
out_ptype=None,
|
|
role='main',
|
|
route_kind='bend',
|
|
route_name='trace',
|
|
)
|
|
except BuildError as err:
|
|
raise_if_fatal(err)
|
|
continue
|
|
except NotImplementedError:
|
|
continue
|
|
options.append((selected,))
|
|
return tuple(options)
|
|
|
|
def su_primitive_options(
|
|
self,
|
|
steps: Sequence[SelectedPrimitive],
|
|
kind: Literal['s', 'u'],
|
|
jog: float | None = None,
|
|
) -> tuple[tuple[SelectedPrimitive, ...], ...]:
|
|
"""Return Tool-provided S/U primitive options for candidate jogs."""
|
|
current_ptype = self.current_ptype(steps)
|
|
options: list[tuple[SelectedPrimitive, ...]] = []
|
|
try:
|
|
offers = self.primitive_offers(kind, current_ptype, out_ptype=None)
|
|
except NotImplementedError:
|
|
return ()
|
|
route_name = 'S-bend' if kind == 's' else 'U-turn'
|
|
for offer in offers:
|
|
if kind == 's' and not isinstance(offer, SOffer):
|
|
continue
|
|
for parameter in self.su_parameters(offer, kind, jog):
|
|
try:
|
|
selected = self.evaluate(
|
|
offer,
|
|
parameter,
|
|
current_ptype,
|
|
out_ptype=None,
|
|
role='main',
|
|
route_kind=kind,
|
|
route_name=route_name,
|
|
)
|
|
except BuildError as err:
|
|
raise_if_fatal(err)
|
|
continue
|
|
except NotImplementedError:
|
|
continue
|
|
options.append((selected,))
|
|
return tuple(options)
|
|
|
|
def su_parameters(
|
|
self,
|
|
offer: PrimitiveOffer,
|
|
kind: Literal['s', 'u'],
|
|
requested_jog: float | None,
|
|
) -> tuple[float, ...]:
|
|
"""Build deterministic jog-parameter probes for Tool-provided S/U offers."""
|
|
candidates: list[float] = []
|
|
if requested_jog is not None and (kind == 'u' or not is_close(requested_jog, 0)):
|
|
candidates.append(float(requested_jog))
|
|
if kind == 'u':
|
|
lower, _upper = offer.parameter_domain
|
|
if numpy.isfinite(lower):
|
|
candidates.append(float(lower))
|
|
else:
|
|
candidates.append(0.0)
|
|
candidates.extend(minimum_nonzero_parameters(offer))
|
|
|
|
selected: list[float] = []
|
|
for candidate in candidates:
|
|
try:
|
|
parameter = offer.canonicalize_parameter(candidate)
|
|
except BuildError:
|
|
continue
|
|
if kind == 's' and is_close(parameter, 0):
|
|
continue
|
|
if not any(is_close(parameter, existing) for existing in selected):
|
|
selected.append(parameter)
|
|
return tuple(selected)
|
|
|
|
def turn_options(
|
|
self,
|
|
steps: Sequence[SelectedPrimitive],
|
|
remaining_bends: int,
|
|
) -> tuple[tuple[tuple[SelectedPrimitive, ...], int], ...]:
|
|
"""Return bend-family options paired with their consumed bend budget."""
|
|
options: list[tuple[tuple[SelectedPrimitive, ...], int]] = []
|
|
if remaining_bends >= 1:
|
|
for ccw in (False, True):
|
|
options.extend((turn, 1) for turn in self.bend_options(steps, ccw))
|
|
if remaining_bends >= 2:
|
|
jog = self.request.jog
|
|
options.extend((turn, 2) for turn in self.su_primitive_options(steps, 's', jog))
|
|
options.extend((turn, 2) for turn in self.su_primitive_options(steps, 'u', jog))
|
|
return tuple(options)
|
|
|
|
def enumerate_grammar(self, max_bends: int) -> Iterable[tuple[SelectedPrimitive, ...]]:
|
|
"""Yield raw primitive sequences allowed by the bounded route grammar."""
|
|
residual_jog = 0.0 if self.request.jog is None else float(self.request.jog)
|
|
base: tuple[SelectedPrimitive, ...] = ()
|
|
for prefix_adapter in self.adapter_options(base, residual_jog=residual_jog):
|
|
prefix = (*base, *prefix_adapter)
|
|
yield from self.enumerate_segments(prefix, max_bends, residual_jog=residual_jog)
|
|
|
|
def enumerate_segments(
|
|
self,
|
|
steps: tuple[SelectedPrimitive, ...],
|
|
remaining_bends: int,
|
|
*,
|
|
residual_jog: float,
|
|
) -> Iterable[tuple[SelectedPrimitive, ...]]:
|
|
"""Recursively enumerate normal/adapter/turn blocks within the bend budget."""
|
|
straight_options = self.straight_options(steps)
|
|
if self.request.strategy == 'straight_first':
|
|
straight_options = (*straight_options[1:], straight_options[0])
|
|
for normal in straight_options:
|
|
after_normal = (*steps, *normal)
|
|
suffix_options = (
|
|
self.adapter_options(after_normal, residual_jog=0)
|
|
if self.request.out_ptype is not None
|
|
else ((),)
|
|
)
|
|
for suffix in suffix_options:
|
|
yield (*after_normal, *suffix)
|
|
|
|
if remaining_bends <= 0:
|
|
continue
|
|
for core_adapter in self.adapter_options(after_normal, residual_jog=residual_jog):
|
|
before_core = (*after_normal, *core_adapter)
|
|
for turn, bend_count in self.turn_options(before_core, remaining_bends):
|
|
after_turn = (*before_core, *turn)
|
|
for post_adapter in self.adapter_options(after_turn, residual_jog=residual_jog):
|
|
yield from self.enumerate_segments(
|
|
(*after_turn, *post_adapter),
|
|
remaining_bends - bend_count,
|
|
residual_jog=residual_jog,
|
|
)
|
|
|
|
def adjustable_indices(self, steps: Sequence[SelectedPrimitive]) -> tuple[int, ...]:
|
|
"""Return non-adapter primitive indices whose parameter can still move."""
|
|
adjustable: list[int] = []
|
|
for index, step in enumerate(steps):
|
|
if step.role == 'adapter':
|
|
continue
|
|
parameter = step.parameter
|
|
lower, upper = step.offer.parameter_domain
|
|
probes = [
|
|
parameter + max(1e-6, abs(parameter) * 1e-6),
|
|
parameter - max(1e-6, abs(parameter) * 1e-6),
|
|
]
|
|
if numpy.isfinite(upper):
|
|
probes.append(numpy.nextafter(upper, -numpy.inf))
|
|
if numpy.isfinite(lower):
|
|
probes.append(lower)
|
|
for probe in probes:
|
|
try:
|
|
step.offer.canonicalize_parameter(probe)
|
|
except BuildError:
|
|
continue
|
|
if abs(float(probe) - float(parameter)) > 1e-12:
|
|
adjustable.append(index)
|
|
break
|
|
return tuple(adjustable)
|
|
|
|
def reevaluate(self, steps: Sequence[SelectedPrimitive], parameters: Sequence[float]) -> tuple[SelectedPrimitive, ...]:
|
|
"""Re-evaluate a primitive sequence with new parameters and flowing ptypes."""
|
|
selected: list[SelectedPrimitive] = []
|
|
current_ptype = self.request.in_ptype
|
|
for step, parameter in zip(steps, parameters, strict=True):
|
|
selected_step = self.evaluate(
|
|
step.offer,
|
|
parameter,
|
|
current_ptype,
|
|
out_ptype=None,
|
|
role=step.role,
|
|
route_kind=step.route_kind,
|
|
)
|
|
selected.append(selected_step)
|
|
current_ptype = selected_step.out_port.ptype
|
|
return tuple(selected)
|
|
|
|
def solve_parameters(
|
|
self,
|
|
steps: Sequence[SelectedPrimitive],
|
|
solve_indices: Sequence[int],
|
|
constraints: Sequence[tuple[Literal['x', 'y'], float]],
|
|
) -> tuple[tuple[SelectedPrimitive, ...], Port] | None:
|
|
"""
|
|
Adjust selected primitive parameters to satisfy endpoint constraints.
|
|
|
|
The solver estimates each adjustable parameter's local linear effect by
|
|
probing the composed endpoint, solves the tiny least-squares system,
|
|
then re-evaluates with canonicalized parameters.
|
|
"""
|
|
parameters = [step.parameter for step in steps]
|
|
for _iteration in range(3):
|
|
selected_steps = self.reevaluate(steps, parameters)
|
|
base_end = self.compose_endpoint(selected_steps)
|
|
if not solve_indices:
|
|
return selected_steps, base_end
|
|
|
|
matrix = [[0.0 for _index in solve_indices] for _constraint in constraints]
|
|
for column, solve_index in enumerate(solve_indices):
|
|
step = steps[solve_index]
|
|
parameter = parameters[solve_index]
|
|
probe = parameter + max(1e-6, abs(parameter) * 1e-6)
|
|
try:
|
|
probe = step.offer.canonicalize_parameter(probe)
|
|
except BuildError:
|
|
probe = numpy.nextafter(parameter, -numpy.inf)
|
|
try:
|
|
probe = step.offer.canonicalize_parameter(probe)
|
|
except BuildError:
|
|
return None
|
|
if abs(float(probe) - float(parameter)) <= 1e-12:
|
|
return None
|
|
probe_parameters = list(parameters)
|
|
probe_parameters[solve_index] = probe
|
|
probe_steps = self.reevaluate(steps, probe_parameters)
|
|
probe_end = self.compose_endpoint(probe_steps)
|
|
for row, (axis, _target) in enumerate(constraints):
|
|
matrix[row][column] = (float(getattr(probe_end, axis)) - float(getattr(base_end, axis))) / (probe - parameter)
|
|
|
|
residual = [target - float(getattr(base_end, axis)) for axis, target in constraints]
|
|
if all(abs(value) <= 1e-9 for value in residual):
|
|
return selected_steps, base_end
|
|
deltas = solve_small_lstsq(matrix, residual)
|
|
if deltas is None:
|
|
return None
|
|
changed = False
|
|
for solve_index, delta in zip(solve_indices, deltas, strict=True):
|
|
parameter = steps[solve_index].offer.canonicalize_parameter(
|
|
clean_parameter(parameters[solve_index] + float(delta)),
|
|
)
|
|
changed = changed or abs(parameter - parameters[solve_index]) > 1e-12
|
|
parameters[solve_index] = parameter
|
|
if not changed:
|
|
return selected_steps, base_end
|
|
selected_steps = self.reevaluate(steps, parameters)
|
|
return selected_steps, self.compose_endpoint(selected_steps)
|
|
|
|
def endpoint_matches(
|
|
self,
|
|
end_port: Port,
|
|
constraints: Sequence[tuple[Literal['x', 'y'], float]],
|
|
) -> bool:
|
|
"""Return true when a composed endpoint satisfies requested position, rotation, and ptype."""
|
|
for axis, target in constraints:
|
|
if not is_close(getattr(end_port, axis), target):
|
|
return False
|
|
if end_port.rotation is None:
|
|
return False
|
|
rotation_delta = (float(end_port.rotation) - self.request.out_rotation) % (2 * pi)
|
|
if not (is_close(rotation_delta, 0) or is_close(rotation_delta, 2 * pi)):
|
|
return False
|
|
return self.request.out_ptype is None or ptypes_compatible(end_port.ptype, self.request.out_ptype)
|
|
|
|
def rotation_matches_request(self, steps: Sequence[SelectedPrimitive]) -> bool:
|
|
"""
|
|
Return true when a sequence can produce the requested output rotation.
|
|
|
|
Primitive parameters do not affect the rotation contract, so
|
|
rotation-impossible candidates can be rejected before parameter solving.
|
|
"""
|
|
angle = 0.0
|
|
for step in steps:
|
|
step_rotation = step.out_port.rotation
|
|
if step_rotation is None:
|
|
raise BuildError('Primitive endpoints must have rotation')
|
|
angle += float(step_rotation) + pi
|
|
rotation_delta = ((angle - pi) - self.request.out_rotation) % (2 * pi)
|
|
return is_close(rotation_delta, 0) or is_close(rotation_delta, 2 * pi)
|
|
|
|
def finalize(self, steps: Sequence[SelectedPrimitive]) -> Candidate:
|
|
"""
|
|
Try all small solve sets for one raw sequence and return the first match.
|
|
|
|
Solve-set order is deterministic and becomes part of the candidate
|
|
ordering only after cost and structural tie-breakers.
|
|
"""
|
|
constraints: list[tuple[Literal['x', 'y'], float]] = []
|
|
if self.request.length is not None:
|
|
constraints.append(('x', float(self.request.length)))
|
|
if self.request.family == 'straight':
|
|
constraints.append(('y', 0.0))
|
|
elif self.request.family in ('s', 'u') or self.request.constrain_jog:
|
|
if self.request.jog is None:
|
|
raise BuildError(f'{self.request.route_name} route requires a jog constraint')
|
|
constraints.append(('y', float(self.request.jog)))
|
|
route_constraints = tuple(constraints)
|
|
if not self.rotation_matches_request(steps):
|
|
raise BuildError(f'{self.request.route_name} composed primitive route is unsupported')
|
|
adjustable = self.adjustable_indices(steps)
|
|
solve_sets: list[tuple[int, ...]] = [()]
|
|
max_solve = min(len(route_constraints), len(adjustable))
|
|
for solve_size in range(1, max_solve + 1):
|
|
solve_sets.extend(combinations(adjustable, solve_size))
|
|
|
|
for solve_indices in solve_sets:
|
|
solved = self.solve_parameters(steps, solve_indices, route_constraints)
|
|
if solved is None:
|
|
continue
|
|
selected_steps, end_port = solved
|
|
if not self.endpoint_matches(end_port, route_constraints):
|
|
continue
|
|
order = self.order
|
|
self.order += 1
|
|
public_length = float(end_port.x) if self.request.length is None else float(self.request.length)
|
|
return Candidate(
|
|
tuple(selected_steps),
|
|
end_port,
|
|
sum(step.cost for step in selected_steps),
|
|
order,
|
|
public_length,
|
|
)
|
|
raise BuildError(f'{self.request.route_name} composed primitive route is unsupported')
|
|
|
|
|
|
@dataclass(frozen=True, slots=True)
|
|
class RouteLeg:
|
|
"""
|
|
One solved route leg tied to the Pather port it will update.
|
|
|
|
`start_port` is the copied route start used for all layout transforms.
|
|
`tool` is stored with the leg so prepared render steps cannot be stamped
|
|
with a mismatched Tool after bundle ordering.
|
|
"""
|
|
portspec: str
|
|
"""Pather port name this leg will update."""
|
|
start_port: Port
|
|
"""Copied layout-space route start."""
|
|
tool: Tool
|
|
"""Tool used for all render steps in this leg."""
|
|
candidate: Candidate
|
|
"""Solved local primitive candidate."""
|
|
plug_into: str | None = None
|
|
"""Optional destination port to consume after applying the final endpoint."""
|
|
|
|
|
|
class RoutingPlanner:
|
|
"""
|
|
Pather-facing stateless route-selection facade.
|
|
|
|
Public Pather methods call this class with copied port contexts. Returned
|
|
`PreparedRouteResult`s contain only committed render steps, final ports,
|
|
plug targets, and renames needed for Pather state mutation.
|
|
"""
|
|
|
|
TRACE_INTO_MAX_BENDS: int = 4
|
|
DEFAULT_STRATEGY: RouteTieBreakStrategy = 'straight_first'
|
|
|
|
def __init__(self, strategy: RouteTieBreakStrategy = DEFAULT_STRATEGY) -> None:
|
|
self.strategy = validate_strategy(strategy)
|
|
|
|
def resolve_strategy(self, strategy: RouteTieBreakStrategy | str | None) -> RouteTieBreakStrategy:
|
|
"""Return the per-route strategy or the planner default."""
|
|
if strategy is None:
|
|
return getattr(self, 'strategy', self.DEFAULT_STRATEGY)
|
|
return validate_strategy(strategy)
|
|
|
|
def trace_into_bend_bands(self, family: PrimitiveKind) -> tuple[tuple[int, int], ...]:
|
|
"""Return non-overlapping trace_into bend-budget bands for staged solving."""
|
|
max_bends = self.TRACE_INTO_MAX_BENDS
|
|
if family == 'bend':
|
|
return tuple(band for band in ((1, 1), (3, 3)) if band[1] <= max_bends)
|
|
bands: list[tuple[int, int]] = []
|
|
if max_bends >= 0:
|
|
bands.append((0, 2 if max_bends >= 2 else 0))
|
|
if max_bends >= 4:
|
|
bands.append((4, 4))
|
|
return tuple(bands)
|
|
|
|
def solver_request(
|
|
self,
|
|
family: PrimitiveKind,
|
|
context: RoutePortContext,
|
|
*,
|
|
length: float | None = None,
|
|
jog: float | None = None,
|
|
ccw: SupportsBool | None = None,
|
|
constrain_jog: bool = False,
|
|
max_bends: int | None = None,
|
|
strategy: RouteTieBreakStrategy | str | None = None,
|
|
tool_options: Mapping[str, Any] | None = None,
|
|
**kwargs: Any,
|
|
) -> SolverRequest:
|
|
"""Build normalized solver input for one route leg."""
|
|
return SolverRequest(
|
|
family=family,
|
|
tool=context.tool,
|
|
in_ptype=context.port.ptype,
|
|
tool_options={} if tool_options is None else dict(tool_options),
|
|
length=length,
|
|
jog=jog,
|
|
ccw=ccw,
|
|
out_ptype=kwargs.get('out_ptype'),
|
|
constrain_jog=constrain_jog,
|
|
max_bends=max_bends,
|
|
strategy=self.resolve_strategy(strategy),
|
|
)
|
|
|
|
def solver_for_request(self, request: SolverRequest) -> Solver:
|
|
"""Construct the solver for a route request."""
|
|
return Solver(request)
|
|
|
|
def route_leg_from_candidate(
|
|
self,
|
|
context: RoutePortContext,
|
|
candidate: Candidate,
|
|
*,
|
|
plug_into: str | None,
|
|
) -> RouteLeg:
|
|
"""Attach a solved candidate to its source Pather context."""
|
|
return RouteLeg(
|
|
portspec=context.portspec,
|
|
start_port=context.port.copy(),
|
|
tool=context.tool,
|
|
candidate=candidate,
|
|
plug_into=plug_into,
|
|
)
|
|
|
|
def plan_leg(
|
|
self,
|
|
family: PrimitiveKind,
|
|
context: RoutePortContext,
|
|
operation: RouteOperation | None = None,
|
|
diagnostic_request: Mapping[str, Any] | None = None,
|
|
/,
|
|
*,
|
|
length: float | None = None,
|
|
jog: float | None = None,
|
|
ccw: SupportsBool | None = None,
|
|
plug_into: str | None = None,
|
|
constrain_jog: bool = False,
|
|
max_bends: int | None = None,
|
|
strategy: RouteTieBreakStrategy | str | None = None,
|
|
tool_options: Mapping[str, Any] | None = None,
|
|
**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_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,
|
|
context=context,
|
|
length=length,
|
|
jog=jog,
|
|
ccw=ccw,
|
|
constrain_jog=constrain_jog,
|
|
max_bends=max_bends,
|
|
strategy=strategy,
|
|
tool_options=tool_options,
|
|
**kwargs,
|
|
)
|
|
try:
|
|
candidate = self.solver_for_request(request).solve()
|
|
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_status: MinimumStatus
|
|
if length is None:
|
|
minimum_cause = cause
|
|
minimum_status = MinimumStatus.NO_ROUTE
|
|
else:
|
|
minimum_request = replace(request, length=None)
|
|
try:
|
|
minimum_candidate = self.solver_for_request(minimum_request).solve()
|
|
minimum_length = minimum_candidate.public_length
|
|
minimum_status = MinimumStatus.FOUND
|
|
except NoLegalRouteError as minimum_err:
|
|
minimum_cause = str(minimum_err)
|
|
minimum_status = MinimumStatus.NO_ROUTE
|
|
except Exception as minimum_err:
|
|
if route_failure_policy(minimum_err) is RouteFailurePolicy.FATAL:
|
|
raise
|
|
minimum_cause = str(minimum_err)
|
|
minimum_status = MinimumStatus.FAILED
|
|
|
|
details = RouteFailureDetails(
|
|
operation=operation,
|
|
portspec=context.portspec,
|
|
in_ptype=context.port.ptype,
|
|
out_ptype=request.out_ptype,
|
|
request=diagnostic_request,
|
|
resolved_length=None if length is None else float(length),
|
|
resolved_jog=None if jog is None else float(jog),
|
|
minimum_length=minimum_length,
|
|
minimum_status=minimum_status,
|
|
cause=cause,
|
|
minimum_cause=minimum_cause,
|
|
)
|
|
raise RouteError(details) from err
|
|
return self.route_leg_from_candidate(context, candidate, plug_into=plug_into)
|
|
|
|
def prepared_route_action_from_leg(
|
|
self,
|
|
leg: RouteLeg,
|
|
) -> PreparedRouteAction:
|
|
"""
|
|
Convert a solved leg into committed render steps and a final live port.
|
|
|
|
Offer commits happen here, after route selection succeeds. Each
|
|
primitive endpoint is transformed from route-local coordinates using
|
|
the previous step's layout-space output port.
|
|
"""
|
|
current = leg.start_port.copy()
|
|
render_steps: list[RenderStep] = []
|
|
for selected in leg.candidate.steps:
|
|
port_rot = current.rotation
|
|
if port_rot is None:
|
|
raise PortError('Ports must have rotation')
|
|
out_port = selected.out_port.copy()
|
|
out_port.rotate_around((0, 0), pi + port_rot)
|
|
out_port.translate(current.offset)
|
|
render_steps.append(RenderStep(
|
|
selected.offer.opcode,
|
|
leg.tool,
|
|
current.copy(),
|
|
out_port.copy(),
|
|
selected.offer.commit(selected.parameter),
|
|
))
|
|
current = out_port
|
|
if not render_steps:
|
|
raise BuildError('Route leg has no primitive steps')
|
|
return PreparedRouteAction(
|
|
portspec=leg.portspec,
|
|
render_steps=tuple(render_steps),
|
|
final_port=current.copy(),
|
|
plug_into=leg.plug_into,
|
|
)
|
|
|
|
def prepared_result_from_legs(
|
|
self,
|
|
legs: Sequence[RouteLeg],
|
|
*,
|
|
renames: tuple[tuple[str, str], ...] = (),
|
|
) -> PreparedRouteResult:
|
|
"""Build a prepared result from solved legs plus deferred port renames."""
|
|
return PreparedRouteResult(
|
|
actions=tuple(self.prepared_route_action_from_leg(leg) for leg in legs),
|
|
renames=renames,
|
|
)
|
|
|
|
def plan_trace_route(
|
|
self,
|
|
contexts: Sequence[RoutePortContext],
|
|
ccw: SupportsBool | None,
|
|
length: float | None = None,
|
|
*,
|
|
spacing: float | ArrayLike | None = None,
|
|
strategy: RouteTieBreakStrategy | str | None = None,
|
|
tool_options: Mapping[str, Any] | None = None,
|
|
**bounds: Any,
|
|
) -> 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
|
|
if tool_options:
|
|
request_details['tool_options'] = dict(tool_options)
|
|
operation: RouteOperation = 'trace'
|
|
diagnostic_request = request_details
|
|
return self._plan_trace_route(
|
|
contexts,
|
|
ccw,
|
|
length,
|
|
operation,
|
|
diagnostic_request,
|
|
spacing=spacing,
|
|
strategy=strategy,
|
|
tool_options=tool_options,
|
|
**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,
|
|
tool_options: Mapping[str, Any] | 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],
|
|
operation,
|
|
diagnostic_request,
|
|
length=length,
|
|
ccw=ccw,
|
|
strategy=strategy,
|
|
tool_options=tool_options,
|
|
**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,
|
|
operation,
|
|
diagnostic_request,
|
|
length=each,
|
|
ccw=ccw,
|
|
strategy=strategy,
|
|
tool_options=tool_options,
|
|
**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],
|
|
operation,
|
|
diagnostic_request,
|
|
length=None,
|
|
ccw=ccw,
|
|
strategy=strategy,
|
|
tool_options=tool_options,
|
|
**route_bounds,
|
|
)
|
|
return self.prepared_result_from_legs((leg,))
|
|
|
|
bound_type = bundle_bounds[0]
|
|
bound_value = route_bounds.pop(bound_type)
|
|
set_rotation = route_bounds.pop('set_rotation', None)
|
|
extensions = ell(
|
|
{context.portspec: context.port for context in contexts},
|
|
ccw,
|
|
spacing=spacing,
|
|
bound=bound_value,
|
|
bound_type=bound_type,
|
|
set_rotation=set_rotation,
|
|
)
|
|
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,
|
|
operation,
|
|
diagnostic_request,
|
|
length=route_length,
|
|
ccw=ccw,
|
|
strategy=strategy,
|
|
tool_options=tool_options,
|
|
**route_bounds,
|
|
)
|
|
actions.append(self.prepared_route_action_from_leg(leg))
|
|
return PreparedRouteResult(tuple(actions))
|
|
|
|
def plan_trace_to_route(
|
|
self,
|
|
contexts: Sequence[RoutePortContext],
|
|
ccw: SupportsBool | None,
|
|
*,
|
|
spacing: float | ArrayLike | None = None,
|
|
strategy: RouteTieBreakStrategy | str | None = None,
|
|
tool_options: Mapping[str, Any] | None = None,
|
|
**bounds: Any,
|
|
) -> 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
|
|
if tool_options:
|
|
request_details['tool_options'] = dict(tool_options)
|
|
operation: RouteOperation = 'trace_to'
|
|
diagnostic_request = request_details
|
|
return self._plan_trace_to_route(
|
|
contexts,
|
|
ccw,
|
|
operation,
|
|
diagnostic_request,
|
|
spacing=spacing,
|
|
strategy=strategy,
|
|
tool_options=tool_options,
|
|
**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,
|
|
tool_options: Mapping[str, Any] | None,
|
|
**route_bounds: Any,
|
|
) -> PreparedRouteResult:
|
|
if len(contexts) == 1:
|
|
resolved = planner_bounds.resolved_position_bound(contexts[0].port, route_bounds, allow_length=False)
|
|
else:
|
|
resolved = None
|
|
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,
|
|
route_bounds.pop('length', None),
|
|
operation,
|
|
diagnostic_request,
|
|
spacing=spacing,
|
|
strategy=strategy,
|
|
tool_options=tool_options,
|
|
**route_bounds,
|
|
)
|
|
|
|
planner_bounds.validate_trace_to_positional_args(spacing=spacing, bounds=route_bounds)
|
|
_key, _value, length = resolved
|
|
other_bounds = {
|
|
key: value
|
|
for key, value in route_bounds.items()
|
|
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],
|
|
operation,
|
|
diagnostic_request,
|
|
length=length,
|
|
ccw=ccw,
|
|
strategy=strategy,
|
|
tool_options=tool_options,
|
|
**other_bounds,
|
|
)
|
|
return self.prepared_result_from_legs((leg,))
|
|
|
|
def plan_jog_route(
|
|
self,
|
|
contexts: Sequence[RoutePortContext],
|
|
offset: float,
|
|
length: float | None = None,
|
|
*,
|
|
spacing: float | ArrayLike | None = None,
|
|
strategy: RouteTieBreakStrategy | str | None = None,
|
|
tool_options: Mapping[str, Any] | None = None,
|
|
**bounds: Any,
|
|
) -> PreparedRouteResult:
|
|
"""Plan S-bend routes for single ports or spaced bundles."""
|
|
offset = planner_bounds.finite_scalar(offset, 'offset')
|
|
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
|
|
if tool_options:
|
|
request_details['tool_options'] = dict(tool_options)
|
|
operation: RouteOperation = 'jog'
|
|
diagnostic_request = request_details
|
|
if numpy.isclose(offset, 0):
|
|
return self._plan_trace_to_route(
|
|
contexts,
|
|
None,
|
|
operation,
|
|
diagnostic_request,
|
|
length=length,
|
|
spacing=spacing,
|
|
strategy=strategy,
|
|
tool_options=tool_options,
|
|
**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)
|
|
other_bounds = dict(route_bounds)
|
|
if length is None and len(contexts) == 1:
|
|
resolved = planner_bounds.resolved_position_bound(contexts[0].port, route_bounds, allow_length=True)
|
|
if resolved is not None:
|
|
_key, _value, length = resolved
|
|
other_bounds = {key: value for key, value in route_bounds.items() if key not in planner_bounds.POSITION_KEYS}
|
|
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,
|
|
operation,
|
|
diagnostic_request,
|
|
strategy=strategy,
|
|
tool_options=tool_options,
|
|
**other_bounds,
|
|
)
|
|
))
|
|
leg = self.plan_leg(
|
|
's',
|
|
contexts[0],
|
|
operation,
|
|
diagnostic_request,
|
|
length=length,
|
|
jog=offset,
|
|
strategy=strategy,
|
|
tool_options=tool_options,
|
|
**other_bounds,
|
|
)
|
|
return self.prepared_result_from_legs((leg,))
|
|
|
|
def plan_uturn_route(
|
|
self,
|
|
contexts: Sequence[RoutePortContext],
|
|
offset: float,
|
|
length: float | None = None,
|
|
*,
|
|
spacing: float | ArrayLike | None = None,
|
|
strategy: RouteTieBreakStrategy | str | None = None,
|
|
tool_options: Mapping[str, Any] | None = None,
|
|
**bounds: Any,
|
|
) -> PreparedRouteResult:
|
|
"""Plan U-turn routes for single ports or spaced bundles."""
|
|
offset = planner_bounds.finite_scalar(offset, 'offset')
|
|
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
|
|
if tool_options:
|
|
request_details['tool_options'] = dict(tool_options)
|
|
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,
|
|
operation,
|
|
diagnostic_request,
|
|
strategy=strategy,
|
|
tool_options=tool_options,
|
|
**route_bounds,
|
|
)
|
|
))
|
|
leg = self.plan_leg(
|
|
'u',
|
|
contexts[0],
|
|
operation,
|
|
diagnostic_request,
|
|
length=length,
|
|
jog=offset,
|
|
strategy=strategy,
|
|
tool_options=tool_options,
|
|
**route_bounds,
|
|
)
|
|
return self.prepared_result_from_legs((leg,))
|
|
|
|
def plan_su_bundle_routes(
|
|
self,
|
|
kind: Literal['s', 'u'],
|
|
contexts: Sequence[RoutePortContext],
|
|
offset: float,
|
|
length: float | None,
|
|
spacing: float | ArrayLike | None,
|
|
operation: RouteOperation,
|
|
diagnostic_request: Mapping[str, Any],
|
|
/,
|
|
*,
|
|
strategy: RouteTieBreakStrategy | str | None = None,
|
|
tool_options: Mapping[str, Any] | None = None,
|
|
**kwargs: Any,
|
|
) -> tuple[RouteLeg, ...]:
|
|
"""
|
|
Solve the anchor S/U route and derive exact routes for the rest of a bundle.
|
|
|
|
The anchor may determine the public length when omitted. Once known,
|
|
`su_bundle_specs()` normalizes every other port into an exact length
|
|
and offset so all legs can be planned independently.
|
|
"""
|
|
if len(contexts) == 1:
|
|
return (self.plan_leg(
|
|
kind,
|
|
contexts[0],
|
|
operation,
|
|
diagnostic_request,
|
|
length=length,
|
|
jog=offset,
|
|
strategy=strategy,
|
|
tool_options=tool_options,
|
|
**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],
|
|
operation,
|
|
diagnostic_request,
|
|
length=length,
|
|
jog=offset,
|
|
strategy=strategy,
|
|
tool_options=tool_options,
|
|
**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]
|
|
routes_by_name = {first_portspec: anchor}
|
|
for spec_portspec, spec_length, spec_offset in specs[1:]:
|
|
if kind == 's' and is_close(spec_offset, 0):
|
|
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,
|
|
tool_options=tool_options,
|
|
**kwargs,
|
|
)
|
|
else:
|
|
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,
|
|
tool_options=tool_options,
|
|
**kwargs,
|
|
)
|
|
return tuple(routes_by_name[spec_portspec] for spec_portspec, _length, _offset in specs)
|
|
|
|
def plan_trace_into(
|
|
self,
|
|
context_src: RoutePortContext,
|
|
portspec_dst: str,
|
|
port_dst: Port,
|
|
*,
|
|
out_ptype: str | None,
|
|
plug_destination: bool,
|
|
thru: str | None,
|
|
strategy: RouteTieBreakStrategy | str | None = None,
|
|
tool_options: Mapping[str, Any] | None = None,
|
|
) -> PreparedRouteResult:
|
|
"""Plan a bounded route from one source port into a destination port."""
|
|
if out_ptype is None:
|
|
out_ptype = port_dst.ptype
|
|
if context_src.port.rotation is None or port_dst.rotation is None:
|
|
raise PortError('Ports must have rotation')
|
|
desired = port_dst.copy()
|
|
desired.rotation = port_dst.rotation - pi
|
|
desired.ptype = out_ptype
|
|
family, length, jog, ccw = self.trace_into_spec(context_src.port, desired)
|
|
request = self.solver_request(
|
|
family,
|
|
context_src,
|
|
length=length,
|
|
jog=jog,
|
|
ccw=ccw,
|
|
constrain_jog=family == 'bend',
|
|
max_bends=self.TRACE_INTO_MAX_BENDS,
|
|
strategy=strategy,
|
|
tool_options=tool_options,
|
|
out_ptype=out_ptype,
|
|
)
|
|
solver = self.solver_for_request(request)
|
|
candidate = None
|
|
last_error: Exception | None = None
|
|
for min_bends, max_bends in self.trace_into_bend_bands(family):
|
|
try:
|
|
candidate = solver.solve(min_bends=min_bends, max_bends=max_bends)
|
|
break
|
|
except (BuildError, NotImplementedError) as err:
|
|
if route_failure_policy(err) is RouteFailurePolicy.FATAL:
|
|
raise
|
|
last_error = err
|
|
if candidate is None:
|
|
if last_error is not None:
|
|
raise last_error
|
|
raise BuildError('No legal primitive offer for trace_into route')
|
|
leg = self.route_leg_from_candidate(
|
|
context_src,
|
|
candidate,
|
|
plug_into=portspec_dst if plug_destination else None,
|
|
)
|
|
renames = ((thru, context_src.portspec),) if thru is not None else ()
|
|
return self.prepared_result_from_legs(
|
|
(leg,),
|
|
renames=renames,
|
|
)
|
|
|
|
def trace_into_spec(
|
|
self,
|
|
start_port: Port,
|
|
end_port: Port,
|
|
) -> tuple[PrimitiveKind, float, float, SupportsBool | None]:
|
|
"""Convert source/destination geometry into a primitive route family and constraints."""
|
|
def quarter_turn(rotation: float) -> int:
|
|
normalized = rotation % (2 * pi)
|
|
if is_close(normalized, 2 * pi):
|
|
normalized = 0.0
|
|
quarter = int(round(normalized / (pi / 2))) % 4
|
|
if not is_close(normalized, (quarter * pi / 2) % (2 * pi)):
|
|
raise BuildError('trace_into() only supports Manhattan port rotations')
|
|
return quarter
|
|
|
|
travel_jog, _angle = start_port.measure_travel(end_port)
|
|
travel, jog = travel_jog
|
|
length = -float(travel)
|
|
offset = -float(jog)
|
|
if start_port.rotation is None or end_port.rotation is None:
|
|
raise PortError('Ports must have rotation')
|
|
relative_quarter = (quarter_turn(end_port.rotation) - quarter_turn(start_port.rotation)) % 4
|
|
if relative_quarter == 0:
|
|
return ('straight', length, 0.0, None) if is_close(offset, 0) else ('s', length, offset, None)
|
|
if relative_quarter == 1:
|
|
return 'bend', length, offset, True
|
|
if relative_quarter == 2:
|
|
return 'u', length, offset, None
|
|
return 'bend', length, offset, False
|