935 lines
36 KiB
Python
935 lines
36 KiB
Python
"""
|
|
Unified Pattern assembly and routing (`Pather`)
|
|
"""
|
|
from typing import Self, Literal, Any, 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 itertools import chain
|
|
from contextlib import contextmanager
|
|
|
|
import numpy
|
|
from numpy import pi
|
|
from numpy.typing import ArrayLike
|
|
|
|
from ..pattern import Pattern
|
|
from ..library import ILibrary, TreeView
|
|
from ..error import BuildError, PortError
|
|
from ..ports import PortList, Port
|
|
from ..abstract import Abstract
|
|
from ..utils import SupportsBool
|
|
from .tools import Tool, RenderStep
|
|
from .utils import ell
|
|
from .logging import logged_op, PatherLogger
|
|
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
|
|
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.) are by default
|
|
deferred: they record the intended path but do not immediately generate
|
|
geometry. `render()` must be called to generate the final layout.
|
|
Alternatively, setting `auto_render=True` in the constructor will
|
|
cause geometry to be generated incrementally after each routing step.
|
|
|
|
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 `auto_render=True`, geometry is added
|
|
immediately. Otherwise, call `pather.render()` later.
|
|
"""
|
|
__slots__ = (
|
|
'pattern', 'library', 'tools', 'paths',
|
|
'_dead', '_logger', '_auto_render', '_auto_render_append'
|
|
)
|
|
|
|
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`.
|
|
"""
|
|
|
|
paths: defaultdict[str, list[RenderStep]]
|
|
""" Per-port list of planned operations, to be used by `render()` """
|
|
|
|
_dead: bool
|
|
""" If True, geometry generation is skipped (for debugging) """
|
|
|
|
_logger: PatherLogger
|
|
""" Handles diagnostic logging of operations """
|
|
|
|
_auto_render: bool
|
|
""" If True, routing operations call render() immediately """
|
|
|
|
PROBE_LENGTH: float = 1e6
|
|
""" Large length used when probing tools for their lateral displacement """
|
|
|
|
@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,
|
|
auto_render: bool = False,
|
|
auto_render_append: bool = True,
|
|
) -> 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.
|
|
auto_render: If True, enables immediate rendering of routing steps.
|
|
auto_render_append: If `auto_render` is True, determines whether
|
|
to append geometry or add a reference.
|
|
"""
|
|
self._dead = False
|
|
self._logger = PatherLogger(debug=debug)
|
|
self._auto_render = auto_render
|
|
self._auto_render_append = auto_render_append
|
|
self.library = library
|
|
self.pattern = pattern if pattern is not None else Pattern()
|
|
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 __del__(self) -> None:
|
|
if any(self.paths.values()):
|
|
logger.warning(f'Pather {self} had unrendered paths', stack_info=True)
|
|
|
|
def __repr__(self) -> str:
|
|
s = f'<Pather {self.pattern} L({len(self.library)}) {pformat(self.tools)}>'
|
|
return s
|
|
|
|
#
|
|
# Core Pattern Operations (Immediate)
|
|
#
|
|
def _record_break(self, names: Iterable[str | None]) -> None:
|
|
""" Record a batch-breaking step for the specified ports. """
|
|
if not self._dead:
|
|
for n in names:
|
|
if n is not None and n in self.paths:
|
|
port = self.ports[n]
|
|
self.paths[n].append(RenderStep('P', None, port.copy(), port.copy(), None))
|
|
|
|
@logged_op(lambda args: list(args['map_in'].keys()))
|
|
def plug(
|
|
self,
|
|
other: Abstract | str | Pattern | TreeView,
|
|
map_in: dict[str, str],
|
|
map_out: dict[str, str | None] | None = None,
|
|
**kwargs,
|
|
) -> Self:
|
|
if not self._dead:
|
|
other_res = self.library.resolve(other, append=kwargs.get('append', False))
|
|
other_ports = other_res.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)
|
|
self._record_break(affected)
|
|
|
|
# Resolve into Abstract or Pattern
|
|
other = self.library.resolve(other, append=kwargs.get('append', False))
|
|
|
|
self.pattern.plug(other=other, map_in=map_in, map_out=map_out, skip_geometry=self._dead, **kwargs)
|
|
return self
|
|
|
|
@logged_op()
|
|
def place(
|
|
self,
|
|
other: Abstract | str | Pattern | TreeView,
|
|
port_map: dict[str, str | None] | None = None,
|
|
**kwargs,
|
|
) -> Self:
|
|
if not self._dead:
|
|
other_res = self.library.resolve(other, append=kwargs.get('append', False))
|
|
other_ports = other_res.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)
|
|
self._record_break(affected)
|
|
|
|
# Resolve into Abstract or Pattern
|
|
other = self.library.resolve(other, append=kwargs.get('append', False))
|
|
|
|
self.pattern.place(other=other, port_map=port_map, skip_geometry=self._dead, **kwargs)
|
|
return self
|
|
|
|
@logged_op(lambda args: list(args['connections'].keys()))
|
|
def plugged(self, connections: dict[str, str]) -> Self:
|
|
self._record_break(chain(connections.keys(), connections.values()))
|
|
self.pattern.plugged(connections)
|
|
return self
|
|
|
|
@logged_op(lambda args: list(args['mapping'].keys()))
|
|
def rename_ports(self, mapping: dict[str, str | None], overwrite: bool = False) -> Self:
|
|
self.pattern.rename_ports(mapping, overwrite)
|
|
renamed: dict[str, list[RenderStep]] = {vv: self.paths.pop(kk) for kk, vv in mapping.items() if kk in self.paths and vv is not None}
|
|
self.paths.update(renamed)
|
|
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
|
|
|
|
@logged_op(lambda args: list(args['self'].ports.keys()))
|
|
def translate(self, offset: ArrayLike) -> Self:
|
|
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
|
|
|
|
@logged_op(lambda args: list(args['self'].ports.keys()))
|
|
def rotate_around(self, pivot: ArrayLike, angle: float) -> Self:
|
|
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
|
|
|
|
@logged_op(lambda args: list(args['self'].ports.keys()))
|
|
def mirror(self, axis: int = 0) -> Self:
|
|
self.pattern.mirror(axis)
|
|
for steps in self.paths.values():
|
|
for i, step in enumerate(steps):
|
|
steps[i] = step.mirrored(axis)
|
|
return self
|
|
|
|
@logged_op(lambda args: args['name'])
|
|
def mkport(self, name: str, value: Port) -> Self:
|
|
super().mkport(name, value)
|
|
return self
|
|
|
|
#
|
|
# Routing Logic (Deferred / Incremental)
|
|
#
|
|
def _apply_step(
|
|
self,
|
|
opcode: Literal['L', 'S', 'U'],
|
|
portspec: str,
|
|
out_port: Port,
|
|
data: Any,
|
|
tool: Tool,
|
|
plug_into: str | None = None,
|
|
) -> None:
|
|
""" Common logic for applying a planned step to a port. """
|
|
port = self.pattern[portspec]
|
|
port_rot = port.rotation
|
|
assert port_rot is not None
|
|
|
|
out_port.rotate_around((0, 0), pi + port_rot)
|
|
out_port.translate(port.offset)
|
|
|
|
if not self._dead:
|
|
step = RenderStep(opcode, tool, port.copy(), out_port.copy(), data)
|
|
self.paths[portspec].append(step)
|
|
|
|
self.pattern.ports[portspec] = out_port.copy()
|
|
|
|
if plug_into is not None:
|
|
self.plugged({portspec: plug_into})
|
|
|
|
if self._auto_render:
|
|
self.render(append=self._auto_render_append)
|
|
|
|
def _get_tool_R(self, tool: Tool, ccw: SupportsBool, in_ptype: str | None, **kwargs) -> float:
|
|
""" Probe a tool to find the lateral displacement (radius) of its bend. """
|
|
kwargs_no_out = kwargs | {'out_ptype': None}
|
|
probe_len = kwargs.get('probe_length', self.PROBE_LENGTH)
|
|
try:
|
|
out_port, _ = tool.planL(ccw, probe_len, in_ptype=in_ptype, **kwargs_no_out)
|
|
return abs(out_port.y)
|
|
except (BuildError, NotImplementedError):
|
|
# Fallback for tools without planL: use traceL and measure the result
|
|
port_names = ('A', 'B')
|
|
tree = tool.traceL(ccw, probe_len, in_ptype=in_ptype, port_names=port_names, **kwargs_no_out)
|
|
pat = tree.top_pattern()
|
|
(_, R), _ = pat[port_names[0]].measure_travel(pat[port_names[1]])
|
|
return abs(R)
|
|
|
|
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,
|
|
) -> None:
|
|
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
|
|
assert port_rot is not None
|
|
out_port = Port((length, jog), rotation=out_rot, ptype=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})
|
|
|
|
@logged_op(lambda args: args['portspec'])
|
|
def _traceL(self, portspec: str, ccw: SupportsBool | None, length: float, *, plug_into: str | None = None, **kwargs: Any) -> Self:
|
|
tool = self.tools.get(portspec, self.tools.get(None))
|
|
if tool is None:
|
|
raise BuildError(f'No tool assigned for port {portspec}')
|
|
in_ptype = self.pattern[portspec].ptype
|
|
try:
|
|
out_port, data = tool.planL(ccw, length, in_ptype=in_ptype, **kwargs)
|
|
except (BuildError, NotImplementedError):
|
|
if not self._dead:
|
|
raise
|
|
self._apply_dead_fallback(portspec, length, 0, ccw, in_ptype, plug_into)
|
|
return self
|
|
if out_port is not None:
|
|
self._apply_step('L', portspec, out_port, data, tool, plug_into)
|
|
return self
|
|
|
|
@logged_op(lambda args: args['portspec'])
|
|
def _traceS(self, portspec: str, length: float, jog: float, *, plug_into: str | None = None, **kwargs: Any) -> Self:
|
|
tool = self.tools.get(portspec, self.tools.get(None))
|
|
if tool is None:
|
|
raise BuildError(f'No tool assigned for port {portspec}')
|
|
in_ptype = self.pattern[portspec].ptype
|
|
try:
|
|
out_port, data = tool.planS(length, jog, in_ptype=in_ptype, **kwargs)
|
|
except (BuildError, NotImplementedError):
|
|
# Try S-bend fallback (two L-bends)
|
|
ccw0 = jog > 0
|
|
try:
|
|
R1 = self._get_tool_R(tool, ccw0, in_ptype, **kwargs)
|
|
R2 = self._get_tool_R(tool, not ccw0, in_ptype, **kwargs)
|
|
L1, L2 = length - R2, abs(jog) - R1
|
|
except (BuildError, NotImplementedError):
|
|
if not self._dead:
|
|
raise
|
|
self._apply_dead_fallback(portspec, length, jog, None, in_ptype, plug_into, out_rot=pi)
|
|
return self
|
|
|
|
if L1 < 0 or L2 < 0:
|
|
if not self._dead:
|
|
raise BuildError(f"Jog {jog} or length {length} too small for double-L fallback") from None
|
|
self._apply_dead_fallback(portspec, length, jog, None, in_ptype, plug_into, out_rot=pi)
|
|
return self
|
|
|
|
try:
|
|
out_port0, data0 = tool.planL(ccw0, L1, in_ptype=in_ptype, **(kwargs | {'out_ptype': None}))
|
|
out_port1, data1 = tool.planL(not ccw0, L2, in_ptype=out_port0.ptype, **kwargs)
|
|
except (BuildError, NotImplementedError):
|
|
if not self._dead:
|
|
raise
|
|
self._apply_dead_fallback(portspec, length, jog, None, in_ptype, plug_into, out_rot=pi)
|
|
return self
|
|
|
|
self._apply_step('L', portspec, out_port0, data0, tool)
|
|
self._apply_step('L', portspec, out_port1, data1, tool, plug_into)
|
|
return self
|
|
if out_port is not None:
|
|
self._apply_step('S', portspec, out_port, data, tool, plug_into)
|
|
return self
|
|
|
|
@logged_op(lambda args: args['portspec'])
|
|
def _traceU(self, portspec: str, jog: float, *, length: float = 0, plug_into: str | None = None, **kwargs: Any) -> Self:
|
|
tool = self.tools.get(portspec, self.tools.get(None))
|
|
if tool is None:
|
|
raise BuildError(f'No tool assigned for port {portspec}')
|
|
in_ptype = self.pattern[portspec].ptype
|
|
try:
|
|
out_port, data = tool.planU(jog, length=length, in_ptype=in_ptype, **kwargs)
|
|
except (BuildError, NotImplementedError):
|
|
# Try U-turn fallback (two L-bends)
|
|
ccw = jog > 0
|
|
try:
|
|
R = self._get_tool_R(tool, ccw, in_ptype, **kwargs)
|
|
L1, L2 = length + R, abs(jog) - R
|
|
out_port0, data0 = tool.planL(ccw, L1, in_ptype=in_ptype, **(kwargs | {'out_ptype': None}))
|
|
out_port1, data1 = tool.planL(ccw, L2, in_ptype=out_port0.ptype, **kwargs)
|
|
except (BuildError, NotImplementedError):
|
|
if not self._dead:
|
|
raise
|
|
self._apply_dead_fallback(portspec, length, jog, None, in_ptype, plug_into, out_rot=0)
|
|
return self
|
|
else:
|
|
self._apply_step('L', portspec, out_port0, data0, tool)
|
|
self._apply_step('L', portspec, out_port1, data1, tool, plug_into)
|
|
return self
|
|
if out_port is not None:
|
|
self._apply_step('U', portspec, out_port, data, tool, plug_into)
|
|
return self
|
|
|
|
#
|
|
# High-level Routing Methods
|
|
#
|
|
def trace(
|
|
self,
|
|
portspec: str | Sequence[str],
|
|
ccw: SupportsBool | None,
|
|
length: float | None = None,
|
|
*,
|
|
spacing: float | ArrayLike | None = None,
|
|
**bounds: Any,
|
|
) -> Self:
|
|
with self._logger.log_operation(self, 'trace', portspec, ccw=ccw, length=length, spacing=spacing, **bounds):
|
|
if isinstance(portspec, str):
|
|
portspec = [portspec]
|
|
if length is not None:
|
|
if len(portspec) > 1:
|
|
raise BuildError('length only allowed with a single port')
|
|
return self._traceL(portspec[0], ccw, length, **bounds)
|
|
if 'each' in bounds:
|
|
each = bounds.pop('each')
|
|
for p in portspec:
|
|
self._traceL(p, ccw, each, **bounds)
|
|
return self
|
|
# Bundle routing
|
|
bt_keys = {'emin', 'emax', 'pmin', 'pmax', 'xmin', 'xmax', 'ymin', 'ymax', 'min_past_furthest'}
|
|
bt = next((k for k in bounds if k in bt_keys), None)
|
|
if not bt:
|
|
raise BuildError('No bound type specified for trace()')
|
|
bval = bounds.pop(bt)
|
|
set_rot = bounds.pop('set_rotation', None)
|
|
exts = ell(self.pattern[tuple(portspec)], ccw, spacing=spacing, bound=bval, bound_type=bt, set_rotation=set_rot)
|
|
for p, length_val in exts.items():
|
|
self._traceL(p, ccw, length_val, **bounds)
|
|
return self
|
|
|
|
def trace_to(
|
|
self,
|
|
portspec: str | Sequence[str],
|
|
ccw: SupportsBool | None,
|
|
*,
|
|
spacing: float | ArrayLike | None = None,
|
|
**bounds: Any,
|
|
) -> Self:
|
|
with self._logger.log_operation(self, 'trace_to', portspec, ccw=ccw, spacing=spacing, **bounds):
|
|
if isinstance(portspec, str):
|
|
portspec = [portspec]
|
|
pos_keys = {'p', 'x', 'y', 'pos', 'position'}
|
|
pb = {k: bounds[k] for k in bounds if k in pos_keys}
|
|
if pb:
|
|
if len(portspec) > 1:
|
|
raise BuildError('Position bounds only allowed with a single port')
|
|
k, v = next(iter(pb.items()))
|
|
port = self.pattern[portspec[0]]
|
|
assert port.rotation is not None
|
|
is_horiz = numpy.isclose(port.rotation % pi, 0)
|
|
if is_horiz:
|
|
if k == 'y':
|
|
raise BuildError('Port is horizontal')
|
|
target = Port((v, port.offset[1]), rotation=None)
|
|
else:
|
|
if k == 'x':
|
|
raise BuildError('Port is vertical')
|
|
target = Port((port.offset[0], v), rotation=None)
|
|
(travel, jog), _ = port.measure_travel(target)
|
|
other_bounds = {bk: bv for bk, bv in bounds.items() if bk not in pos_keys and bk != 'length'}
|
|
return self._traceL(portspec[0], ccw, -travel, **other_bounds)
|
|
return self.trace(portspec, ccw, spacing=spacing, **bounds)
|
|
|
|
def straight(self, portspec: str | Sequence[str], length: float | None = None, **bounds) -> Self:
|
|
return self.trace_to(portspec, None, length=length, **bounds)
|
|
|
|
def bend(self, portspec: str | Sequence[str], ccw: SupportsBool, length: float | None = None, **bounds) -> Self:
|
|
return self.trace_to(portspec, ccw, length=length, **bounds)
|
|
|
|
def ccw(self, portspec: str | Sequence[str], length: float | None = None, **bounds) -> Self:
|
|
return self.bend(portspec, True, length, **bounds)
|
|
|
|
def cw(self, portspec: str | Sequence[str], length: float | None = None, **bounds) -> Self:
|
|
return self.bend(portspec, False, length, **bounds)
|
|
|
|
def jog(self, portspec: str | Sequence[str], offset: float, length: float | None = None, **bounds: Any) -> Self:
|
|
with self._logger.log_operation(self, 'jog', portspec, offset=offset, length=length, **bounds):
|
|
if isinstance(portspec, str):
|
|
portspec = [portspec]
|
|
for p in portspec:
|
|
self._traceS(p, length, offset, **bounds)
|
|
return self
|
|
|
|
def uturn(self, portspec: str | Sequence[str], offset: float, length: float | None = None, **bounds: Any) -> Self:
|
|
with self._logger.log_operation(self, 'uturn', portspec, offset=offset, length=length, **bounds):
|
|
if isinstance(portspec, str):
|
|
portspec = [portspec]
|
|
for p in portspec:
|
|
self._traceU(p, offset, length=length if length else 0, **bounds)
|
|
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,
|
|
**kwargs: Any,
|
|
) -> Self:
|
|
with self._logger.log_operation(
|
|
self,
|
|
'trace_into',
|
|
[portspec_src, portspec_dst],
|
|
out_ptype=out_ptype,
|
|
plug_destination=plug_destination,
|
|
thru=thru,
|
|
**kwargs,
|
|
):
|
|
if self._dead:
|
|
return self
|
|
port_src, port_dst = self.pattern[portspec_src], self.pattern[portspec_dst]
|
|
if out_ptype is None:
|
|
out_ptype = port_dst.ptype
|
|
if port_src.rotation is None or port_dst.rotation is None:
|
|
raise PortError('Ports must have rotation')
|
|
src_horiz = numpy.isclose(port_src.rotation % pi, 0)
|
|
dst_horiz = numpy.isclose(port_dst.rotation % pi, 0)
|
|
xd, yd = port_dst.offset
|
|
angle = (port_dst.rotation - port_src.rotation) % (2 * pi)
|
|
dst_args = {**kwargs, 'out_ptype': out_ptype}
|
|
if plug_destination:
|
|
dst_args['plug_into'] = portspec_dst
|
|
if src_horiz and not dst_horiz:
|
|
self.trace_to(portspec_src, angle > pi, x=xd, **kwargs)
|
|
self.trace_to(portspec_src, None, y=yd, **dst_args)
|
|
elif dst_horiz and not src_horiz:
|
|
self.trace_to(portspec_src, angle > pi, y=yd, **kwargs)
|
|
self.trace_to(portspec_src, None, x=xd, **dst_args)
|
|
elif numpy.isclose(angle, pi):
|
|
(travel, jog), _ = port_src.measure_travel(port_dst)
|
|
if numpy.isclose(jog, 0):
|
|
self.trace_to(
|
|
portspec_src,
|
|
None,
|
|
x=xd if src_horiz else None,
|
|
y=yd if not src_horiz else None,
|
|
**dst_args,
|
|
)
|
|
else:
|
|
self.jog(portspec_src, -jog, -travel, **dst_args)
|
|
elif numpy.isclose(angle, 0):
|
|
(travel, jog), _ = port_src.measure_travel(port_dst)
|
|
self.uturn(portspec_src, -jog, length=-travel, **dst_args)
|
|
else:
|
|
raise BuildError(f"Cannot route relative angle {angle}")
|
|
if thru:
|
|
self.rename_ports({thru: portspec_src})
|
|
return self
|
|
|
|
#
|
|
# Rendering
|
|
#
|
|
def render(self, append: bool = True) -> Self:
|
|
""" Generate geometry for all planned paths. """
|
|
with self._logger.log_operation(self, 'render', None, append=append):
|
|
tool_port_names = ('A', 'B')
|
|
pat = Pattern()
|
|
|
|
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)
|
|
name = self.library << tree
|
|
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)
|
|
|
|
for portspec, steps in self.paths.items():
|
|
if not steps:
|
|
continue
|
|
batch: list[RenderStep] = []
|
|
for step in steps:
|
|
appendable = step.opcode in ('L', 'S', 'U')
|
|
same_tool = batch and step.tool == 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.opcode == 'P' 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]) -> 'PortPather':
|
|
return PortPather(portspec, self)
|
|
|
|
|
|
class PortPather:
|
|
""" Port state manager for fluent pathing. """
|
|
def __init__(self, ports: str | Iterable[str], pather: Pather) -> None:
|
|
self.ports = [ports] if isinstance(ports, str) else list(ports)
|
|
self.pather = pather
|
|
|
|
def retool(self, tool: Tool) -> Self:
|
|
self.pather.retool(tool, self.ports)
|
|
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, **kw: Any) -> Self:
|
|
self.pather.trace(self.ports, ccw, length, **kw)
|
|
return self
|
|
|
|
def trace_to(self, ccw: SupportsBool | None, **kw: Any) -> Self:
|
|
self.pather.trace_to(self.ports, ccw, **kw)
|
|
return self
|
|
|
|
def straight(self, length: float | None = None, **kw: Any) -> Self:
|
|
return self.trace_to(None, length=length, **kw)
|
|
|
|
def bend(self, ccw: SupportsBool, length: float | None = None, **kw: Any) -> Self:
|
|
return self.trace_to(ccw, length=length, **kw)
|
|
|
|
def ccw(self, length: float | None = None, **kw: Any) -> Self:
|
|
return self.bend(True, length, **kw)
|
|
|
|
def cw(self, length: float | None = None, **kw: Any) -> Self:
|
|
return self.bend(False, length, **kw)
|
|
|
|
def jog(self, offset: float, length: float | None = None, **kw: Any) -> Self:
|
|
self.pather.jog(self.ports, offset, length, **kw)
|
|
return self
|
|
|
|
def uturn(self, offset: float, length: float | None = None, **kw: Any) -> Self:
|
|
self.pather.uturn(self.ports, offset, length, **kw)
|
|
return self
|
|
|
|
def trace_into(self, target_port: str, **kwargs) -> Self:
|
|
if len(self.ports) > 1:
|
|
raise BuildError(f'Unable use implicit trace_into() with {len(self.ports)} (>1) ports.')
|
|
self.pather.trace_into(self.ports[0], target_port, **kwargs)
|
|
return self
|
|
|
|
def plug(self, other: Abstract | str, other_port: str, **kwargs) -> Self:
|
|
if len(self.ports) > 1:
|
|
raise BuildError(f'Unable use implicit plug() with {len(self.ports)} ports.'
|
|
'Use the pather or pattern directly to plug multiple ports.')
|
|
self.pather.plug(other, {self.ports[0]: 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))
|
|
elif len(self.ports) > 1:
|
|
raise BuildError(f'Unable use implicit plugged() with {len(self.ports)} (>1) ports.')
|
|
else:
|
|
self.pather.plugged({self.ports[0]: other_port})
|
|
return self
|
|
|
|
#
|
|
# Delegate to 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):
|
|
if len(self.ports) > 1:
|
|
raise BuildError('Use a mapping to rename >1 port')
|
|
name_map = {self.ports[0]: name}
|
|
else:
|
|
name_map = dict(name)
|
|
self.pather.rename_ports(name_map)
|
|
self.ports = [mm for mm in [name_map.get(pp, pp) for pp in self.ports] if mm is not None]
|
|
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 mark(self, name: str | Mapping[str, str]) -> Self:
|
|
""" Bookmark current port(s). """
|
|
name_map: Mapping[str, str] = {self.ports[0]: name} if isinstance(name, str) else name
|
|
if isinstance(name, str) and len(self.ports) > 1:
|
|
raise BuildError('Use a mapping to mark >1 port')
|
|
for src, dst in name_map.items():
|
|
self.pather.pattern.ports[dst] = self.pather.pattern[src].copy()
|
|
return self
|
|
|
|
def fork(self, name: str | Mapping[str, str]) -> Self:
|
|
""" Split and follow new name. """
|
|
name_map: Mapping[str, str] = {self.ports[0]: name} if isinstance(name, str) else name
|
|
if isinstance(name, str) and len(self.ports) > 1:
|
|
raise BuildError('Use a mapping to fork >1 port')
|
|
for src, dst in name_map.items():
|
|
self.pather.pattern.ports[dst] = self.pather.pattern[src].copy()
|
|
self.ports = [(dst if pp == src else pp) for pp in self.ports]
|
|
return self
|
|
|
|
def drop(self) -> Self:
|
|
""" Remove selected ports from the pattern and the PortPather. """
|
|
for pp in self.ports:
|
|
del self.pather.pattern.ports[pp]
|
|
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
|
|
del self.pather.pattern.ports[name]
|
|
self.ports = [pp for pp in self.ports if pp != name]
|
|
return self
|
|
|
|
|
|
class Builder(Pather):
|
|
"""
|
|
Backward-compatible wrapper for Pather with auto_render=True.
|
|
"""
|
|
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,
|
|
) -> None:
|
|
super().__init__(
|
|
library=library,
|
|
pattern=pattern,
|
|
ports=ports,
|
|
tools=tools,
|
|
name=name,
|
|
debug=debug,
|
|
auto_render=True,
|
|
)
|
|
|
|
|
|
class RenderPather(Pather):
|
|
"""
|
|
Backward-compatible wrapper for Pather with auto_render=False.
|
|
"""
|
|
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,
|
|
) -> None:
|
|
super().__init__(
|
|
library=library,
|
|
pattern=pattern,
|
|
ports=ports,
|
|
tools=tools,
|
|
name=name,
|
|
debug=debug,
|
|
auto_render=False,
|
|
)
|