1788 lines
73 KiB
Python
1788 lines
73 KiB
Python
"""
|
|
Unified Pattern assembly and routing (`Pather`).
|
|
|
|
`Pather` is the public object that owns layout state. It holds the working
|
|
Pattern, Library, per-port Tool assignments, pending `RenderStep`s, and routing
|
|
side effects such as plug consumption, port renames, and automatic rendering. The
|
|
planner package is intentionally internal: custom route generators should
|
|
extend `Tool.primitive_offers()` and `Tool.render()` rather than depending on
|
|
planner classes or search details.
|
|
|
|
Public routing arguments are explicit. Planner-specific per-route settings
|
|
belong in `plan_options`, while custom Tool offer values belong in
|
|
`tool_options`. Pather keeps those namespaces separate and forwards Tool
|
|
options only to offer discovery. A Tool must capture any selected render-time
|
|
value in its offer's committed data.
|
|
|
|
Routing is split into four ownership phases:
|
|
- snapshot: `Pather` resolves the active Tool for each requested port and
|
|
passes copied ports to the planner as `RoutePortContext` values,
|
|
- selection/preparation: the planner validates the route mode, selects a legal
|
|
primitive-offer composition, commits only the chosen offers into opaque
|
|
`RenderStep.data`, and returns prepared actions,
|
|
- application: `Pather` appends the prepared steps to its pending queue,
|
|
replaces live output ports, consumes plug destinations, applies deferred
|
|
trace-thru renames, and batches immediate rendering around the whole route,
|
|
- rendering: pending steps are grouped by live port, Tool, and continuity before
|
|
`Tool.render()` builds geometry for each compatible batch.
|
|
|
|
This split keeps selection failures largely transactional for live Pather
|
|
state: unsupported primitive combinations can fail before the Pattern, pending
|
|
step queue, or Library are touched. Once prepared actions are applied, plug,
|
|
rename, render, or insertion failures may leave partial output; that mutation
|
|
boundary belongs to Pather, not to Tool implementations or the route solver.
|
|
|
|
Routing policy follows port names. Port-specific Tool assignments and
|
|
`PortPather` selections remain attached to their names rather than following a
|
|
physical port through arbitrary renames. Pending `RenderStep`s are different:
|
|
they snapshot their geometric endpoints when planned. The `_paths` keys are
|
|
rendering buckets used for ordering and batching those historical steps, not
|
|
identities that can retarget their saved geometry when a name is deleted or
|
|
reused.
|
|
|
|
While pending steps exist, mutate ports through Pather's methods. Directly
|
|
editing `pather.pattern.ports` bypasses the bookkeeping that maintains render
|
|
buckets and is unsupported.
|
|
|
|
Rendering is not transactional across Pattern and Library mutations. A render
|
|
exception is terminal for that Pather: callers may catch it for reporting or
|
|
cleanup, but must not retry rendering or continue routing with the same object.
|
|
"""
|
|
from typing import Self, Any, Literal, Protocol, overload
|
|
from collections.abc import Iterator, Iterable, Mapping, MutableMapping, Sequence
|
|
import copy
|
|
import logging
|
|
from collections import defaultdict
|
|
from functools import wraps
|
|
from pprint import pformat
|
|
from contextlib import contextmanager
|
|
from itertools import chain
|
|
from types import TracebackType
|
|
from types import MappingProxyType
|
|
|
|
import numpy
|
|
from numpy import pi
|
|
from numpy.typing import ArrayLike
|
|
|
|
from ..pattern import Pattern
|
|
from ..library import ILibrary, TreeView, SINGLE_USE_PREFIX
|
|
from ..error import BuildError, PortError
|
|
from ..ports import PortList, Port
|
|
from ..abstract import Abstract
|
|
from ..utils import SupportsBool, ptypes_compatible
|
|
from .tools import (
|
|
Tool,
|
|
RenderStep,
|
|
)
|
|
from .planner.interface import (
|
|
PreparedRouteResult,
|
|
RoutePortContext,
|
|
route_failure_policy,
|
|
)
|
|
from .error import RouteFailurePolicy, ToolContractError
|
|
from .planner import RoutingPlanner
|
|
from .planner.bounds import resolved_position_bound
|
|
from .logging import PatherLogger
|
|
from ._tolerances import angles_equal, array_close
|
|
|
|
|
|
logger = logging.getLogger(__name__)
|
|
RenderPolicy = Literal['auto', 'immediate', 'deferred', 'warn', 'error', 'ignore']
|
|
RENDER_POLICIES: tuple[RenderPolicy, ...] = ('auto', 'immediate', 'deferred', 'warn', 'error', 'ignore')
|
|
RESERVED_TOOL_OPTION_KEYS = frozenset(('kind', 'in_ptype', 'out_ptype', 'ccw'))
|
|
|
|
|
|
class RouteCompletionCallback(Protocol):
|
|
"""Callback invoked after one routing operation updates live Pather state."""
|
|
|
|
def __call__(
|
|
self,
|
|
pather: 'Pather',
|
|
endpoints: Mapping[str, Port],
|
|
/,
|
|
) -> None:
|
|
"""Handle copied final endpoints keyed by their original routed names."""
|
|
...
|
|
|
|
|
|
def _present_route_args(**kwargs: Any) -> dict[str, Any]:
|
|
"""Return explicitly supplied route arguments for planner validation."""
|
|
return {key: value for key, value in kwargs.items() if value is not None}
|
|
|
|
|
|
def _validated_tool_options(tool_options: Mapping[str, Any] | None) -> dict[str, Any]:
|
|
"""Copy and validate custom planning options before any Tool is queried."""
|
|
if tool_options is None:
|
|
return {}
|
|
try:
|
|
options = dict(tool_options)
|
|
except (TypeError, ValueError) as err:
|
|
raise BuildError('tool_options must be a mapping with string keys') from err
|
|
nonstring = [key for key in options if not isinstance(key, str)]
|
|
if nonstring:
|
|
raise BuildError(f'tool_options keys must be strings; got {nonstring!r}')
|
|
collisions = sorted(RESERVED_TOOL_OPTION_KEYS & options.keys())
|
|
if collisions:
|
|
raise BuildError(f'tool_options cannot override Tool arguments: {", ".join(collisions)}')
|
|
return options
|
|
|
|
|
|
def _validated_plan_options(plan_options: Mapping[str, Any] | None) -> dict[str, Any]:
|
|
"""Copy generic per-route planner options without interpreting their keys."""
|
|
if plan_options is None:
|
|
return {}
|
|
try:
|
|
options = dict(plan_options)
|
|
except (TypeError, ValueError) as err:
|
|
raise BuildError('plan_options must be a mapping with string keys') from err
|
|
nonstring = [key for key in options if not isinstance(key, str)]
|
|
if nonstring:
|
|
raise BuildError(f'plan_options keys must be strings; got {nonstring!r}')
|
|
return options
|
|
|
|
|
|
class Pather(PortList):
|
|
"""
|
|
A `Pather` is a helper object used for snapping together multiple
|
|
lower-level patterns at their `Port`s, and for routing single-use
|
|
patterns (e.g. wires or waveguides) between them.
|
|
|
|
The `Pather` holds context in the form of a `Library`, its underlying
|
|
pattern, and a set of `Tool`s for generating routing segments.
|
|
|
|
Routing operations (`trace`, `jog`, `uturn`, etc.) select primitive offers
|
|
from the active `Tool` and compose them into `RenderStep`s. By default,
|
|
geometry is rendered after each route, unless the `Pather` is used as a
|
|
context manager.
|
|
|
|
Examples: Creating a Pather
|
|
===========================
|
|
- `Pather(library, tools=my_tool)` makes an empty pattern with no ports.
|
|
The default routing tool for all ports is set to `my_tool`.
|
|
|
|
- `Pather(library, name='mypat')` makes an empty pattern and adds it to
|
|
`library` under the name `'mypat'`.
|
|
|
|
Examples: Adding to a pattern
|
|
=============================
|
|
- `pather.plug(subdevice, {'A': 'C'})` instantiates `subdevice` and
|
|
connects port 'A' of the current pattern to port 'C' of `subdevice`.
|
|
|
|
- `pather.trace('my_port', ccw=True, length=100)` plans a 100-unit bend
|
|
starting at 'my_port'. If the `Pather` is used as a context manager,
|
|
geometry is generated on clean context exit.
|
|
|
|
Examples: Route completion
|
|
==========================
|
|
Route completion callbacks can add labels or other endpoint annotations
|
|
without wrapping the individual routing methods::
|
|
|
|
def label_route_endpoints(pather, endpoints):
|
|
for name, port in endpoints.items():
|
|
pather.label('PORT_LABELS', string=name, offset=port.offset)
|
|
|
|
pather = Pather(
|
|
library,
|
|
tools=my_tool,
|
|
on_route_complete=label_route_endpoints,
|
|
)
|
|
"""
|
|
__slots__ = (
|
|
'pattern', 'library', 'tools', 'planner', '_paths',
|
|
'on_route_complete', '_dead', '_logger', '_render_policy', '_render_append', '_context_depth'
|
|
)
|
|
|
|
pattern: Pattern
|
|
""" Layout of this device """
|
|
|
|
library: ILibrary
|
|
""" Library from which patterns should be referenced """
|
|
|
|
tools: dict[str | None, Tool]
|
|
"""
|
|
Tool objects used to dynamically generate new routing segments.
|
|
A key of `None` indicates the default `Tool`.
|
|
|
|
Non-`None` keys are policies attached to port names. Renaming a port does
|
|
not transfer its Tool assignment to the new name.
|
|
"""
|
|
|
|
planner: RoutingPlanner
|
|
"""
|
|
Stateless route-selection facade.
|
|
|
|
Per-solve mutable state belongs in routing search/catalog objects rather
|
|
than on the planner instance.
|
|
"""
|
|
|
|
on_route_complete: RouteCompletionCallback | None
|
|
"""Optional callback run once after each routing operation updates live state."""
|
|
|
|
_dead: bool
|
|
""" If True, geometry generation is skipped (for debugging) """
|
|
|
|
_logger: PatherLogger
|
|
""" Handles diagnostic logging of operations """
|
|
|
|
_render_policy: RenderPolicy
|
|
""" Routing render behavior """
|
|
|
|
_render_append: bool
|
|
""" If True, automatic render calls append geometry instead of adding references """
|
|
|
|
_context_depth: int
|
|
""" Number of active context-manager entries """
|
|
|
|
_paths: defaultdict[str, list[RenderStep]]
|
|
""" Per-port pending render steps, consumed by `render()` """
|
|
|
|
def _route_context(self, portspec: str) -> RoutePortContext:
|
|
"""
|
|
Snapshot the live port and selected Tool for planning.
|
|
|
|
The port copy lets route-selection failures leave live Pather state
|
|
unchanged. Tool lookup prefers a port-specific Tool and falls back to
|
|
the `None` default.
|
|
"""
|
|
tool = self.tools.get(portspec, self.tools.get(None))
|
|
if tool is None:
|
|
raise BuildError(f'No tool assigned for port {portspec}')
|
|
return RoutePortContext(portspec, self.pattern[portspec].copy(), tool)
|
|
|
|
def _route_contexts(self, portspecs: Sequence[str]) -> tuple[RoutePortContext, ...]:
|
|
"""Snapshot several ports in request order for bundle planning."""
|
|
if not portspecs:
|
|
raise BuildError('Routing requires at least one port')
|
|
seen: set[str] = set()
|
|
duplicates: set[str] = set()
|
|
for portspec in portspecs:
|
|
if portspec in seen:
|
|
duplicates.add(portspec)
|
|
seen.add(portspec)
|
|
if duplicates:
|
|
raise BuildError(f'Routing port names must be unique; got duplicates: {sorted(duplicates)}')
|
|
return tuple(self._route_context(portspec) for portspec in portspecs)
|
|
|
|
@property
|
|
def ports(self) -> dict[str, Port]:
|
|
return self.pattern.ports
|
|
|
|
@ports.setter
|
|
def ports(self, value: dict[str, Port]) -> None:
|
|
self.pattern.ports = value
|
|
|
|
def __init__(
|
|
self,
|
|
library: ILibrary,
|
|
*,
|
|
pattern: Pattern | None = None,
|
|
ports: str | Mapping[str, Port] | None = None,
|
|
tools: Tool | MutableMapping[str | None, Tool] | None = None,
|
|
name: str | None = None,
|
|
debug: bool = False,
|
|
render: RenderPolicy = 'auto',
|
|
render_append: bool = True,
|
|
planner: RoutingPlanner | None = None,
|
|
on_route_complete: RouteCompletionCallback | None = None,
|
|
) -> None:
|
|
"""
|
|
Args:
|
|
library: The library for pattern references and generated segments.
|
|
pattern: The pattern to modify. If `None`, a new one is created.
|
|
ports: Initial set of ports. May be a string (name in `library`)
|
|
or a port mapping.
|
|
tools: Tool(s) to use for routing segments.
|
|
name: If specified, `library[name]` is set to `self.pattern`.
|
|
debug: If True, enables detailed logging.
|
|
render: Routing render policy. `'auto'` renders after each route
|
|
outside a context manager and defers until clean context exit
|
|
inside one. Use `'immediate'` to always render after each route,
|
|
`'deferred'` to keep paths pending until `render()` or clean
|
|
context exit, `'warn'` to log pending paths on clean context
|
|
exit, `'error'` to reject pending paths on clean context exit,
|
|
or `'ignore'` to leave pending paths silent.
|
|
render_append: If an automatic render is triggered, determines
|
|
whether to append geometry or add a reference.
|
|
planner: Optional stateless route-selection planner. If omitted,
|
|
a new `RoutingPlanner` is used.
|
|
on_route_complete: Optional callback invoked once per routing
|
|
operation after port updates, plugs, and renames, but before
|
|
automatic rendering. It receives this Pather and a read-only,
|
|
request-ordered mapping from original routed names to copied
|
|
final Ports. Callback exceptions propagate without rollback.
|
|
"""
|
|
if render not in RENDER_POLICIES:
|
|
raise BuildError(f'Invalid render policy {render!r}; expected one of {RENDER_POLICIES}')
|
|
|
|
self._dead = False
|
|
self._logger = PatherLogger(debug=debug)
|
|
self._render_policy = render
|
|
self._render_append = render_append
|
|
self._context_depth = 0
|
|
self.library = library
|
|
self.pattern = pattern if pattern is not None else Pattern()
|
|
self.planner = RoutingPlanner() if planner is None else planner
|
|
self.on_route_complete = on_route_complete
|
|
self._paths = defaultdict(list)
|
|
|
|
if ports is not None:
|
|
if self.pattern.ports:
|
|
raise BuildError('Ports supplied for pattern with pre-existing ports!')
|
|
if isinstance(ports, str):
|
|
ports = library.abstract(ports).ports
|
|
self.pattern.ports.update(copy.deepcopy(dict(ports)))
|
|
|
|
if tools is None:
|
|
self.tools = {}
|
|
elif isinstance(tools, Tool):
|
|
self.tools = {None: tools}
|
|
else:
|
|
self.tools = dict(tools)
|
|
|
|
if name is not None:
|
|
library[name] = self.pattern
|
|
|
|
def __enter__(self) -> Self:
|
|
self._context_depth += 1
|
|
return self
|
|
|
|
def __exit__(
|
|
self,
|
|
exc_type: type[BaseException] | None,
|
|
exc_value: BaseException | None,
|
|
traceback: TracebackType | None,
|
|
) -> bool:
|
|
_ = exc_value, traceback
|
|
self._context_depth -= 1
|
|
if exc_type is not None or self._context_depth != 0 or not any(self._paths.values()):
|
|
return False
|
|
|
|
if self._render_policy in ('auto', 'deferred'):
|
|
self.render(append=self._render_append)
|
|
elif self._render_policy == 'warn':
|
|
logger.warning(
|
|
'Pather context exited with %s; call render() or use render="deferred"',
|
|
self._pending_render_summary(),
|
|
)
|
|
elif self._render_policy == 'error':
|
|
raise BuildError(f'Pather context exited with {self._pending_render_summary()}')
|
|
return False
|
|
|
|
def _pending_render_summary(self) -> str:
|
|
ports = [(portspec, len(steps)) for portspec, steps in self._paths.items() if steps]
|
|
port_count = len(ports)
|
|
step_count = sum(count for _portspec, count in ports)
|
|
return (
|
|
f'{step_count} pending render step{"s" if step_count != 1 else ""} '
|
|
f'on {port_count} port{"s" if port_count != 1 else ""}'
|
|
)
|
|
|
|
def __repr__(self) -> str:
|
|
s = f'<Pather {self.pattern} L({len(self.library)}) {pformat(self.tools)}>'
|
|
return s
|
|
|
|
#
|
|
# Core Pattern Operations (Immediate)
|
|
#
|
|
def _prepare_breaks(self, names: Iterable[str | None]) -> list[tuple[str, RenderStep]]:
|
|
""" Snapshot break markers to be committed after a successful mutation. """
|
|
prepared: list[tuple[str, RenderStep]] = []
|
|
if self._dead:
|
|
return prepared
|
|
for name in names:
|
|
if name is None:
|
|
continue
|
|
steps = self._paths.get(name)
|
|
if not steps:
|
|
continue
|
|
port = self.ports.get(name, steps[-1].end_port)
|
|
prepared.append((name, RenderStep('plug', None, port.copy(), port.copy(), None)))
|
|
return prepared
|
|
|
|
def _commit_breaks(self, prepared: Iterable[tuple[str, RenderStep]]) -> None:
|
|
""" Append previously prepared break markers. """
|
|
for name, step in prepared:
|
|
self._paths[name].append(step)
|
|
|
|
def plug(
|
|
self,
|
|
other: Abstract | str | Pattern | TreeView,
|
|
map_in: dict[str, str],
|
|
map_out: dict[str, str | None] | None = None,
|
|
**kwargs,
|
|
) -> Self:
|
|
with self._logger.log_operation(self, 'plug', list(map_in.keys()), map_out=map_out, **kwargs):
|
|
other = self.library.resolve(other, append=kwargs.get('append', False))
|
|
|
|
prepared_breaks: list[tuple[str, RenderStep]] = []
|
|
if not self._dead:
|
|
other_ports = other.ports
|
|
affected = set(map_in.keys())
|
|
plugged = set(map_in.values())
|
|
for name in other_ports:
|
|
if name not in plugged:
|
|
new_name = (map_out or {}).get(name, name)
|
|
if new_name is not None:
|
|
affected.add(new_name)
|
|
prepared_breaks = self._prepare_breaks(affected)
|
|
elif self._logger.debug:
|
|
logger.warning("Skipping geometry for plug() since device is dead")
|
|
|
|
self.pattern.plug(other=other, map_in=map_in, map_out=map_out, skip_geometry=self._dead, **kwargs)
|
|
self._commit_breaks(prepared_breaks)
|
|
return self
|
|
|
|
def place(
|
|
self,
|
|
other: Abstract | str | Pattern | TreeView,
|
|
port_map: dict[str, str | None] | None = None,
|
|
**kwargs,
|
|
) -> Self:
|
|
with self._logger.log_operation(self, 'place', None, port_map=port_map, **kwargs):
|
|
other = self.library.resolve(other, append=kwargs.get('append', False))
|
|
|
|
prepared_breaks: list[tuple[str, RenderStep]] = []
|
|
if not self._dead:
|
|
other_ports = other.ports
|
|
affected = set()
|
|
for name in other_ports:
|
|
new_name = (port_map or {}).get(name, name)
|
|
if new_name is not None:
|
|
affected.add(new_name)
|
|
prepared_breaks = self._prepare_breaks(affected)
|
|
elif self._logger.debug:
|
|
logger.warning("Skipping geometry for place() since device is dead")
|
|
|
|
self.pattern.place(other=other, port_map=port_map, skip_geometry=self._dead, **kwargs)
|
|
self._commit_breaks(prepared_breaks)
|
|
return self
|
|
|
|
def plugged(self, connections: dict[str, str]) -> Self:
|
|
with self._logger.log_operation(self, 'plugged', list(connections.keys()), connections=connections):
|
|
prepared_breaks = self._prepare_breaks(chain(connections.keys(), connections.values()))
|
|
self.pattern.plugged(connections)
|
|
self._commit_breaks(prepared_breaks)
|
|
return self
|
|
|
|
def rename_ports(self, mapping: dict[str, str | None], overwrite: bool = False) -> Self:
|
|
with self._logger.log_operation(self, 'rename_ports', list(mapping.keys()), mapping=mapping, overwrite=overwrite):
|
|
winners = self.pattern._rename_ports_impl(
|
|
mapping,
|
|
overwrite=overwrite or self._dead,
|
|
allow_collisions=self._dead,
|
|
)
|
|
|
|
moved_steps = {kk: self._paths.pop(kk) for kk in mapping if kk in self._paths}
|
|
for kk, steps in moved_steps.items():
|
|
vv = mapping[kk]
|
|
# Preserve deferred geometry even if the live port is deleted.
|
|
# `render()` can still materialize the saved steps using their stored start/end ports.
|
|
# Current semantics intentionally keep deleted ports' queued steps under the old key,
|
|
# so if a new live port later reuses that name it does not retarget the old geometry;
|
|
# the old and new routes merely share a render bucket until `render()` consumes them.
|
|
target = kk if vv is None else vv
|
|
if self._dead and vv is not None and winners.get(vv) != kk:
|
|
target = kk
|
|
self._paths[target].extend(steps)
|
|
return self
|
|
|
|
def set_dead(self) -> Self:
|
|
self._dead = True
|
|
return self
|
|
|
|
#
|
|
# Pattern Wrappers
|
|
#
|
|
@wraps(Pattern.label)
|
|
def label(self, *args, **kwargs) -> Self:
|
|
self.pattern.label(*args, **kwargs)
|
|
return self
|
|
|
|
@wraps(Pattern.ref)
|
|
def ref(self, *args, **kwargs) -> Self:
|
|
self.pattern.ref(*args, **kwargs)
|
|
return self
|
|
|
|
@wraps(Pattern.polygon)
|
|
def polygon(self, *args, **kwargs) -> Self:
|
|
self.pattern.polygon(*args, **kwargs)
|
|
return self
|
|
|
|
@wraps(Pattern.rect)
|
|
def rect(self, *args, **kwargs) -> Self:
|
|
self.pattern.rect(*args, **kwargs)
|
|
return self
|
|
|
|
@wraps(Pattern.path)
|
|
def path(self, *args, **kwargs) -> Self:
|
|
self.pattern.path(*args, **kwargs)
|
|
return self
|
|
|
|
def translate(self, offset: ArrayLike) -> Self:
|
|
with self._logger.log_operation(self, 'translate', list(self.ports.keys()), offset=offset):
|
|
offset_arr = numpy.asarray(offset)
|
|
self.pattern.translate_elements(offset_arr)
|
|
for steps in self._paths.values():
|
|
for i, step in enumerate(steps):
|
|
steps[i] = step.transformed(offset_arr, 0, numpy.zeros(2))
|
|
return self
|
|
|
|
def rotate_around(self, pivot: ArrayLike, angle: float) -> Self:
|
|
with self._logger.log_operation(self, 'rotate_around', list(self.ports.keys()), pivot=pivot, angle=angle):
|
|
pivot_arr = numpy.asarray(pivot)
|
|
self.pattern.rotate_around(pivot_arr, angle)
|
|
for steps in self._paths.values():
|
|
for i, step in enumerate(steps):
|
|
steps[i] = step.transformed(numpy.zeros(2), angle, pivot_arr)
|
|
return self
|
|
|
|
def mirror(self, axis: int = 0) -> Self:
|
|
with self._logger.log_operation(self, 'mirror', list(self.ports.keys()), axis=axis):
|
|
self.pattern.mirror(axis)
|
|
for steps in self._paths.values():
|
|
for i, step in enumerate(steps):
|
|
steps[i] = step.mirrored(axis)
|
|
return self
|
|
|
|
def mkport(self, name: str, value: Port) -> Self:
|
|
with self._logger.log_operation(self, 'mkport', name, value=value):
|
|
super().mkport(name, value)
|
|
return self
|
|
|
|
#
|
|
# Routing Logic (Deferred / Incremental)
|
|
#
|
|
def _notify_route_complete(self, endpoints: Iterable[tuple[str, Port]]) -> None:
|
|
"""Invoke the configured route callback with isolated endpoint snapshots."""
|
|
callback = self.on_route_complete
|
|
if callback is None:
|
|
return
|
|
snapshots = MappingProxyType({name: port.copy() for name, port in endpoints})
|
|
callback(self, snapshots)
|
|
|
|
def _apply_route_result(self, result: PreparedRouteResult) -> None:
|
|
"""
|
|
Apply every action and deferred rename in a prepared route result.
|
|
|
|
Route actions may contain several primitive render steps and port
|
|
mutations. Immediate rendering happens once after the whole prepared
|
|
result has been applied.
|
|
"""
|
|
for action in result.actions:
|
|
if not action.render_steps:
|
|
raise BuildError('Prepared route action has no render steps')
|
|
|
|
if not self._dead:
|
|
self._paths[action.portspec].extend(action.render_steps)
|
|
|
|
self.pattern.ports[action.portspec] = action.final_port.copy()
|
|
|
|
if action.plug_into is not None:
|
|
self.plugged({action.portspec: action.plug_into})
|
|
for old_name, new_name in result.renames:
|
|
self.rename_ports({old_name: new_name})
|
|
self._notify_route_complete(
|
|
(action.portspec, action.final_port) for action in result.actions
|
|
)
|
|
render_immediately = (
|
|
self._render_policy == 'immediate'
|
|
or (self._render_policy == 'auto' and self._context_depth == 0)
|
|
)
|
|
if render_immediately and any(self._paths.values()):
|
|
self.render(append=self._render_append)
|
|
|
|
def _apply_dead_fallback(
|
|
self,
|
|
portspec: str,
|
|
length: float,
|
|
jog: float,
|
|
ccw: SupportsBool | None,
|
|
in_ptype: str,
|
|
plug_into: str | None = None,
|
|
*,
|
|
out_rot: float | None = None,
|
|
out_ptype: str | None = None,
|
|
) -> Port:
|
|
"""
|
|
Move a dead Pather port without generating geometry.
|
|
|
|
Dead fallback is only for debugging or dry layout flow. Fatal route
|
|
errors bypass it because they indicate an invalid Tool offer contract.
|
|
"""
|
|
if out_rot is None:
|
|
if ccw is None:
|
|
out_rot = pi
|
|
elif bool(ccw):
|
|
out_rot = -pi / 2
|
|
else:
|
|
out_rot = pi / 2
|
|
logger.warning(f"Tool planning failed for dead pather. Using dummy extension for {portspec}.")
|
|
port = self.pattern[portspec]
|
|
port_rot = port.rotation
|
|
if port_rot is None:
|
|
raise PortError('Ports must have rotation')
|
|
out_port = Port((length, jog), rotation=out_rot, ptype=out_ptype or in_ptype)
|
|
out_port.rotate_around((0, 0), pi + port_rot)
|
|
out_port.translate(port.offset)
|
|
self.pattern.ports[portspec] = out_port
|
|
if plug_into is not None:
|
|
self.plugged({portspec: plug_into})
|
|
return out_port.copy()
|
|
|
|
#
|
|
# High-level Routing Methods
|
|
#
|
|
def trace(
|
|
self,
|
|
portspec: str | Sequence[str],
|
|
ccw: SupportsBool | None,
|
|
length: float | None = None,
|
|
*,
|
|
spacing: float | ArrayLike | None = None,
|
|
plan_options: Mapping[str, Any] | None = None,
|
|
out_ptype: str | None = None,
|
|
each: float | None = None,
|
|
set_rotation: float | None = None,
|
|
emin: float | None = None,
|
|
emax: float | None = None,
|
|
pmin: float | None = None,
|
|
pmax: float | None = None,
|
|
xmin: float | None = None,
|
|
xmax: float | None = None,
|
|
ymin: float | None = None,
|
|
ymax: float | None = None,
|
|
min_past_furthest: float | None = None,
|
|
tool_options: Mapping[str, Any] | None = None,
|
|
) -> Self:
|
|
"""
|
|
Route one or more ports using straight segments or single 90-degree bends.
|
|
|
|
Provide exactly one routing mode:
|
|
- `length` for a single port,
|
|
- `each` to extend each selected port independently by the same amount, or
|
|
- one bundle bound such as `xmin`, `emax`, or `min_past_furthest`.
|
|
|
|
For a single port with no length or bound, legal primitive-offer
|
|
candidates are evaluated at their minimum legal length-like parameters,
|
|
then cost selects among those minimum-length candidates. `out_ptype`,
|
|
when provided, constrains only the final route endpoint. Planner-specific
|
|
per-route settings belong in `plan_options`.
|
|
|
|
`spacing` and `set_rotation` are only valid when using a bundle bound.
|
|
"""
|
|
bounds = _present_route_args(
|
|
out_ptype=out_ptype,
|
|
each=each,
|
|
set_rotation=set_rotation,
|
|
emin=emin,
|
|
emax=emax,
|
|
pmin=pmin,
|
|
pmax=pmax,
|
|
xmin=xmin,
|
|
xmax=xmax,
|
|
ymin=ymin,
|
|
ymax=ymax,
|
|
min_past_furthest=min_past_furthest,
|
|
)
|
|
plan_opts = _validated_plan_options(plan_options)
|
|
tool_opts = _validated_tool_options(tool_options)
|
|
with self._logger.log_operation(
|
|
self, 'trace', portspec, ccw=ccw, length=length, spacing=spacing,
|
|
plan_options=plan_opts, tool_options=tool_opts, **bounds,
|
|
):
|
|
if isinstance(portspec, str):
|
|
portspec = [portspec]
|
|
contexts = self._route_contexts(portspec)
|
|
try:
|
|
result = self.planner.plan_trace_route(
|
|
contexts, ccw, length, spacing=spacing, plan_options=plan_opts,
|
|
tool_options=tool_opts, **bounds,
|
|
)
|
|
except (BuildError, NotImplementedError) as err:
|
|
if not self._dead or route_failure_policy(err) is RouteFailurePolicy.FATAL:
|
|
raise
|
|
if length is not None and len(contexts) == 1:
|
|
context = contexts[0]
|
|
endpoint = self._apply_dead_fallback(
|
|
context.portspec,
|
|
length,
|
|
0,
|
|
ccw,
|
|
context.port.ptype,
|
|
out_ptype = out_ptype,
|
|
)
|
|
self._notify_route_complete(((context.portspec, endpoint),))
|
|
return self
|
|
if bounds.get('each') is not None:
|
|
each = bounds['each']
|
|
endpoints: list[tuple[str, Port]] = []
|
|
for context in contexts:
|
|
endpoint = self._apply_dead_fallback(
|
|
context.portspec,
|
|
each,
|
|
0,
|
|
ccw,
|
|
context.port.ptype,
|
|
out_ptype = out_ptype,
|
|
)
|
|
endpoints.append((context.portspec, endpoint))
|
|
self._notify_route_complete(endpoints)
|
|
return self
|
|
raise
|
|
self._apply_route_result(result)
|
|
return self
|
|
|
|
def trace_to(
|
|
self,
|
|
portspec: str | Sequence[str],
|
|
ccw: SupportsBool | None,
|
|
*,
|
|
length: float | None = None,
|
|
spacing: float | ArrayLike | None = None,
|
|
plan_options: Mapping[str, Any] | None = None,
|
|
out_ptype: str | None = None,
|
|
each: float | None = None,
|
|
set_rotation: float | None = None,
|
|
p: float | None = None,
|
|
pos: float | None = None,
|
|
position: float | None = None,
|
|
x: float | None = None,
|
|
y: float | None = None,
|
|
emin: float | None = None,
|
|
emax: float | None = None,
|
|
pmin: float | None = None,
|
|
pmax: float | None = None,
|
|
xmin: float | None = None,
|
|
xmax: float | None = None,
|
|
ymin: float | None = None,
|
|
ymax: float | None = None,
|
|
min_past_furthest: float | None = None,
|
|
tool_options: Mapping[str, Any] | None = None,
|
|
) -> Self:
|
|
"""
|
|
Route until a single positional bound is reached, or delegate to `trace()` for length/bundle bounds.
|
|
|
|
Exactly one of `p`, `pos`, `position`, `x`, or `y` may be used as a positional
|
|
bound. Positional bounds are only valid for a single port and may not be combined
|
|
with `length`, `spacing`, `each`, or bundle-bound keywords such as `xmin`/`emax`.
|
|
|
|
With no positional or bundle bound, single-port `trace_to()` uses the
|
|
same omitted minimum-length primitive-offer behavior as `trace()`.
|
|
Planner-specific per-route settings belong in `plan_options`.
|
|
"""
|
|
bounds = _present_route_args(
|
|
length=length,
|
|
out_ptype=out_ptype,
|
|
each=each,
|
|
set_rotation=set_rotation,
|
|
p=p,
|
|
pos=pos,
|
|
position=position,
|
|
x=x,
|
|
y=y,
|
|
emin=emin,
|
|
emax=emax,
|
|
pmin=pmin,
|
|
pmax=pmax,
|
|
xmin=xmin,
|
|
xmax=xmax,
|
|
ymin=ymin,
|
|
ymax=ymax,
|
|
min_past_furthest=min_past_furthest,
|
|
)
|
|
plan_opts = _validated_plan_options(plan_options)
|
|
tool_opts = _validated_tool_options(tool_options)
|
|
with self._logger.log_operation(
|
|
self, 'trace_to', portspec, ccw=ccw, spacing=spacing,
|
|
plan_options=plan_opts, tool_options=tool_opts, **bounds,
|
|
):
|
|
if isinstance(portspec, str):
|
|
portspec = [portspec]
|
|
contexts = self._route_contexts(portspec)
|
|
try:
|
|
result = self.planner.plan_trace_to_route(
|
|
contexts, ccw, spacing=spacing, plan_options=plan_opts,
|
|
tool_options=tool_opts, **bounds,
|
|
)
|
|
except (BuildError, NotImplementedError) as err:
|
|
if (
|
|
not self._dead
|
|
or len(contexts) != 1
|
|
or route_failure_policy(err) is RouteFailurePolicy.FATAL
|
|
):
|
|
raise
|
|
if bounds.get('length') is not None:
|
|
length = bounds['length']
|
|
else:
|
|
resolved = resolved_position_bound(
|
|
contexts[0].port,
|
|
bounds,
|
|
allow_length=False,
|
|
)
|
|
if resolved is None:
|
|
raise
|
|
_key, _value, length = resolved
|
|
context = contexts[0]
|
|
endpoint = self._apply_dead_fallback(
|
|
context.portspec,
|
|
length,
|
|
0,
|
|
ccw,
|
|
context.port.ptype,
|
|
out_ptype = out_ptype,
|
|
)
|
|
self._notify_route_complete(((context.portspec, endpoint),))
|
|
return self
|
|
self._apply_route_result(result)
|
|
return self
|
|
|
|
def straight(
|
|
self,
|
|
portspec: str | Sequence[str],
|
|
length: float | None = None,
|
|
*,
|
|
spacing: float | ArrayLike | None = None,
|
|
plan_options: Mapping[str, Any] | None = None,
|
|
out_ptype: str | None = None,
|
|
each: float | None = None,
|
|
set_rotation: float | None = None,
|
|
p: float | None = None,
|
|
pos: float | None = None,
|
|
position: float | None = None,
|
|
x: float | None = None,
|
|
y: float | None = None,
|
|
emin: float | None = None,
|
|
emax: float | None = None,
|
|
pmin: float | None = None,
|
|
pmax: float | None = None,
|
|
xmin: float | None = None,
|
|
xmax: float | None = None,
|
|
ymin: float | None = None,
|
|
ymax: float | None = None,
|
|
min_past_furthest: float | None = None,
|
|
tool_options: Mapping[str, Any] | None = None,
|
|
) -> Self:
|
|
return self.trace_to(
|
|
portspec, None, length=length, spacing=spacing, plan_options=plan_options,
|
|
out_ptype=out_ptype, each=each, set_rotation=set_rotation,
|
|
p=p, pos=pos, position=position, x=x, y=y,
|
|
emin=emin, emax=emax, pmin=pmin, pmax=pmax,
|
|
xmin=xmin, xmax=xmax, ymin=ymin, ymax=ymax,
|
|
min_past_furthest=min_past_furthest, tool_options=tool_options,
|
|
)
|
|
|
|
def bend(
|
|
self,
|
|
portspec: str | Sequence[str],
|
|
ccw: SupportsBool,
|
|
length: float | None = None,
|
|
*,
|
|
spacing: float | ArrayLike | None = None,
|
|
plan_options: Mapping[str, Any] | None = None,
|
|
out_ptype: str | None = None,
|
|
each: float | None = None,
|
|
set_rotation: float | None = None,
|
|
p: float | None = None,
|
|
pos: float | None = None,
|
|
position: float | None = None,
|
|
x: float | None = None,
|
|
y: float | None = None,
|
|
emin: float | None = None,
|
|
emax: float | None = None,
|
|
pmin: float | None = None,
|
|
pmax: float | None = None,
|
|
xmin: float | None = None,
|
|
xmax: float | None = None,
|
|
ymin: float | None = None,
|
|
ymax: float | None = None,
|
|
min_past_furthest: float | None = None,
|
|
tool_options: Mapping[str, Any] | None = None,
|
|
) -> Self:
|
|
return self.trace_to(
|
|
portspec, ccw, length=length, spacing=spacing, plan_options=plan_options,
|
|
out_ptype=out_ptype, each=each, set_rotation=set_rotation,
|
|
p=p, pos=pos, position=position, x=x, y=y,
|
|
emin=emin, emax=emax, pmin=pmin, pmax=pmax,
|
|
xmin=xmin, xmax=xmax, ymin=ymin, ymax=ymax,
|
|
min_past_furthest=min_past_furthest, tool_options=tool_options,
|
|
)
|
|
|
|
def ccw(
|
|
self,
|
|
portspec: str | Sequence[str],
|
|
length: float | None = None,
|
|
*,
|
|
spacing: float | ArrayLike | None = None,
|
|
plan_options: Mapping[str, Any] | None = None,
|
|
out_ptype: str | None = None,
|
|
each: float | None = None,
|
|
set_rotation: float | None = None,
|
|
p: float | None = None,
|
|
pos: float | None = None,
|
|
position: float | None = None,
|
|
x: float | None = None,
|
|
y: float | None = None,
|
|
emin: float | None = None,
|
|
emax: float | None = None,
|
|
pmin: float | None = None,
|
|
pmax: float | None = None,
|
|
xmin: float | None = None,
|
|
xmax: float | None = None,
|
|
ymin: float | None = None,
|
|
ymax: float | None = None,
|
|
min_past_furthest: float | None = None,
|
|
tool_options: Mapping[str, Any] | None = None,
|
|
) -> Self:
|
|
return self.bend(
|
|
portspec, True, length, spacing=spacing, plan_options=plan_options, out_ptype=out_ptype,
|
|
each=each, set_rotation=set_rotation, p=p, pos=pos, position=position, x=x, y=y,
|
|
emin=emin, emax=emax, pmin=pmin, pmax=pmax, xmin=xmin, xmax=xmax, ymin=ymin, ymax=ymax,
|
|
min_past_furthest=min_past_furthest, tool_options=tool_options,
|
|
)
|
|
|
|
def cw(
|
|
self,
|
|
portspec: str | Sequence[str],
|
|
length: float | None = None,
|
|
*,
|
|
spacing: float | ArrayLike | None = None,
|
|
plan_options: Mapping[str, Any] | None = None,
|
|
out_ptype: str | None = None,
|
|
each: float | None = None,
|
|
set_rotation: float | None = None,
|
|
p: float | None = None,
|
|
pos: float | None = None,
|
|
position: float | None = None,
|
|
x: float | None = None,
|
|
y: float | None = None,
|
|
emin: float | None = None,
|
|
emax: float | None = None,
|
|
pmin: float | None = None,
|
|
pmax: float | None = None,
|
|
xmin: float | None = None,
|
|
xmax: float | None = None,
|
|
ymin: float | None = None,
|
|
ymax: float | None = None,
|
|
min_past_furthest: float | None = None,
|
|
tool_options: Mapping[str, Any] | None = None,
|
|
) -> Self:
|
|
return self.bend(
|
|
portspec, False, length, spacing=spacing, plan_options=plan_options, out_ptype=out_ptype,
|
|
each=each, set_rotation=set_rotation, p=p, pos=pos, position=position, x=x, y=y,
|
|
emin=emin, emax=emax, pmin=pmin, pmax=pmax, xmin=xmin, xmax=xmax, ymin=ymin, ymax=ymax,
|
|
min_past_furthest=min_past_furthest, tool_options=tool_options,
|
|
)
|
|
|
|
def jog(
|
|
self,
|
|
portspec: str | Sequence[str],
|
|
offset: float,
|
|
length: float | None = None,
|
|
*,
|
|
spacing: float | ArrayLike | None = None,
|
|
plan_options: Mapping[str, Any] | None = None,
|
|
out_ptype: str | None = None,
|
|
p: float | None = None,
|
|
pos: float | None = None,
|
|
position: float | None = None,
|
|
x: float | None = None,
|
|
y: float | None = None,
|
|
tool_options: Mapping[str, Any] | None = None,
|
|
) -> Self:
|
|
"""
|
|
Route an S-bend.
|
|
|
|
`length` is the along-travel displacement. If omitted and no positional
|
|
bound is supplied, a single-port jog evaluates legal S-like candidates
|
|
at their minimum legal length or primitive endpoint length for the
|
|
requested offset, then cost selects among those candidates. If exactly
|
|
one positional bound (`p`, `pos`, `position`, `x`, or `y`) is supplied,
|
|
the required travel distance is derived from that bound.
|
|
|
|
Multi-port jogs require `spacing`; the innermost first-bend port uses
|
|
the base `length` or omitted-length solve, and other ports derive exact
|
|
route lengths and offsets from that base route. `out_ptype`, when
|
|
provided, constrains only each final route endpoint. Planner-specific
|
|
per-route settings belong in `plan_options`.
|
|
"""
|
|
bounds = _present_route_args(out_ptype=out_ptype, p=p, pos=pos, position=position, x=x, y=y)
|
|
plan_opts = _validated_plan_options(plan_options)
|
|
tool_opts = _validated_tool_options(tool_options)
|
|
with self._logger.log_operation(
|
|
self, 'jog', portspec, offset=offset, length=length, spacing=spacing,
|
|
plan_options=plan_opts, tool_options=tool_opts, **bounds,
|
|
):
|
|
if isinstance(portspec, str):
|
|
portspec = [portspec]
|
|
contexts = self._route_contexts(portspec)
|
|
try:
|
|
result = self.planner.plan_jog_route(
|
|
contexts, offset, length, spacing=spacing, plan_options=plan_opts,
|
|
tool_options=tool_opts, **bounds,
|
|
)
|
|
except (BuildError, NotImplementedError) as err:
|
|
if (
|
|
not self._dead
|
|
or len(contexts) != 1
|
|
or route_failure_policy(err) is RouteFailurePolicy.FATAL
|
|
):
|
|
raise
|
|
if numpy.isclose(offset, 0):
|
|
if length is None:
|
|
raise
|
|
context = contexts[0]
|
|
endpoint = self._apply_dead_fallback(
|
|
context.portspec,
|
|
length,
|
|
0,
|
|
None,
|
|
context.port.ptype,
|
|
out_ptype = out_ptype,
|
|
)
|
|
self._notify_route_complete(((context.portspec, endpoint),))
|
|
return self
|
|
fallback_length = length if length is not None else 0
|
|
context = contexts[0]
|
|
endpoint = self._apply_dead_fallback(
|
|
context.portspec,
|
|
fallback_length,
|
|
offset,
|
|
None,
|
|
context.port.ptype,
|
|
out_rot = pi,
|
|
out_ptype = out_ptype,
|
|
)
|
|
self._notify_route_complete(((context.portspec, endpoint),))
|
|
return self
|
|
self._apply_route_result(result)
|
|
return self
|
|
|
|
def uturn(
|
|
self,
|
|
portspec: str | Sequence[str],
|
|
offset: float,
|
|
length: float | None = None,
|
|
*,
|
|
spacing: float | ArrayLike | None = None,
|
|
plan_options: Mapping[str, Any] | None = None,
|
|
out_ptype: str | None = None,
|
|
tool_options: Mapping[str, Any] | None = None,
|
|
) -> Self:
|
|
"""
|
|
Route a U-turn.
|
|
|
|
`length` is the along-travel displacement to the final port. If omitted,
|
|
legal U-like candidates are evaluated at their minimum legal length or
|
|
primitive endpoint length for the requested offset, then cost selects
|
|
among those candidates. Multi-port U-turns require nonzero `offset` and
|
|
`spacing`; the innermost first-bend port supplies the base route and
|
|
other ports derive exact lengths and offsets from it. Use `length=0` to
|
|
request the old zero-public-length U-turn shape. Positional and
|
|
bundle-bound keywords are not supported for this operation. `out_ptype`,
|
|
when provided, constrains only each final route endpoint. Planner-specific
|
|
per-route settings belong in `plan_options`.
|
|
"""
|
|
bounds = _present_route_args(out_ptype=out_ptype)
|
|
plan_opts = _validated_plan_options(plan_options)
|
|
tool_opts = _validated_tool_options(tool_options)
|
|
with self._logger.log_operation(
|
|
self, 'uturn', portspec, offset=offset, length=length, spacing=spacing,
|
|
plan_options=plan_opts, tool_options=tool_opts, **bounds,
|
|
):
|
|
if isinstance(portspec, str):
|
|
portspec = [portspec]
|
|
contexts = self._route_contexts(portspec)
|
|
try:
|
|
result = self.planner.plan_uturn_route(
|
|
contexts, offset, length, spacing=spacing, plan_options=plan_opts,
|
|
tool_options=tool_opts, **bounds,
|
|
)
|
|
except (BuildError, NotImplementedError) as err:
|
|
if (
|
|
not self._dead
|
|
or len(contexts) != 1
|
|
or length is None
|
|
or route_failure_policy(err) is RouteFailurePolicy.FATAL
|
|
):
|
|
raise
|
|
context = contexts[0]
|
|
endpoint = self._apply_dead_fallback(
|
|
context.portspec,
|
|
length,
|
|
offset,
|
|
None,
|
|
context.port.ptype,
|
|
out_rot = 0.0,
|
|
out_ptype = out_ptype,
|
|
)
|
|
self._notify_route_complete(((context.portspec, endpoint),))
|
|
return self
|
|
self._apply_route_result(result)
|
|
return self
|
|
|
|
def trace_into(
|
|
self,
|
|
portspec_src: str,
|
|
portspec_dst: str,
|
|
*,
|
|
out_ptype: str | None = None,
|
|
plug_destination: bool = True,
|
|
thru: str | None = None,
|
|
plan_options: Mapping[str, Any] | None = None,
|
|
tool_options: Mapping[str, Any] | None = None,
|
|
) -> Self:
|
|
"""
|
|
Route one port into another using a bounded primitive-offer selection.
|
|
|
|
By default, searches only the exact main-route bend count required by
|
|
the endpoint relationship: zero for a straight, one for a quarter-turn,
|
|
and two for S- and U-like connections. This rejects extra dogleg and
|
|
loop-like fallback routes without inspecting primitive geometry. Ptype
|
|
adapters do not consume this bend budget.
|
|
|
|
Set `plan_options={'bend_policy': 'flexible'}` to search bounded
|
|
primitive-offer routes with up to four bend roles. Bend-family requests try one-bend routes
|
|
before three-bend routes; other families try zero-to-two-bend routes
|
|
before four-bend routes. The first band with a legal candidate wins.
|
|
Within a band, candidates are ordered by total cost, adapter count,
|
|
step count, the requested straight-vs-turn topology preference, and
|
|
deterministic discovery order. The default planner's `strategy` option
|
|
therefore affects only otherwise tied candidates.
|
|
Custom planning options may be supplied through `tool_options`; they
|
|
are forwarded only to primitive offer generation.
|
|
|
|
If `plug_destination` is `True`, the destination port is consumed by the final step.
|
|
If `thru` is provided, that port is renamed to the source name after the route is complete.
|
|
`out_ptype` constrains only the final route endpoint. Route selection
|
|
failures occur before live port state and deferred routing steps are
|
|
mutated; failures during selected-route execution, including primitive
|
|
commit, plug/thru application, or render, may leave partial output.
|
|
"""
|
|
plan_opts = _validated_plan_options(plan_options)
|
|
tool_opts = _validated_tool_options(tool_options)
|
|
with self._logger.log_operation(
|
|
self,
|
|
'trace_into',
|
|
[portspec_src, portspec_dst],
|
|
out_ptype=out_ptype,
|
|
plug_destination=plug_destination,
|
|
thru=thru,
|
|
plan_options=plan_opts,
|
|
tool_options=tool_opts,
|
|
):
|
|
result = self.planner.plan_trace_into(
|
|
self._route_context(portspec_src),
|
|
portspec_dst,
|
|
self.pattern[portspec_dst].copy(),
|
|
out_ptype = out_ptype,
|
|
plug_destination = plug_destination,
|
|
thru = thru,
|
|
plan_options = plan_opts,
|
|
tool_options = tool_opts,
|
|
)
|
|
self._apply_route_result(result)
|
|
return self
|
|
|
|
#
|
|
# Rendering
|
|
#
|
|
def render(self, append: bool = True) -> Self:
|
|
"""
|
|
Generate geometry for all pending render steps.
|
|
|
|
Consecutive compatible `RenderStep`s are batched by port and Tool, then
|
|
passed to `Tool.render()`. After insertion, the rendered output port is
|
|
checked against the endpoint that planning selected.
|
|
|
|
Rendering may modify the Library before every batch has completed and
|
|
does not provide rollback. If this method raises, the Pather must be
|
|
treated as unusable; retrying or continuing to route with it is
|
|
unsupported.
|
|
"""
|
|
with self._logger.log_operation(self, 'render', None, append=append):
|
|
tool_port_names = ('A', 'B')
|
|
pat = Pattern()
|
|
|
|
def validate_tree(portspec: str, batch: list[RenderStep], tree: ILibrary) -> None:
|
|
missing = sorted(
|
|
name
|
|
for name in tree.dangling_refs(tree.top())
|
|
if isinstance(name, str) and name.startswith(SINGLE_USE_PREFIX)
|
|
)
|
|
if not missing:
|
|
return
|
|
|
|
tool_name = type(batch[0].tool).__name__
|
|
raise ToolContractError(
|
|
f'Tool {tool_name}.render() returned missing single-use refs for {portspec}: {missing}'
|
|
)
|
|
|
|
def validate_rendered_endpoint(portspec: str, batch: list[RenderStep]) -> None:
|
|
expected = batch[-1].end_port
|
|
actual = pat.ports.get(portspec)
|
|
tool_name = type(batch[0].tool).__name__
|
|
if actual is None:
|
|
raise ToolContractError(
|
|
f'Tool {tool_name}.render() did not produce output port {portspec!r}; '
|
|
f'expected {expected.describe()}'
|
|
)
|
|
|
|
offsets_match = array_close(actual.offset, expected.offset)
|
|
rotations_match = (
|
|
actual.rotation is None
|
|
or expected.rotation is None
|
|
or angles_equal(actual.rotation, expected.rotation)
|
|
)
|
|
ptypes_match = ptypes_compatible(actual.ptype, expected.ptype)
|
|
if offsets_match and rotations_match and ptypes_match:
|
|
return
|
|
|
|
raise ToolContractError(
|
|
f'Tool {tool_name}.render() output port {portspec!r} does not match planned endpoint: '
|
|
f'expected {expected.describe()}, got {actual.describe()}'
|
|
)
|
|
|
|
def render_batch(portspec: str, batch: list[RenderStep], append: bool) -> None:
|
|
assert batch[0].tool is not None
|
|
tree = batch[0].tool.render(batch, port_names=tool_port_names)
|
|
validate_tree(portspec, batch, tree)
|
|
name = self.library << tree
|
|
try:
|
|
if portspec in pat.ports:
|
|
del pat.ports[portspec]
|
|
pat.ports[portspec] = batch[0].start_port.copy()
|
|
if append:
|
|
pat.plug(self.library[name], {portspec: tool_port_names[0]}, append=True)
|
|
del self.library[name]
|
|
else:
|
|
pat.plug(self.library.abstract(name), {portspec: tool_port_names[0]}, append=False)
|
|
if portspec not in pat.ports and tool_port_names[1] in pat.ports:
|
|
pat.rename_ports({tool_port_names[1]: portspec}, overwrite=True)
|
|
validate_rendered_endpoint(portspec, batch)
|
|
except Exception:
|
|
if name in self.library:
|
|
del self.library[name]
|
|
raise
|
|
|
|
for portspec, steps in self._paths.items():
|
|
if not steps:
|
|
continue
|
|
batch: list[RenderStep] = []
|
|
for step in steps:
|
|
appendable = step.kind != 'plug'
|
|
same_tool = batch and step.tool is batch[0].tool
|
|
if batch and (not appendable or not same_tool or not batch[-1].is_continuous_with(step)):
|
|
render_batch(portspec, batch, append)
|
|
batch = []
|
|
if appendable:
|
|
batch.append(step)
|
|
elif step.kind == 'plug' and portspec in pat.ports:
|
|
del pat.ports[portspec]
|
|
if batch:
|
|
render_batch(portspec, batch, append)
|
|
|
|
self._paths.clear()
|
|
pat.ports.clear()
|
|
self.pattern.append(pat)
|
|
return self
|
|
|
|
#
|
|
# Utilities
|
|
#
|
|
@classmethod
|
|
def interface(
|
|
cls,
|
|
source: PortList | Mapping[str, Port] | str,
|
|
*,
|
|
library: ILibrary | None = None,
|
|
tools: Tool | MutableMapping[str | None, Tool] | None = None,
|
|
in_prefix: str = 'in_',
|
|
out_prefix: str = '',
|
|
port_map: dict[str, str] | Sequence[str] | None = None,
|
|
name: str | None = None,
|
|
**kwargs: Any,
|
|
) -> Self:
|
|
if library is None:
|
|
if hasattr(source, 'library') and isinstance(source.library, ILibrary):
|
|
library = source.library
|
|
else:
|
|
raise BuildError('No library provided')
|
|
if tools is None and hasattr(source, 'tools') and isinstance(source.tools, dict):
|
|
tools = source.tools
|
|
if isinstance(source, str):
|
|
source = library.abstract(source).ports
|
|
pat = Pattern.interface(source, in_prefix=in_prefix, out_prefix=out_prefix, port_map=port_map)
|
|
return cls(library=library, pattern=pat, name=name, tools=tools, **kwargs)
|
|
|
|
def retool(self, tool: Tool, keys: str | Sequence[str | None] | None = None) -> Self:
|
|
if keys is None or isinstance(keys, str):
|
|
self.tools[keys] = tool
|
|
else:
|
|
for k in keys:
|
|
self.tools[k] = tool
|
|
return self
|
|
|
|
@contextmanager
|
|
def toolctx(self, tool: Tool, keys: str | Sequence[str | None] | None = None) -> Iterator[Self]:
|
|
if keys is None or isinstance(keys, str):
|
|
keys = [keys]
|
|
saved = {k: self.tools.get(k) for k in keys}
|
|
try:
|
|
yield self.retool(tool, keys)
|
|
finally:
|
|
for k, t in saved.items():
|
|
if t is None:
|
|
self.tools.pop(k, None)
|
|
else:
|
|
self.tools[k] = t
|
|
|
|
def flatten(self) -> Self:
|
|
self.pattern.flatten(self.library)
|
|
return self
|
|
|
|
def at(
|
|
self,
|
|
portspec: str | Iterable[str],
|
|
*,
|
|
spacing: float | ArrayLike | None = None,
|
|
) -> 'PortPather':
|
|
return PortPather(portspec, self, default_spacing=spacing)
|
|
|
|
|
|
class PortPather:
|
|
"""
|
|
Port-name selection for fluent pathing.
|
|
|
|
The selection stores names, not stable physical port identities. Its own
|
|
rename/delete helpers update the selection, but unrelated changes made
|
|
through the parent Pather do not retarget it.
|
|
"""
|
|
def __init__(
|
|
self,
|
|
ports: str | Iterable[str],
|
|
pather: Pather,
|
|
*,
|
|
default_spacing: float | ArrayLike | None = None,
|
|
) -> None:
|
|
self.ports = [ports] if isinstance(ports, str) else list(ports)
|
|
self.pather = pather
|
|
self.default_spacing = default_spacing
|
|
|
|
def _single_port(self, action: str) -> str:
|
|
"""Return the selected port for an exact-one operation."""
|
|
if len(self.ports) != 1:
|
|
raise BuildError(
|
|
f'Unable to use implicit {action}() with {len(self.ports)} ports; expected exactly one.'
|
|
)
|
|
return self.ports[0]
|
|
|
|
def retool(self, tool: Tool) -> Self:
|
|
self.pather.retool(tool, self.ports)
|
|
return self
|
|
|
|
def set_spacing(self, spacing: float | ArrayLike | None) -> Self:
|
|
self.default_spacing = spacing
|
|
return self
|
|
|
|
@contextmanager
|
|
def toolctx(self, tool: Tool) -> Iterator[Self]:
|
|
with self.pather.toolctx(tool, keys=self.ports):
|
|
yield self
|
|
|
|
def trace(
|
|
self,
|
|
ccw: SupportsBool | None,
|
|
length: float | None = None,
|
|
*,
|
|
spacing: float | ArrayLike | None = None,
|
|
plan_options: Mapping[str, Any] | None = None,
|
|
out_ptype: str | None = None,
|
|
each: float | None = None,
|
|
set_rotation: float | None = None,
|
|
emin: float | None = None,
|
|
emax: float | None = None,
|
|
pmin: float | None = None,
|
|
pmax: float | None = None,
|
|
xmin: float | None = None,
|
|
xmax: float | None = None,
|
|
ymin: float | None = None,
|
|
ymax: float | None = None,
|
|
min_past_furthest: float | None = None,
|
|
tool_options: Mapping[str, Any] | None = None,
|
|
) -> Self:
|
|
if spacing is None and self.default_spacing is not None and len(self.ports) > 1 and ccw is not None:
|
|
spacing = self.default_spacing
|
|
self.pather.trace(
|
|
self.ports, ccw, length, spacing=spacing, plan_options=plan_options, out_ptype=out_ptype,
|
|
each=each, set_rotation=set_rotation, emin=emin, emax=emax, pmin=pmin, pmax=pmax,
|
|
xmin=xmin, xmax=xmax, ymin=ymin, ymax=ymax, min_past_furthest=min_past_furthest,
|
|
tool_options=tool_options,
|
|
)
|
|
return self
|
|
|
|
def trace_to(
|
|
self,
|
|
ccw: SupportsBool | None,
|
|
*,
|
|
length: float | None = None,
|
|
spacing: float | ArrayLike | None = None,
|
|
plan_options: Mapping[str, Any] | None = None,
|
|
out_ptype: str | None = None,
|
|
each: float | None = None,
|
|
set_rotation: float | None = None,
|
|
p: float | None = None,
|
|
pos: float | None = None,
|
|
position: float | None = None,
|
|
x: float | None = None,
|
|
y: float | None = None,
|
|
emin: float | None = None,
|
|
emax: float | None = None,
|
|
pmin: float | None = None,
|
|
pmax: float | None = None,
|
|
xmin: float | None = None,
|
|
xmax: float | None = None,
|
|
ymin: float | None = None,
|
|
ymax: float | None = None,
|
|
min_past_furthest: float | None = None,
|
|
tool_options: Mapping[str, Any] | None = None,
|
|
) -> Self:
|
|
if spacing is None and self.default_spacing is not None and len(self.ports) > 1 and ccw is not None:
|
|
spacing = self.default_spacing
|
|
self.pather.trace_to(
|
|
self.ports, ccw, length=length, spacing=spacing, plan_options=plan_options, out_ptype=out_ptype,
|
|
each=each, set_rotation=set_rotation, p=p, pos=pos, position=position, x=x, y=y,
|
|
emin=emin, emax=emax, pmin=pmin, pmax=pmax, xmin=xmin, xmax=xmax, ymin=ymin, ymax=ymax,
|
|
min_past_furthest=min_past_furthest, tool_options=tool_options,
|
|
)
|
|
return self
|
|
|
|
def straight(
|
|
self,
|
|
length: float | None = None,
|
|
*,
|
|
spacing: float | ArrayLike | None = None,
|
|
plan_options: Mapping[str, Any] | None = None,
|
|
out_ptype: str | None = None,
|
|
each: float | None = None,
|
|
set_rotation: float | None = None,
|
|
p: float | None = None,
|
|
pos: float | None = None,
|
|
position: float | None = None,
|
|
x: float | None = None,
|
|
y: float | None = None,
|
|
emin: float | None = None,
|
|
emax: float | None = None,
|
|
pmin: float | None = None,
|
|
pmax: float | None = None,
|
|
xmin: float | None = None,
|
|
xmax: float | None = None,
|
|
ymin: float | None = None,
|
|
ymax: float | None = None,
|
|
min_past_furthest: float | None = None,
|
|
tool_options: Mapping[str, Any] | None = None,
|
|
) -> Self:
|
|
return self.trace_to(
|
|
None, length=length, spacing=spacing, plan_options=plan_options, out_ptype=out_ptype,
|
|
each=each, set_rotation=set_rotation, p=p, pos=pos, position=position, x=x, y=y,
|
|
emin=emin, emax=emax, pmin=pmin, pmax=pmax, xmin=xmin, xmax=xmax, ymin=ymin, ymax=ymax,
|
|
min_past_furthest=min_past_furthest, tool_options=tool_options,
|
|
)
|
|
|
|
def bend(
|
|
self,
|
|
ccw: SupportsBool,
|
|
length: float | None = None,
|
|
*,
|
|
spacing: float | ArrayLike | None = None,
|
|
plan_options: Mapping[str, Any] | None = None,
|
|
out_ptype: str | None = None,
|
|
each: float | None = None,
|
|
set_rotation: float | None = None,
|
|
p: float | None = None,
|
|
pos: float | None = None,
|
|
position: float | None = None,
|
|
x: float | None = None,
|
|
y: float | None = None,
|
|
emin: float | None = None,
|
|
emax: float | None = None,
|
|
pmin: float | None = None,
|
|
pmax: float | None = None,
|
|
xmin: float | None = None,
|
|
xmax: float | None = None,
|
|
ymin: float | None = None,
|
|
ymax: float | None = None,
|
|
min_past_furthest: float | None = None,
|
|
tool_options: Mapping[str, Any] | None = None,
|
|
) -> Self:
|
|
return self.trace_to(
|
|
ccw, length=length, spacing=spacing, plan_options=plan_options, out_ptype=out_ptype,
|
|
each=each, set_rotation=set_rotation, p=p, pos=pos, position=position, x=x, y=y,
|
|
emin=emin, emax=emax, pmin=pmin, pmax=pmax, xmin=xmin, xmax=xmax, ymin=ymin, ymax=ymax,
|
|
min_past_furthest=min_past_furthest, tool_options=tool_options,
|
|
)
|
|
|
|
def ccw(
|
|
self,
|
|
length: float | None = None,
|
|
*,
|
|
spacing: float | ArrayLike | None = None,
|
|
plan_options: Mapping[str, Any] | None = None,
|
|
out_ptype: str | None = None,
|
|
each: float | None = None,
|
|
set_rotation: float | None = None,
|
|
p: float | None = None,
|
|
pos: float | None = None,
|
|
position: float | None = None,
|
|
x: float | None = None,
|
|
y: float | None = None,
|
|
emin: float | None = None,
|
|
emax: float | None = None,
|
|
pmin: float | None = None,
|
|
pmax: float | None = None,
|
|
xmin: float | None = None,
|
|
xmax: float | None = None,
|
|
ymin: float | None = None,
|
|
ymax: float | None = None,
|
|
min_past_furthest: float | None = None,
|
|
tool_options: Mapping[str, Any] | None = None,
|
|
) -> Self:
|
|
return self.bend(
|
|
True, length, spacing=spacing, plan_options=plan_options, out_ptype=out_ptype,
|
|
each=each, set_rotation=set_rotation, p=p, pos=pos, position=position, x=x, y=y,
|
|
emin=emin, emax=emax, pmin=pmin, pmax=pmax, xmin=xmin, xmax=xmax, ymin=ymin, ymax=ymax,
|
|
min_past_furthest=min_past_furthest, tool_options=tool_options,
|
|
)
|
|
|
|
def cw(
|
|
self,
|
|
length: float | None = None,
|
|
*,
|
|
spacing: float | ArrayLike | None = None,
|
|
plan_options: Mapping[str, Any] | None = None,
|
|
out_ptype: str | None = None,
|
|
each: float | None = None,
|
|
set_rotation: float | None = None,
|
|
p: float | None = None,
|
|
pos: float | None = None,
|
|
position: float | None = None,
|
|
x: float | None = None,
|
|
y: float | None = None,
|
|
emin: float | None = None,
|
|
emax: float | None = None,
|
|
pmin: float | None = None,
|
|
pmax: float | None = None,
|
|
xmin: float | None = None,
|
|
xmax: float | None = None,
|
|
ymin: float | None = None,
|
|
ymax: float | None = None,
|
|
min_past_furthest: float | None = None,
|
|
tool_options: Mapping[str, Any] | None = None,
|
|
) -> Self:
|
|
return self.bend(
|
|
False, length, spacing=spacing, plan_options=plan_options, out_ptype=out_ptype,
|
|
each=each, set_rotation=set_rotation, p=p, pos=pos, position=position, x=x, y=y,
|
|
emin=emin, emax=emax, pmin=pmin, pmax=pmax, xmin=xmin, xmax=xmax, ymin=ymin, ymax=ymax,
|
|
min_past_furthest=min_past_furthest, tool_options=tool_options,
|
|
)
|
|
|
|
def jog(
|
|
self,
|
|
offset: float,
|
|
length: float | None = None,
|
|
*,
|
|
spacing: float | ArrayLike | None = None,
|
|
plan_options: Mapping[str, Any] | None = None,
|
|
out_ptype: str | None = None,
|
|
p: float | None = None,
|
|
pos: float | None = None,
|
|
position: float | None = None,
|
|
x: float | None = None,
|
|
y: float | None = None,
|
|
tool_options: Mapping[str, Any] | None = None,
|
|
) -> Self:
|
|
if spacing is None and self.default_spacing is not None and len(self.ports) > 1 and not numpy.isclose(offset, 0):
|
|
spacing = self.default_spacing
|
|
self.pather.jog(
|
|
self.ports, offset, length, spacing=spacing, plan_options=plan_options, out_ptype=out_ptype,
|
|
p=p, pos=pos, position=position, x=x, y=y, tool_options=tool_options,
|
|
)
|
|
return self
|
|
|
|
def uturn(
|
|
self,
|
|
offset: float,
|
|
length: float | None = None,
|
|
*,
|
|
spacing: float | ArrayLike | None = None,
|
|
plan_options: Mapping[str, Any] | None = None,
|
|
out_ptype: str | None = None,
|
|
tool_options: Mapping[str, Any] | None = None,
|
|
) -> Self:
|
|
if spacing is None and self.default_spacing is not None and len(self.ports) > 1:
|
|
spacing = self.default_spacing
|
|
self.pather.uturn(
|
|
self.ports, offset, length, spacing=spacing, plan_options=plan_options,
|
|
out_ptype=out_ptype, tool_options=tool_options,
|
|
)
|
|
return self
|
|
|
|
def trace_into(
|
|
self,
|
|
target_port: str,
|
|
*,
|
|
out_ptype: str | None = None,
|
|
plug_destination: bool = True,
|
|
thru: str | None = None,
|
|
plan_options: Mapping[str, Any] | None = None,
|
|
tool_options: Mapping[str, Any] | None = None,
|
|
) -> Self:
|
|
port = self._single_port('trace_into')
|
|
self.pather.trace_into(
|
|
port, target_port, out_ptype=out_ptype, plug_destination=plug_destination,
|
|
thru=thru, plan_options=plan_options, tool_options=tool_options,
|
|
)
|
|
return self
|
|
|
|
def plug(self, other: Abstract | str, other_port: str, **kwargs) -> Self:
|
|
port = self._single_port('plug')
|
|
self.pather.plug(other, {port: other_port}, **kwargs)
|
|
return self
|
|
|
|
def plugged(self, other_port: str | Mapping[str, str]) -> Self:
|
|
if isinstance(other_port, Mapping):
|
|
self.pather.plugged(dict(other_port))
|
|
else:
|
|
port = self._single_port('plugged')
|
|
self.pather.plugged({port: other_port})
|
|
return self
|
|
|
|
#
|
|
# Delegate to port
|
|
#
|
|
# These mutate only the selected live port state. They do not rewrite already planned
|
|
# RenderSteps, so deferred geometry remains as previously planned and only future routing
|
|
# starts from the updated port.
|
|
def set_ptype(self, ptype: str) -> Self:
|
|
for port in self.ports:
|
|
self.pather.pattern[port].set_ptype(ptype)
|
|
return self
|
|
|
|
def translate(self, *args, **kwargs) -> Self:
|
|
for port in self.ports:
|
|
self.pather.pattern[port].translate(*args, **kwargs)
|
|
return self
|
|
|
|
def mirror(self, *args, **kwargs) -> Self:
|
|
for port in self.ports:
|
|
self.pather.pattern[port].mirror(*args, **kwargs)
|
|
return self
|
|
|
|
def rotate(self, rotation: float) -> Self:
|
|
for port in self.ports:
|
|
self.pather.pattern[port].rotate(rotation)
|
|
return self
|
|
|
|
def set_rotation(self, rotation: float | None) -> Self:
|
|
for port in self.ports:
|
|
self.pather.pattern[port].set_rotation(rotation)
|
|
return self
|
|
|
|
def rename(self, name: str | Mapping[str, str | None]) -> Self:
|
|
""" Rename active ports. """
|
|
name_map: dict[str, str | None]
|
|
if isinstance(name, str):
|
|
name_map = {self._single_port('rename'): name}
|
|
else:
|
|
name_map = dict(name)
|
|
self.pather.rename_ports(name_map)
|
|
renamed_ports: list[str] = []
|
|
for port in self.ports:
|
|
renamed = name_map.get(port, port)
|
|
if renamed is not None and renamed not in renamed_ports:
|
|
renamed_ports.append(renamed)
|
|
self.ports = renamed_ports
|
|
return self
|
|
|
|
def select(self, ports: str | Iterable[str]) -> Self:
|
|
""" Add ports to the selection. """
|
|
if isinstance(ports, str):
|
|
ports = [ports]
|
|
for port in ports:
|
|
if port not in self.ports:
|
|
self.ports.append(port)
|
|
return self
|
|
|
|
def deselect(self, ports: str | Iterable[str]) -> Self:
|
|
""" Remove ports from the selection. """
|
|
if isinstance(ports, str):
|
|
ports = [ports]
|
|
ports_set = set(ports)
|
|
self.ports = [pp for pp in self.ports if pp not in ports_set]
|
|
return self
|
|
|
|
def _normalize_copy_map(self, name: str | Mapping[str, str], action: str) -> dict[str, str]:
|
|
if isinstance(name, str):
|
|
name_map = {self._single_port(action): name}
|
|
else:
|
|
name_map = dict(name)
|
|
|
|
missing_selected = set(name_map) - set(self.ports)
|
|
if missing_selected:
|
|
raise PortError(f'Can only {action} selected ports: {missing_selected}')
|
|
|
|
missing_pattern = set(name_map) - set(self.pather.pattern.ports)
|
|
if missing_pattern:
|
|
raise PortError(f'Ports to {action} were not found: {missing_pattern}')
|
|
|
|
if not self.pather._dead:
|
|
targets = list(name_map.values())
|
|
duplicate_targets = {vv for vv in targets if targets.count(vv) > 1}
|
|
if duplicate_targets:
|
|
raise PortError(f'{action.capitalize()} targets would collide: {duplicate_targets}')
|
|
|
|
overwritten = {
|
|
dst for src, dst in name_map.items()
|
|
if dst in self.pather.pattern.ports and dst != src
|
|
}
|
|
if overwritten:
|
|
raise PortError(f'{action.capitalize()} would overwrite existing ports: {overwritten}')
|
|
|
|
return name_map
|
|
|
|
def mark(self, name: str | Mapping[str, str]) -> Self:
|
|
""" Bookmark current port(s). """
|
|
name_map = self._normalize_copy_map(name, 'mark')
|
|
source_ports = {src: self.pather.pattern[src].copy() for src in name_map}
|
|
for src, dst in name_map.items():
|
|
self.pather.pattern.ports[dst] = source_ports[src].copy()
|
|
return self
|
|
|
|
def fork(self, name: str | Mapping[str, str]) -> Self:
|
|
""" Split and follow new name. """
|
|
name_map = self._normalize_copy_map(name, 'fork')
|
|
source_ports = {src: self.pather.pattern[src].copy() for src in name_map}
|
|
for src, dst in name_map.items():
|
|
self.pather.pattern.ports[dst] = source_ports[src].copy()
|
|
self.ports = [(dst if pp == src else pp) for pp in self.ports]
|
|
self.ports = list(dict.fromkeys(self.ports))
|
|
return self
|
|
|
|
def drop(self) -> Self:
|
|
""" Remove selected ports from the pattern and the PortPather. """
|
|
self.pather.rename_ports(dict.fromkeys(self.ports))
|
|
self.ports = []
|
|
return self
|
|
|
|
@overload
|
|
def delete(self, name: None) -> None: ...
|
|
|
|
@overload
|
|
def delete(self, name: str) -> Self: ...
|
|
|
|
def delete(self, name: str | None = None) -> Self | None:
|
|
if name is None:
|
|
self.drop()
|
|
return None
|
|
self.pather.rename_ports({name: None})
|
|
self.ports = [pp for pp in self.ports if pp != name]
|
|
return self
|