masque/masque/builder/tools.py

1391 lines
49 KiB
Python

"""
Routing Tool contracts and built-in Tool implementations.
A `Tool` is the user-extensible side of Pather routing. It does not receive
Pather state and it does not choose high-level route topology. Instead,
`primitive_offers()` exposes the local primitive moves that are legal for the
Tool: parameter domains, endpoint ptypes and geometry, additive costs,
optional footprint bounds, and a commit hook for the selected parameter.
`masque.builder.planner` composes those offers into `trace()`, `jog()`,
`uturn()`, and `trace_into()` routes.
Offer geometry is described in local route coordinates. The current input port
is `(0, 0)` with rotation `0`; length-like parameters advance along +x; positive
jog is left of travel; returned endpoint ports describe the primitive output in
that same local frame. The planner transforms selected endpoints into layout
coordinates only after a complete route has been chosen.
Tool authors should treat offer planning callbacks as pure descriptions. The
solver may call `endpoint_at()`, `cost_at()`, and `bbox_at()` many times while
enumerating candidate compositions, ptype adapters, and parameter solutions.
Those callbacks should be deterministic and should not mutate a Library,
Pattern, or live Pather. `commit()` is the first selected-offer hook: it runs
only for primitives in the chosen route and returns the opaque value stored in
`RenderStep.data`.
`render()` is the geometry-construction hook. Pather calls it later with a
compatible batch of committed `RenderStep`s, already expressed in layout
coordinates and grouped by Tool/continuity. The returned single-top tree is
plugged into the pending route by Pather, which also validates that the rendered
output port matches the endpoint selected during planning.
Routing uses the Tool assigned to the routed input port. The solver does not
search across Tools or infer `Pather.retool()` boundaries; transitions,
cross-ptype routes, and adapter shapes must be exposed by the active Tool as
primitive offers.
"""
from typing import Literal, Any, Self
from collections import ChainMap
from collections.abc import Sequence, Callable, Mapping
from abc import ABC, abstractmethod
from dataclasses import dataclass, field
from math import isclose as scalar_isclose, isfinite as scalar_isfinite, isnan as scalar_isnan, sqrt
import numpy
from numpy.typing import NDArray
from numpy import pi
from ..utils import (
layer_t,
ptypes_compatible as ptypes_compatible,
rotation_matrix_2d,
)
from ..ports import Port
from ..pattern import Pattern
from ..abstract import Abstract
from ..library import ILibrary, Library, SINGLE_USE_PREFIX
from ..error import BuildError
from ..shapes import Path
def _canonicalize_domain_value(
value: float,
domain: tuple[float, float],
*,
rtol: float = 1e-9,
atol: float = 1e-12,
) -> float:
"""
Canonicalize a solved primitive parameter against a route domain.
Normal domains are half-open `[min, max)`. A `(value, value)` domain is a
special closed singleton used for fixed parameters.
"""
vv = float(value)
lower, upper = (float(domain[0]), float(domain[1]))
if scalar_isnan(lower) or scalar_isnan(upper):
raise BuildError(f'Parameter domain must not contain NaN values: {domain}')
if lower > upper:
raise BuildError(f'Parameter domain lower bound must not exceed upper bound: {domain}')
if not scalar_isfinite(vv):
raise BuildError(f'Parameter {vv:g} must be finite')
if lower == upper:
if not scalar_isfinite(lower):
raise BuildError(f'Singleton parameter domain must be finite: {domain}')
if scalar_isclose(vv, lower, rel_tol=rtol, abs_tol=atol):
return lower
raise BuildError(f'Parameter {vv:g} is outside singleton domain {{{lower:g}}}')
if scalar_isclose(vv, lower, rel_tol=rtol, abs_tol=atol):
vv = lower
if vv < lower or vv >= upper:
raise BuildError(f'Parameter {vv:g} is outside half-open domain [{lower:g}, {upper:g})')
return vv
EndpointCallable = Callable[[float], Port]
CommitCallable = Callable[[float], Any]
BBoxCallable = Callable[[float], NDArray[numpy.float64]]
DataCallable = Callable[[float], Any]
BBoxForDataCallable = Callable[[Any], NDArray[numpy.float64]]
PrimitiveKind = Literal['straight', 'bend', 's', 'u']
BUILTIN_PRIORITY_STEP = 1e7
def _generated_offer_callbacks(
endpoint_at: EndpointCallable,
data_at: DataCallable,
bbox_for_data: BBoxForDataCallable | None,
) -> tuple[EndpointCallable, CommitCallable, BBoxCallable | None]:
def commit(parameter: float) -> Any:
return data_at(parameter)
if bbox_for_data is None:
return endpoint_at, commit, None
def bbox(parameter: float) -> NDArray[numpy.float64]:
return bbox_for_data(data_at(parameter))
return endpoint_at, commit, bbox
def _prebuilt_offer_callbacks(
endpoint: Port,
data: Any,
bbox_for_data: BBoxForDataCallable | None,
) -> tuple[EndpointCallable, CommitCallable, BBoxCallable | None]:
prebuilt_endpoint = endpoint.copy()
def endpoint_at(parameter: float) -> Port:
_ = parameter
return prebuilt_endpoint.copy()
def commit(parameter: float) -> Any:
_ = parameter
return data
if bbox_for_data is None:
return endpoint_at, commit, None
def bbox(parameter: float) -> NDArray[numpy.float64]:
_ = parameter
return bbox_for_data(data)
return endpoint_at, commit, bbox
@dataclass(frozen=True, slots=True)
class PrimitiveOffer(ABC):
"""
Shared base type for local routing primitives made available by a `Tool`.
Custom tools normally construct one of the concrete offer classes:
`StraightOffer`, `BendOffer`, `SOffer`, or `UOffer`. `PrimitiveOffer`
exists to hold common callback, ptype, cost, and footprint behavior and is
useful for annotations when handling offers generically.
Offers are pure planning objects. `endpoint_at()` returns a local output
`Port`, `cost_at()` returns an additive scalar cost, `bbox_at()` returns
local primitive bounds when a footprint hook is available, and `commit()`
returns opaque render data only after an offer has been selected. These
methods should be deterministic and must not mutate the user's target
library.
Parameter domains are half-open `[min, max)` ranges, except `(value, value)`
is a closed singleton for fixed-size primitives. `None` and `"unk"` ptypes
are wildcards; incompatible concrete ptypes are rejected by `Pather`.
"""
in_ptype: str | None
out_ptype: str | None
priority_bias: float = 0.0
bbox_planner: BBoxCallable | None = None
parameterized_bbox: Any | None = None
endpoint_planner: EndpointCallable | None = None
commit_planner: CommitCallable | None = None
def __post_init__(self) -> None:
has_endpoint = self.endpoint_planner is not None
has_commit = self.commit_planner is not None
if has_endpoint != has_commit:
raise BuildError('PrimitiveOffer split callbacks require both endpoint_planner and commit_planner')
if not numpy.isfinite(self.priority_bias) or self.priority_bias < 0:
raise BuildError(f'PrimitiveOffer priority_bias must be nonnegative and finite, got {self.priority_bias:g}')
@property
@abstractmethod
def opcode(self) -> Literal['L', 'S', 'U']:
raise NotImplementedError
@property
@abstractmethod
def parameter_domain(self) -> tuple[float, float]:
raise NotImplementedError
def canonicalize_parameter(self, parameter: float) -> float:
"""Return a finite selected parameter inside this offer's domain."""
return _canonicalize_domain_value(parameter, self.parameter_domain)
def endpoint_at(self, parameter: float) -> Port:
"""
Evaluate the local endpoint for a candidate parameter.
The returned port is in Tool-local public route coordinates. It must
not depend on live Pather state or mutate the user's target library.
"""
selected = self.canonicalize_parameter(parameter)
if self.endpoint_planner is not None:
return self.endpoint_planner(selected)
raise NotImplementedError
def cost_at(self, parameter: float) -> float:
"""
Return this primitive's additive planning cost.
Lower cost is preferred before internal tie-breakers. The default cost
is based on local endpoint displacement plus `priority_bias`.
"""
selected = self.canonicalize_parameter(parameter)
if self.endpoint_planner is not None:
out_port = self.endpoint_planner(selected)
else:
out_port = self.endpoint_at(selected)
return self.priority_bias + abs(float(out_port.x)) + (pi / 2) * abs(float(out_port.y))
def bbox_at(self, parameter: float) -> NDArray[numpy.float64]:
"""
Return local primitive bounds for footprint-aware planning.
Tools may omit this hook by raising `NotImplementedError`; when present
it must return a finite `(2, 2)` min/max array in local coordinates.
"""
if self.bbox_planner is None:
raise NotImplementedError
bounds = numpy.asarray(
self.bbox_planner(self.canonicalize_parameter(parameter)),
dtype=float,
)
if bounds.shape != (2, 2):
raise BuildError(f'Primitive bbox must have shape (2, 2), got {bounds.shape}')
if not numpy.all(numpy.isfinite(bounds)):
raise BuildError('Primitive bbox must contain only finite values')
if numpy.any(bounds[0, :] > bounds[1, :]):
raise BuildError('Primitive bbox minimum corner must not exceed maximum corner')
return bounds
def commit(self, parameter: float) -> Any:
"""
Produce opaque render data for a selected primitive.
`Pather` calls this only for selected primitives while preparing
`RenderStep.data`. Unselected candidates are evaluated by endpoint/cost
only and should not need commit-side work.
"""
selected = self.canonicalize_parameter(parameter)
if self.commit_planner is not None:
return self.commit_planner(selected)
raise NotImplementedError
@dataclass(frozen=True, slots=True)
class StraightOffer(PrimitiveOffer):
"""Straight or straight-like primitive parameterized by public route length."""
length_domain: tuple[float, float] = (0.0, numpy.inf)
@classmethod
def generated(
cls,
ptype: str | None,
data_at: DataCallable,
*,
priority_bias: float = 0.0,
bbox_for_data: BBoxForDataCallable | None = None,
length_domain: tuple[float, float] = (0.0, numpy.inf),
) -> Self:
"""
Build a generated straight offer with the default route-frame endpoint.
"""
def endpoint_at(length: float) -> Port:
return Port((length, 0), rotation=pi, ptype=ptype)
endpoint_planner, commit_planner, bbox_planner = _generated_offer_callbacks(
endpoint_at,
data_at,
bbox_for_data,
)
return cls(
in_ptype = ptype,
out_ptype = ptype,
priority_bias = priority_bias,
bbox_planner = bbox_planner,
endpoint_planner = endpoint_planner,
commit_planner = commit_planner,
length_domain = length_domain,
)
@classmethod
def prebuilt(
cls,
in_ptype: str | None,
out_ptype: str | None,
endpoint: Port,
data: Any,
*,
priority_bias: float = 0.0,
bbox_for_data: BBoxForDataCallable | None = None,
) -> Self:
"""
Build a prebuilt straight-like offer backed by precomputed render data.
"""
endpoint_planner, commit_planner, bbox_planner = _prebuilt_offer_callbacks(
endpoint,
data,
bbox_for_data,
)
return cls(
in_ptype = in_ptype,
out_ptype = out_ptype,
priority_bias = priority_bias,
bbox_planner = bbox_planner,
endpoint_planner = endpoint_planner,
commit_planner = commit_planner,
length_domain = (float(endpoint.x), float(endpoint.x)),
)
@property
def opcode(self) -> Literal['L']:
return 'L'
@property
def parameter_domain(self) -> tuple[float, float]:
return self.length_domain
@dataclass(frozen=True, slots=True)
class BendOffer(PrimitiveOffer):
"""Single-turn L-route primitive parameterized by public route length."""
ccw: bool = True
length_domain: tuple[float, float] = (0.0, numpy.inf)
@classmethod
def generated(
cls,
ptype: str | None,
endpoint_at: EndpointCallable,
data_at: DataCallable,
*,
ccw: bool,
priority_bias: float = 0.0,
bbox_for_data: BBoxForDataCallable | None = None,
length_domain: tuple[float, float] = (0.0, numpy.inf),
) -> Self:
"""
Build a generated bend offer from endpoint and render-data callbacks.
"""
endpoint_planner, commit_planner, bbox_planner = _generated_offer_callbacks(
endpoint_at,
data_at,
bbox_for_data,
)
return cls(
in_ptype = ptype,
out_ptype = ptype,
priority_bias = priority_bias,
bbox_planner = bbox_planner,
endpoint_planner = endpoint_planner,
commit_planner = commit_planner,
ccw = ccw,
length_domain = length_domain,
)
@classmethod
def prebuilt(
cls,
in_ptype: str | None,
out_ptype: str | None,
endpoint: Port,
data: Any,
*,
ccw: bool,
priority_bias: float = 0.0,
bbox_for_data: BBoxForDataCallable | None = None,
) -> Self:
"""
Build a prebuilt bend offer backed by precomputed render data.
"""
endpoint_planner, commit_planner, bbox_planner = _prebuilt_offer_callbacks(
endpoint,
data,
bbox_for_data,
)
return cls(
in_ptype = in_ptype,
out_ptype = out_ptype,
priority_bias = priority_bias,
bbox_planner = bbox_planner,
endpoint_planner = endpoint_planner,
commit_planner = commit_planner,
ccw = ccw,
length_domain = (float(endpoint.x), float(endpoint.x)),
)
@property
def opcode(self) -> Literal['L']:
return 'L'
@property
def parameter_domain(self) -> tuple[float, float]:
return self.length_domain
@dataclass(frozen=True, slots=True)
class SOffer(PrimitiveOffer):
"""Non-turning S-route primitive parameterized by jog for a fixed route length."""
jog_domain: tuple[float, float] = (-numpy.inf, numpy.inf)
@classmethod
def generated(
cls,
ptype: str | None,
endpoint_at: EndpointCallable,
data_at: DataCallable,
*,
priority_bias: float = 0.0,
bbox_for_data: BBoxForDataCallable | None = None,
jog_domain: tuple[float, float] = (-numpy.inf, numpy.inf),
) -> Self:
"""
Build a generated S-like offer from endpoint and render-data callbacks.
"""
endpoint_planner, commit_planner, bbox_planner = _generated_offer_callbacks(
endpoint_at,
data_at,
bbox_for_data,
)
return cls(
in_ptype = ptype,
out_ptype = ptype,
priority_bias = priority_bias,
bbox_planner = bbox_planner,
endpoint_planner = endpoint_planner,
commit_planner = commit_planner,
jog_domain = jog_domain,
)
@classmethod
def prebuilt(
cls,
in_ptype: str | None,
out_ptype: str | None,
endpoint: Port,
data: Any,
*,
priority_bias: float = 0.0,
bbox_for_data: BBoxForDataCallable | None = None,
) -> Self:
"""
Build a prebuilt S-like offer backed by precomputed render data.
"""
endpoint_planner, commit_planner, bbox_planner = _prebuilt_offer_callbacks(
endpoint,
data,
bbox_for_data,
)
return cls(
in_ptype = in_ptype,
out_ptype = out_ptype,
priority_bias = priority_bias,
bbox_planner = bbox_planner,
endpoint_planner = endpoint_planner,
commit_planner = commit_planner,
jog_domain = (float(endpoint.y), float(endpoint.y)),
)
@property
def opcode(self) -> Literal['S']:
return 'S'
@property
def parameter_domain(self) -> tuple[float, float]:
return self.jog_domain
@dataclass(frozen=True, slots=True)
class UOffer(PrimitiveOffer):
"""U-turn-like primitive parameterized by jog for a fixed route length."""
jog_domain: tuple[float, float] = (-numpy.inf, numpy.inf)
@classmethod
def generated(
cls,
ptype: str | None,
endpoint_at: EndpointCallable,
data_at: DataCallable,
*,
priority_bias: float = 0.0,
bbox_for_data: BBoxForDataCallable | None = None,
jog_domain: tuple[float, float] = (-numpy.inf, numpy.inf),
) -> Self:
"""
Build a generated U-like offer from endpoint and render-data callbacks.
"""
endpoint_planner, commit_planner, bbox_planner = _generated_offer_callbacks(
endpoint_at,
data_at,
bbox_for_data,
)
return cls(
in_ptype = ptype,
out_ptype = ptype,
priority_bias = priority_bias,
bbox_planner = bbox_planner,
endpoint_planner = endpoint_planner,
commit_planner = commit_planner,
jog_domain = jog_domain,
)
@classmethod
def prebuilt(
cls,
in_ptype: str | None,
out_ptype: str | None,
endpoint: Port,
data: Any,
*,
priority_bias: float = 0.0,
bbox_for_data: BBoxForDataCallable | None = None,
) -> Self:
"""
Build a prebuilt U-like offer backed by precomputed render data.
"""
endpoint_planner, commit_planner, bbox_planner = _prebuilt_offer_callbacks(
endpoint,
data,
bbox_for_data,
)
return cls(
in_ptype = in_ptype,
out_ptype = out_ptype,
priority_bias = priority_bias,
bbox_planner = bbox_planner,
endpoint_planner = endpoint_planner,
commit_planner = commit_planner,
jog_domain = (float(endpoint.y), float(endpoint.y)),
)
@property
def opcode(self) -> Literal['U']:
return 'U'
@property
def parameter_domain(self) -> tuple[float, float]:
return self.jog_domain
@dataclass(frozen=True, slots=True)
class RenderStep:
"""
A single deferred routing operation.
`Pather(render='deferred')` stores these records while routing and later
passes batches of compatible steps to `Tool.render()` when `Pather.render()`
is called.
"""
opcode: Literal['L', 'S', 'U', 'P']
""" What operation is being performed.
L: straight or single-bend primitive
S: S-like primitive
U: U-like primitive
P: plug
"""
tool: 'Tool | None'
""" Tool that produced this step, or `None` for `opcode='P'`. """
start_port: Port
""" Input-side port before this step is rendered. """
end_port: Port
""" Output-side port after this step is rendered. """
data: Any
""" Arbitrary tool-specific data"""
def __post_init__(self) -> None:
if self.opcode != 'P' and self.tool is None:
raise BuildError('Got tool=None but the opcode is not "P"')
def is_continuous_with(self, other: 'RenderStep') -> bool:
"""
Check if another RenderStep can be appended to this one.
"""
# Check continuity with tolerance
offsets_match = bool(numpy.allclose(other.start_port.offset, self.end_port.offset))
rotations_match = (other.start_port.rotation is None and self.end_port.rotation is None) or (
other.start_port.rotation is not None and self.end_port.rotation is not None and
bool(numpy.isclose(other.start_port.rotation, self.end_port.rotation))
)
return offsets_match and rotations_match
def transformed(self, translation: NDArray[numpy.float64], rotation: float, pivot: NDArray[numpy.float64]) -> 'RenderStep':
"""
Return a new RenderStep with transformed start and end ports.
"""
new_start = self.start_port.copy()
new_end = self.end_port.copy()
for pp in (new_start, new_end):
pp.rotate_around(pivot, rotation)
pp.translate(translation)
return RenderStep(
opcode = self.opcode,
tool = self.tool,
start_port = new_start,
end_port = new_end,
data = self.data,
)
def mirrored(self, axis: int) -> 'RenderStep':
"""
Return a new RenderStep with mirrored start and end ports.
"""
new_start = self.start_port.copy()
new_end = self.end_port.copy()
new_start.flip_across(axis=axis)
new_end.flip_across(axis=axis)
return RenderStep(
opcode = self.opcode,
tool = self.tool,
start_port = new_start,
end_port = new_end,
data = self.data,
)
class Tool(ABC):
"""
Interface for path (e.g. wire or waveguide) generation.
Subclasses must override `primitive_offers()` and explicitly return `()`
for recognized primitive kinds they do not support.
Custom tools should return concrete offer objects (`StraightOffer`,
`BendOffer`, `SOffer`, or `UOffer`) rather than parsing offer identity from
strings after construction.
"""
@abstractmethod
def primitive_offers(
self,
kind: PrimitiveKind,
*,
in_ptype: str | None = None,
out_ptype: str | None = None,
**kwargs,
) -> tuple[PrimitiveOffer, ...]:
"""
Return local primitive offers available for the requested route role.
Tools override this to expose multiple legal primitive variants with explicit domains and costs.
Direct offer implementations should declare the actual endpoint ptype produced by the offer when it
can differ from the requested value.
`kind` is one of:
- `'straight'`: a non-turning `StraightOffer`
- `'bend'`: a 90-degree `BendOffer`; `ccw` is supplied in `kwargs`
- `'s'`: a non-turning `SOffer`
- `'u'`: an `UOffer`
`Pather` applies any requested `out_ptype` to the final route endpoint,
not to every primitive in the route. Intermediate ptypes are
solver-selected, and heterogeneous straight/S offers may be used as
adapters when legal.
"""
raise NotImplementedError
@abstractmethod
def render(
self,
batch: Sequence[RenderStep],
*,
port_names: tuple[str, str] = ('A', 'B'),
**kwargs,
) -> ILibrary:
"""
Render a compatible batch of selected route steps into geometry.
`Pather.render()` passes batches that share one Tool and are continuous
in layout coordinates. The returned tree must have one top cell whose
input and output ports are named by `port_names`; `Pather` plugs the
input port into the pending route start and validates the output port
against the planned final endpoint.
Args:
batch: A sequence of `RenderStep` objects containing committed
primitive render data.
port_names: The topcell's input and output ports should be named
`port_names[0]` and `port_names[1]` respectively.
kwargs: Custom tool-specific parameters.
"""
raise NotImplementedError
GeneratedPrimitiveFn = Callable[..., Pattern | Library]
GeneratedEndpointFn = Callable[[float], Port]
def circular_arc_sbend_endpoint(radius: float, ptype: str) -> GeneratedEndpointFn:
"""
Return an S-bend endpoint planner for two abutting circular arcs.
The returned callback assumes a pure generated S-bend made from two equal
circular arcs with no attached straight or non-circular pieces. Positive
and negative jogs are supported; the output rotation is always `pi`.
"""
rr = float(radius)
if not scalar_isfinite(rr) or rr <= 0:
raise BuildError(f'S-bend radius must be positive and finite, got {rr:g}')
def endpoint(jog: float) -> Port:
jj = float(jog)
jog_magnitude = abs(jj)
if scalar_isclose(jog_magnitude, 0.0, rel_tol=1e-9, abs_tol=1e-12):
return Port((0, 0), rotation=pi, ptype=ptype)
if jog_magnitude > 2 * rr and not scalar_isclose(jog_magnitude, 2 * rr, rel_tol=1e-9, abs_tol=1e-12):
raise BuildError(f'S-bend jog magnitude {jog_magnitude:g} exceeds diameter {2 * rr:g}')
dx = sqrt(max(0.0, 4 * rr * jog_magnitude - jog_magnitude ** 2))
return Port((dx, jj), rotation=pi, ptype=ptype)
return endpoint
@dataclass
class AutoTool(Tool):
"""
A routing tool assembled from reusable path primitives.
`AutoTool` chooses among prioritized straight generators, pre-rendered bends,
optional generated S-bend primitives, pre-rendered U-turns, and
pre-rendered transitions registered through `add_straight()`,
`add_bend()`, `add_sbend()`, `add_uturn()`, and `add_transition()`.
Registration call order defines primitive priority.
Straight and bend offers use one straight and, if turning, one bend.
`add_sbend()` exposes generated S-bend primitives. `add_uturn()` exposes
reusable U-turn primitives; otherwise U-turns are left to `Pather`'s
composed-route planning.
Transitions are bidirectional by default: `add_transition(external,
internal)` exposes adapter offers in both directions. Pass `one_way=True`
when only the declared direction should be available.
Straight and S-bend generator functions may return either a `Pattern` or a
single-top `Library`. Extra keyword arguments passed to `render()` are
forwarded to those generators.
"""
@dataclass(frozen=True, slots=True)
class GeneratedData:
""" Deferred render data for one generated primitive offer. """
fn: GeneratedPrimitiveFn
port_name: str
parameter: float
mirrored: bool = False
@dataclass(frozen=True, slots=True)
class ReusableData:
""" Deferred render data for one reusable abstract primitive offer. """
abstract: Abstract
port_name: str
mirrored: bool = False
bbox_library: Mapping[str, Pattern] | None = None
""" Optional source library used to resolve reusable refs during `bbox_at()` measurement. """
_straight_offers: list[PrimitiveOffer] = field(
default_factory = list,
init = False,
repr = False,
)
_bend_offers: tuple[list[PrimitiveOffer], list[PrimitiveOffer]] = field(
default_factory = lambda: ([], []),
init = False,
repr = False,
)
_s_offers: list[PrimitiveOffer] = field(
default_factory = list,
init = False,
repr = False,
)
_u_offers: list[PrimitiveOffer] = field(
default_factory = list,
init = False,
repr = False,
)
_transition_adapter_offers_by_key: dict[tuple[Literal['straight', 's'], str], list[PrimitiveOffer]] = field(
default_factory=dict,
init = False,
repr = False,
)
_transition_adapter_offer_keys: set[tuple[int, str, str]] = field(
default_factory=set,
init = False,
repr = False,
)
def add_straight(
self,
ptype: str,
fn: GeneratedPrimitiveFn,
in_port_name: str,
*,
length_range: tuple[float, float] = (0, numpy.inf),
) -> Self:
"""
Register a generated straight primitive.
"""
priority_bias = len(self._straight_offers) * BUILTIN_PRIORITY_STEP
def data_at(length: float) -> AutoTool.GeneratedData:
return self.GeneratedData(fn, in_port_name, length)
self._straight_offers.append(StraightOffer.generated(
ptype,
data_at,
priority_bias = priority_bias,
bbox_for_data = self._bbox_for_data,
length_domain = length_range,
))
return self
def add_bend(
self,
abstract: Abstract,
in_port_name: str,
out_port_name: str,
*,
clockwise: bool = True,
mirror: bool = True,
) -> Self:
"""
Register a reusable L-bend primitive.
"""
priority_bias = len(self._bend_offers[0]) * BUILTIN_PRIORITY_STEP
in_port = abstract.ports[in_port_name]
out_port = abstract.ports[out_port_name]
out_ptype = out_port.ptype
for ccw in (False, True):
bend_dxy, bend_angle = self._bend2dxy(in_port, out_port, ccw)
bend_dx = float(bend_dxy[0])
bend_dy = float(bend_dxy[1])
mirrored = mirror and (ccw == clockwise)
port_name = in_port_name if (mirror or ccw != clockwise) else out_port_name
reusable_data = self.ReusableData(abstract, port_name, mirrored)
endpoint = Port((bend_dx, bend_dy), rotation=bend_angle, ptype=out_ptype)
self._bend_offers[int(ccw)].append(BendOffer.prebuilt(
in_ptype = in_port.ptype,
out_ptype = out_ptype,
endpoint = endpoint,
data = reusable_data,
ccw = ccw,
priority_bias = priority_bias,
bbox_for_data = self._bbox_for_data,
))
return self
def add_sbend(
self,
ptype: str,
fn: GeneratedPrimitiveFn,
in_port_name: str,
out_port_name: str,
*,
jog_range: tuple[float, float] = (0, numpy.inf),
endpoint: GeneratedEndpointFn | None = None,
) -> Self:
"""
Register a generated S-bend primitive.
`endpoint`, when supplied, describes the generated S-bend output port
directly during planning and avoids instantiating `fn()` inside
`endpoint_at()`.
"""
if endpoint is None:
def endpoint_at(jog: float) -> Port:
jog_magnitude = abs(jog)
sbend_dxy = self._sbend2dxy(fn, in_port_name, out_port_name, jog_magnitude)
return Port((float(sbend_dxy[0]), float(jog)), rotation=pi, ptype=ptype)
else:
def endpoint_at(jog: float) -> Port:
out_port = endpoint(jog)
if not ptypes_compatible(out_port.ptype, ptype):
raise BuildError('S-bend endpoint ptype does not match registered ptype')
return out_port
for jog_domain in self._signed_jog_domains(jog_range):
priority_bias = len(self._s_offers) * BUILTIN_PRIORITY_STEP
def data_at(jog: float) -> AutoTool.GeneratedData:
return self.GeneratedData(
fn,
in_port_name,
abs(jog),
mirrored = jog < 0,
)
self._s_offers.append(SOffer.generated(
ptype,
endpoint_at,
data_at,
priority_bias = priority_bias,
bbox_for_data = self._bbox_for_data,
jog_domain = jog_domain,
))
return self
def add_uturn(
self,
abstract: Abstract,
in_port_name: str,
out_port_name: str,
*,
mirror: bool = True,
) -> Self:
"""
Register a reusable U-turn primitive.
"""
in_port = abstract.ports[in_port_name]
out_port = abstract.ports[out_port_name]
dxy, angle = in_port.measure_travel(out_port)
if angle is None:
raise BuildError('U-turn primitive output port must have the same route-frame rotation as its input port')
normalized_angle = angle % (2 * pi)
if not (numpy.isclose(normalized_angle, 0) or numpy.isclose(normalized_angle, 2 * pi)):
raise BuildError('U-turn primitive output port must have the same route-frame rotation as its input port')
length = float(dxy[0])
jog = float(dxy[1])
out_ptype = out_port.ptype
def add_offer(
offer_jog: float,
*,
mirrored: bool,
) -> None:
reusable_data = self.ReusableData(abstract, in_port_name, mirrored)
priority_bias = len(self._u_offers) * BUILTIN_PRIORITY_STEP
endpoint = Port((length, offer_jog), rotation=0, ptype=out_ptype)
self._u_offers.append(UOffer.prebuilt(
in_ptype = in_port.ptype,
out_ptype = out_ptype,
endpoint = endpoint,
data = reusable_data,
priority_bias = priority_bias,
bbox_for_data = self._bbox_for_data,
))
add_offer(jog, mirrored=False)
if mirror and not numpy.isclose(jog, 0):
add_offer(-jog, mirrored=True)
return self
def add_transition(
self,
abstract: Abstract,
their_port_name: str,
our_port_name: str,
*,
one_way: bool = False,
) -> Self:
"""
Register a reusable port-type transition and expose it as router-visible adapter offers.
"""
self._add_transition_direction(abstract, their_port_name, our_port_name)
if not one_way:
self._add_transition_direction(abstract, our_port_name, their_port_name)
return self
@staticmethod
def _bend2dxy(in_port: Port, out_port: Port, ccw: bool) -> tuple[NDArray[numpy.float64], float]:
bend_dxy, bend_angle = in_port.measure_travel(out_port)
assert bend_angle is not None
if ccw:
bend_dxy[1] *= -1
bend_angle *= -1
return bend_dxy, bend_angle
@staticmethod
def _wildcard_ptype_key(ptype: str | None) -> str:
return 'unk' if ptype in (None, 'unk') else ptype
@staticmethod
def _sbend2dxy(
fn: GeneratedPrimitiveFn,
in_port_name: str,
out_port_name: str,
jog_magnitude: float,
) -> NDArray[numpy.float64]:
if numpy.isclose(jog_magnitude, 0):
return numpy.zeros(2)
sbend_pat_or_tree = fn(jog_magnitude)
sbpat = sbend_pat_or_tree if isinstance(sbend_pat_or_tree, Pattern) else sbend_pat_or_tree.top_pattern()
dxy, _ = sbpat[in_port_name].measure_travel(sbpat[out_port_name])
return dxy
def _rendered_bbox(self, render: Callable[[ILibrary, tuple[str, str]], None]) -> NDArray[numpy.float64]:
port_names = ('A', 'B')
tree, pat = Library.mktree(SINGLE_USE_PREFIX + 'primitive_bbox')
pat.add_port_pair(names=port_names)
render(tree, port_names)
if self.bbox_library is None:
library: Mapping[str, Pattern] = tree
else:
library = ChainMap(dict(tree), self.bbox_library)
try:
bounds = pat.get_bounds(library=library)
except KeyError as err:
raise NotImplementedError(
'AutoTool bbox_at() requires bbox_library to resolve reusable primitive refs'
) from err
if bounds is None:
return numpy.zeros((2, 2), dtype=float)
return numpy.asarray(bounds, dtype=float)
def _bbox_for_data(self, data: Any) -> NDArray[numpy.float64]:
return self._rendered_bbox(lambda tree, names: self._render_data(data, tree, names, {}))
def _add_transition_direction(
self,
abstract: Abstract,
their_port_name: str,
our_port_name: str,
) -> None:
their_port = abstract.ports[their_port_name]
our_port = abstract.ports[our_port_name]
transition_data = self.ReusableData(abstract, their_port_name)
key = (
id(abstract),
their_port_name,
our_port_name,
)
if key in self._transition_adapter_offer_keys:
return
self._transition_adapter_offer_keys.add(key)
dxy, angle = their_port.measure_travel(our_port)
if angle is None or not numpy.isclose(angle, pi):
return
dx = float(dxy[0])
dy = float(dxy[1])
kind: Literal['straight', 's'] = 'straight' if numpy.isclose(dy, 0) else 's'
in_key = self._wildcard_ptype_key(their_port.ptype)
endpoint = Port((dx, dy), rotation=pi, ptype=our_port.ptype)
offers = self._transition_adapter_offers_by_key.setdefault((kind, in_key), [])
priority_bias = len(offers) * BUILTIN_PRIORITY_STEP
if kind == 'straight':
offers.append(StraightOffer.prebuilt(
in_ptype = their_port.ptype,
out_ptype = our_port.ptype,
endpoint = endpoint,
data = transition_data,
priority_bias = priority_bias,
bbox_for_data = self._bbox_for_data,
))
return
offers.append(SOffer.prebuilt(
in_ptype = their_port.ptype,
out_ptype = our_port.ptype,
endpoint = endpoint,
data = transition_data,
priority_bias = priority_bias,
bbox_for_data = self._bbox_for_data,
))
@staticmethod
def _signed_jog_domains(magnitude_range: tuple[float, float]) -> tuple[tuple[float, float], ...]:
lower, upper = (float(magnitude_range[0]), float(magnitude_range[1]))
if lower < 0 or lower > upper:
return ()
if lower == upper:
if lower == 0:
return ((0.0, 0.0),)
return ((lower, lower), (-lower, -lower))
positive = (lower, upper)
neg_lower = -numpy.inf if numpy.isinf(upper) else float(numpy.nextafter(-upper, numpy.inf))
negative = (neg_lower, -lower)
domains: list[tuple[float, float]] = [positive]
if neg_lower < -lower:
domains.append(negative)
if lower > 0:
domains.append((-lower, -lower))
return tuple(domains)
def primitive_offers(
self,
kind: PrimitiveKind,
*,
in_ptype: str | None = None,
out_ptype: str | None = None,
**kwargs,
) -> tuple[PrimitiveOffer, ...]:
_ = out_ptype
if kind == 'straight':
in_key = self._wildcard_ptype_key(in_ptype)
return (
*self._transition_adapter_offers_by_key.get(('straight', in_key), ()),
*self._straight_offers,
)
if kind == 'bend':
return tuple(self._bend_offers[int(bool(kwargs['ccw']))])
if kind == 's':
in_key = self._wildcard_ptype_key(in_ptype)
return (
*self._transition_adapter_offers_by_key.get(('s', in_key), ()),
*self._s_offers,
)
if kind == 'u':
return tuple(self._u_offers)
raise BuildError(f'Unrecognized primitive offer kind {kind!r}')
def _render_generated(
self,
data: GeneratedData,
tree: ILibrary,
port_names: tuple[str, str],
gen_kwargs: dict[str, Any],
) -> ILibrary:
if numpy.isclose(data.parameter, 0):
return tree
pat = tree.top_pattern()
generated = data.fn(data.parameter, **gen_kwargs)
pmap = {port_names[1]: data.port_name}
if isinstance(generated, Pattern):
pat.plug(generated, pmap, append=True, mirrored=data.mirrored)
else:
top = generated.top()
generated.flatten(top, dangling_ok=True)
pat.plug(generated[top], pmap, append=True, mirrored=data.mirrored)
return tree
def _render_reusable(
self,
data: ReusableData,
tree: ILibrary,
port_names: tuple[str, str],
) -> ILibrary:
pat = tree.top_pattern()
pat.plug(data.abstract, {port_names[1]: data.port_name}, mirrored=data.mirrored)
return tree
def _render_data(
self,
data: Any,
tree: ILibrary,
port_names: tuple[str, str],
gen_kwargs: dict[str, Any],
) -> ILibrary:
if isinstance(data, self.GeneratedData):
return self._render_generated(data=data, tree=tree, port_names=port_names, gen_kwargs=gen_kwargs)
if isinstance(data, self.ReusableData):
return self._render_reusable(data=data, tree=tree, port_names=port_names)
raise BuildError(f'Unexpected AutoTool render data {type(data).__name__}')
def render(
self,
batch: Sequence[RenderStep],
*,
port_names: tuple[str, str] = ('A', 'B'),
**kwargs,
) -> ILibrary:
tree, pat = Library.mktree(SINGLE_USE_PREFIX + 'traceL')
pat.add_port_pair(names=(port_names[0], port_names[1]))
for step in batch:
assert step.tool == self
self._render_data(step.data, tree, port_names, kwargs)
return tree
@dataclass
class PathTool(Tool):
"""
Tool that renders routes directly as `Pattern.path()` geometry.
`PathTool` supports L and S primitive offers. `render()` combines a
compatible batch of L/S `RenderStep`s into one multi-vertex path. U routes
are left to `Pather` synthesis or to a different tool.
"""
layer: layer_t
""" Layer to draw generated path geometry on. """
width: float
""" Width of generated path geometry. """
ptype: str = 'unk'
""" Port type for generated input and output ports. """
def _bend_radius(self) -> float:
return self.width / 2
def _plan_l_vertices(self, length: float, bend_run: float) -> NDArray[numpy.float64]:
vertices = [(0.0, 0.0), (length, 0.0)]
if not numpy.isclose(bend_run, 0):
vertices.append((length, bend_run))
return numpy.array(vertices, dtype=float)
def _plan_s_vertices(self, length: float, jog: float) -> NDArray[numpy.float64]:
if numpy.isclose(jog, 0):
return numpy.array([(0.0, 0.0), (length, 0.0)], dtype=float)
if length < self.width:
raise BuildError(
f'Asked to draw S-path with total length {length:,g}, shorter than required bend: {self.width:,g}'
)
# Match AutoTool's straight-then-s-bend placement so the jog happens
# width/2 before the end while still allowing smaller lateral offsets.
jog_x = length - self._bend_radius()
vertices = [
(0.0, 0.0),
(jog_x, 0.0),
(jog_x, jog),
(length, jog),
]
return numpy.array(vertices, dtype=float)
def _path_bbox(self, vertices: NDArray[numpy.float64]) -> NDArray[numpy.float64]:
return Path(vertices=vertices, width=self.width).get_bounds_single()
def primitive_offers(
self,
kind: PrimitiveKind,
*,
in_ptype: str | None = None,
out_ptype: str | None = None,
**kwargs,
) -> tuple[PrimitiveOffer, ...]:
if kind == 'u':
return ()
if not ptypes_compatible(out_ptype, self.ptype):
raise BuildError(f'Requested {out_ptype=} does not match path ptype {self.ptype}')
ptype = self.ptype
if kind == 'straight':
def straight_data(length: float) -> NDArray[numpy.float64]:
return numpy.array((length, 0.0))
def endpoint_straight(length: float) -> Port:
data = straight_data(length)
return Port(data, rotation=pi, ptype=ptype)
def bbox_straight(length: float) -> NDArray[numpy.float64]:
data = straight_data(length)
return self._path_bbox(self._plan_l_vertices(float(data[0]), float(data[1])))
return (StraightOffer(
in_ptype=in_ptype,
out_ptype=ptype,
bbox_planner=bbox_straight,
endpoint_planner=endpoint_straight,
commit_planner=straight_data,
),)
if kind == 'bend':
ccw = kwargs['ccw']
radius = self._bend_radius()
bend_run = radius if bool(ccw) else -radius
bend_angle = -pi / 2 if bool(ccw) else pi / 2
def bend_data(length: float) -> NDArray[numpy.float64]:
_ = length
return numpy.array((length, bend_run))
def endpoint_bend(length: float) -> Port:
data = bend_data(length)
return Port(data, rotation=bend_angle, ptype=ptype)
def bbox_bend(length: float) -> NDArray[numpy.float64]:
data = bend_data(length)
return self._path_bbox(self._plan_l_vertices(float(data[0]), float(data[1])))
return (BendOffer(
in_ptype=in_ptype,
out_ptype=ptype,
ccw=bool(ccw),
length_domain=(radius, radius),
bbox_planner=bbox_bend,
endpoint_planner=endpoint_bend,
commit_planner=bend_data,
),)
if kind == 's':
def minimum_length(jog: float) -> float:
if numpy.isclose(jog, 0):
return 0.0
return self.width
def s_data(jog: float) -> NDArray[numpy.float64]:
length = minimum_length(jog)
self._plan_s_vertices(length, jog)
return numpy.array((length, jog))
def endpoint_s(jog: float) -> Port:
data = s_data(jog)
return Port(data, rotation=pi, ptype=ptype)
def bbox_s(jog: float) -> NDArray[numpy.float64]:
data = s_data(jog)
return self._path_bbox(self._plan_s_vertices(float(data[0]), float(data[1])))
return (SOffer(
in_ptype=in_ptype,
out_ptype=ptype,
bbox_planner=bbox_s,
endpoint_planner=endpoint_s,
commit_planner=s_data,
),)
raise BuildError(f'Unrecognized primitive offer kind {kind!r}')
def render(
self,
batch: Sequence[RenderStep],
*,
port_names: tuple[str, str] = ('A', 'B'),
**kwargs, # noqa: ARG002 (unused)
) -> ILibrary:
# Transform the batch so the first port is local (at 0,0) but retains its global rotation.
# This allows the path to be rendered with its original orientation, simplified by
# translation to the origin. Pather.render will handle the final placement
# (including rotation alignment) via `pat.plug`.
first_port = batch[0].start_port
translation = -first_port.offset
rotation = 0
pivot = first_port.offset
# Localize the batch for rendering
local_batch = [step.transformed(translation, rotation, pivot) for step in batch]
path_vertices = [local_batch[0].start_port.offset]
for step in local_batch:
assert step.tool == self
port_rot = step.start_port.rotation
# Masque convention: Port rotation points INTO the device.
# So the direction of travel for the path is AWAY from the port, i.e., port_rot + pi.
assert port_rot is not None
transform = rotation_matrix_2d(port_rot + pi)
delta = step.end_port.offset - step.start_port.offset
local_end = rotation_matrix_2d(-(port_rot + pi)) @ delta
if step.opcode == 'L':
local_vertices = self._plan_l_vertices(float(local_end[0]), float(local_end[1]))
elif step.opcode == 'S':
local_vertices = self._plan_s_vertices(float(local_end[0]), float(local_end[1]))
else:
raise BuildError(f'Unrecognized opcode "{step.opcode}"')
for vertex in local_vertices[1:]:
path_vertices.append(step.start_port.offset + transform @ vertex)
tree, pat = Library.mktree(SINGLE_USE_PREFIX + 'traceL')
pat.path(layer=self.layer, width=self.width, vertices=path_vertices)
pat.ports = {
port_names[0]: local_batch[0].start_port.copy().rotate(pi),
port_names[1]: local_batch[-1].end_port.copy().rotate(pi),
}
return tree