2353 lines
85 KiB
Python
2353 lines
85 KiB
Python
"""
|
|
Library classes for managing unique name->pattern mappings and deferred loading or execution.
|
|
|
|
Classes include:
|
|
- `ILibraryView`: Defines a general interface for read-only name->pattern mappings.
|
|
- `LibraryView`: An implementation of `ILibraryView` backed by an arbitrary `Mapping`.
|
|
Can be used to wrap any arbitrary `Mapping` to give it all the functionality in `ILibraryView`
|
|
- `ILibrary`: Defines a general interface for mutable name->pattern mappings.
|
|
- `Library`: An implementation of `ILibrary` backed by an arbitrary `MutableMapping`.
|
|
Can be used to wrap any arbitrary `MutableMapping` to give it all the functionality in `ILibrary`.
|
|
By default, uses a `dict` as the underylingmapping.
|
|
- `LazyLibrary`: An implementation of `ILibrary` which enables on-demand loading or generation
|
|
of patterns.
|
|
- `AbstractView`: Provides a way to use []-indexing to generate abstracts for patterns in the linked
|
|
library. Generated with `ILibraryView.abstract_view()`.
|
|
"""
|
|
from typing import Self, TYPE_CHECKING, Any, cast, TypeAlias, Protocol, Literal
|
|
from collections.abc import Container, Iterator, Mapping, MutableMapping, Sequence, Callable
|
|
import logging
|
|
import re
|
|
import copy
|
|
from functools import wraps
|
|
from pprint import pformat
|
|
from collections import defaultdict
|
|
from abc import ABCMeta, abstractmethod
|
|
from contextvars import ContextVar
|
|
from dataclasses import dataclass, replace
|
|
from graphlib import TopologicalSorter, CycleError
|
|
|
|
import numpy
|
|
from numpy.typing import ArrayLike, NDArray
|
|
|
|
from .error import BuildError, LibraryError, PatternError
|
|
from .utils import layer_t, apply_transforms
|
|
from .shapes import Shape, Polygon
|
|
from .label import Label
|
|
from .abstract import Abstract
|
|
from .pattern import map_layers
|
|
|
|
if TYPE_CHECKING:
|
|
from .pattern import Pattern
|
|
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
_ACTIVE_BUILD_SESSIONS: ContextVar[dict[int, '_BuildSessionLibrary'] | None] = ContextVar(
|
|
'masque_active_build_sessions',
|
|
default=None,
|
|
)
|
|
|
|
|
|
class visitor_function_t(Protocol):
|
|
""" Signature for `Library.dfs()` visitor functions. """
|
|
def __call__(
|
|
self,
|
|
pattern: 'Pattern',
|
|
hierarchy: tuple[str | None, ...],
|
|
memo: dict,
|
|
transform: NDArray[numpy.float64] | Literal[False],
|
|
) -> 'Pattern':
|
|
...
|
|
|
|
|
|
TreeView: TypeAlias = Mapping[str, 'Pattern']
|
|
""" A name-to-`Pattern` mapping which is expected to have only one top-level cell """
|
|
|
|
Tree: TypeAlias = MutableMapping[str, 'Pattern']
|
|
""" A mutable name-to-`Pattern` mapping which is expected to have only one top-level cell """
|
|
|
|
dangling_mode_t: TypeAlias = Literal['error', 'ignore', 'include']
|
|
""" How helpers should handle refs whose targets are not present in the library. """
|
|
|
|
|
|
@dataclass(frozen=True)
|
|
class CellProvenance:
|
|
"""
|
|
Provenance record for one cell in a completed build output.
|
|
|
|
Each output name in a `BuildReport` maps to one `CellProvenance`. The
|
|
record captures both where the cell came from and how its visible name was
|
|
chosen.
|
|
|
|
Attributes:
|
|
requested_name: First name requested for this cell during the build.
|
|
kind: Whether the cell came from a declaration, helper emission, or an
|
|
imported source library.
|
|
owner_declared_name: Declared cell responsible for this output cell, if
|
|
any. Imported source cells leave this as `None`.
|
|
build_chain: Declared-cell dependency chain that was active when the
|
|
cell was emitted.
|
|
"""
|
|
requested_name: str
|
|
kind: Literal['declared', 'helper', 'source']
|
|
owner_declared_name: str | None
|
|
build_chain: tuple[str, ...]
|
|
|
|
|
|
@dataclass(frozen=True)
|
|
class BuildReport:
|
|
"""
|
|
Immutable summary of one `BuildLibrary.validate()` or `.build()` run.
|
|
|
|
The report is designed to answer two questions after a build completes:
|
|
which declared cells depended on which other declared cells, and where each
|
|
output cell came from.
|
|
|
|
Attributes:
|
|
requested_roots: Roots explicitly requested for the run. A full
|
|
`build()` uses all declared cells.
|
|
provenance: Mapping from final output name to provenance metadata.
|
|
dependency_graph: Declared-cell dependency graph discovered through
|
|
library-mediated reads and explicit recipe hints.
|
|
"""
|
|
requested_roots: tuple[str, ...]
|
|
provenance: Mapping[str, CellProvenance]
|
|
dependency_graph: Mapping[str, frozenset[str]]
|
|
|
|
|
|
SINGLE_USE_PREFIX = '_'
|
|
"""
|
|
Names starting with this prefix are assumed to refer to single-use patterns,
|
|
which may be renamed automatically by `ILibrary.add()` (via
|
|
`rename_theirs=_rename_patterns()` )
|
|
"""
|
|
# TODO what are the consequences of making '_' special? maybe we can make this decision everywhere?
|
|
|
|
|
|
def _rename_patterns(lib: 'ILibraryView', name: str) -> str:
|
|
"""
|
|
The default `rename_theirs` function for `ILibrary.add`.
|
|
|
|
Treats names starting with `SINGLE_USE_PREFIX` (default: one underscore) as
|
|
"one-offs" for which name conflicts should be automatically resolved.
|
|
Conflicts are resolved by calling `lib.get_name(SINGLE_USE_PREFIX + stem)`
|
|
where `stem = name.removeprefix(SINGLE_USE_PREFIX).split('$')[0]`.
|
|
Names lacking the prefix are directly returned (not renamed).
|
|
|
|
Args:
|
|
lib: The library into which `name` is to be added (but is presumed to conflict)
|
|
name: The original name, to be modified
|
|
|
|
Returns:
|
|
The new name, not guaranteed to be conflict-free!
|
|
"""
|
|
if not name.startswith(SINGLE_USE_PREFIX):
|
|
return name
|
|
|
|
stem = name.removeprefix(SINGLE_USE_PREFIX).split('$')[0]
|
|
return lib.get_name(SINGLE_USE_PREFIX + stem)
|
|
|
|
|
|
def _plan_source_names(
|
|
target: 'ILibraryView',
|
|
source_order: Sequence[str],
|
|
existing_names: Container[str],
|
|
*,
|
|
rename_theirs: Callable[['ILibraryView', str], str] | None = None,
|
|
rename_when: Literal['conflict', 'always'] = 'conflict',
|
|
) -> dict[str, str]:
|
|
if rename_when not in ('conflict', 'always'):
|
|
raise ValueError(f'Unknown source rename mode: {rename_when!r}')
|
|
if rename_when == 'always' and rename_theirs is None:
|
|
raise TypeError('rename_theirs is required when rename_when="always"')
|
|
|
|
source_to_visible: dict[str, str] = {}
|
|
visible_names: set[str] = set()
|
|
|
|
for name in source_order:
|
|
visible = name
|
|
if rename_when == 'always':
|
|
assert rename_theirs is not None
|
|
visible = rename_theirs(target, name)
|
|
elif visible in existing_names or visible in visible_names:
|
|
if rename_theirs is None:
|
|
raise LibraryError(f'Conflicting name while adding source: {name!r}')
|
|
visible = rename_theirs(target, name)
|
|
if visible in existing_names or visible in visible_names:
|
|
raise LibraryError(f'Unresolved duplicate key encountered while adding source: {name!r} -> {visible!r}')
|
|
source_to_visible[name] = visible
|
|
visible_names.add(visible)
|
|
|
|
return source_to_visible
|
|
|
|
|
|
def _source_rename_map(source_to_visible: Mapping[str, str]) -> dict[str, str]:
|
|
return {
|
|
source_name: visible_name
|
|
for source_name, visible_name in source_to_visible.items()
|
|
if source_name != visible_name
|
|
}
|
|
|
|
|
|
class ILibraryView(Mapping[str, 'Pattern'], metaclass=ABCMeta):
|
|
"""
|
|
Interface for a read-only library.
|
|
|
|
A library is a mapping from unique names (str) to collections of geometry (`Pattern`).
|
|
"""
|
|
# inherited abstract functions
|
|
#def __getitem__(self, key: str) -> 'Pattern':
|
|
#def __iter__(self) -> Iterator[str]:
|
|
#def __len__(self) -> int:
|
|
|
|
#__contains__, keys, items, values, get, __eq__, __ne__ supplied by Mapping
|
|
|
|
def __repr__(self) -> str:
|
|
return '<ILibraryView with keys\n' + pformat(list(self.keys())) + '>'
|
|
|
|
def abstract_view(self) -> 'AbstractView':
|
|
"""
|
|
Returns:
|
|
An AbstractView into this library
|
|
"""
|
|
return AbstractView(self)
|
|
|
|
def abstract(self, name: str) -> Abstract:
|
|
"""
|
|
Return an `Abstract` (name & ports) for the pattern in question.
|
|
|
|
Args:
|
|
name: The pattern name
|
|
|
|
Returns:
|
|
An `Abstract` object for the pattern
|
|
"""
|
|
return Abstract(name=name, ports=self[name].ports)
|
|
|
|
def source_order(self) -> tuple[str, ...]:
|
|
"""
|
|
Return names in the library's preferred source order.
|
|
|
|
Source-backed views may override this to preserve on-disk ordering
|
|
without materializing patterns.
|
|
"""
|
|
return tuple(self.keys())
|
|
|
|
def dangling_refs(
|
|
self,
|
|
tops: str | Sequence[str] | None = None,
|
|
) -> set[str | None]:
|
|
"""
|
|
Get the set of all pattern names not present in the library but referenced
|
|
by `tops`, recursively traversing any refs.
|
|
|
|
If `tops` are not given, all patterns in the library are checked.
|
|
|
|
Args:
|
|
tops: Name(s) of the pattern(s) to check.
|
|
Default is all patterns in the library.
|
|
|
|
Returns:
|
|
Set of all referenced pattern names
|
|
"""
|
|
if tops is None:
|
|
tops = tuple(self.keys())
|
|
|
|
referenced = self.referenced_patterns(tops)
|
|
return referenced - set(self.keys())
|
|
|
|
def referenced_patterns(
|
|
self,
|
|
tops: str | Sequence[str] | None = None,
|
|
skip: set[str | None] | None = None,
|
|
) -> set[str | None]:
|
|
"""
|
|
Get the set of all pattern names referenced by `tops`. Recursively traverses into any refs.
|
|
|
|
If `tops` are not given, all patterns in the library are checked.
|
|
|
|
Args:
|
|
tops: Name(s) of the pattern(s) to check.
|
|
Default is all patterns in the library.
|
|
skip: Memo, set patterns which have already been traversed.
|
|
|
|
Returns:
|
|
Set of all referenced pattern names
|
|
"""
|
|
if tops is None:
|
|
tops = tuple(self.keys())
|
|
|
|
if skip is None:
|
|
skip = {None}
|
|
|
|
if isinstance(tops, str):
|
|
tops = (tops,)
|
|
tops = set(tops)
|
|
skip |= tops # don't re-visit tops
|
|
|
|
# Get referenced patterns for all tops
|
|
targets = set()
|
|
for top in set(tops):
|
|
targets |= self[top].referenced_patterns()
|
|
|
|
# Perform recursive lookups, but only once for each name
|
|
for target in targets - skip:
|
|
assert target is not None
|
|
skip.add(target)
|
|
if target in self:
|
|
targets |= self.referenced_patterns(target, skip=skip)
|
|
|
|
return targets
|
|
|
|
def subtree(
|
|
self,
|
|
tops: str | Sequence[str],
|
|
) -> 'ILibraryView':
|
|
"""
|
|
Return a new `ILibraryView`, containing only the specified patterns and the patterns they
|
|
reference (recursively).
|
|
Dangling references do not cause an error.
|
|
|
|
Args:
|
|
tops: Name(s) of patterns to keep
|
|
|
|
Returns:
|
|
A `LibraryView` containing only `tops` and the patterns they reference.
|
|
"""
|
|
if isinstance(tops, str):
|
|
tops = (tops,)
|
|
|
|
keep = cast('set[str]', self.referenced_patterns(tops) - {None})
|
|
keep |= set(tops)
|
|
|
|
filtered = {kk: vv for kk, vv in self.items() if kk in keep}
|
|
new = LibraryView(filtered)
|
|
return new
|
|
|
|
def polygonize(
|
|
self,
|
|
num_vertices: int | None = None,
|
|
max_arclen: float | None = None,
|
|
) -> Self:
|
|
"""
|
|
Calls `.polygonize(...)` on each pattern in this library.
|
|
Arguments are passed on to `shape.to_polygons(...)`.
|
|
|
|
Args:
|
|
num_vertices: Number of points to use for each polygon. Can be overridden by
|
|
`max_arclen` if that results in more points. Optional, defaults to shapes'
|
|
internal defaults.
|
|
max_arclen: Maximum arclength which can be approximated by a single line
|
|
segment. Optional, defaults to shapes' internal defaults.
|
|
|
|
Returns:
|
|
self
|
|
"""
|
|
for pat in self.values():
|
|
pat.polygonize(num_vertices, max_arclen)
|
|
return self
|
|
|
|
def manhattanize(
|
|
self,
|
|
grid_x: ArrayLike,
|
|
grid_y: ArrayLike,
|
|
) -> Self:
|
|
"""
|
|
Calls `.manhattanize(grid_x, grid_y)` on each pattern in this library.
|
|
|
|
Args:
|
|
grid_x: List of allowed x-coordinates for the Manhattanized polygon edges.
|
|
grid_y: List of allowed y-coordinates for the Manhattanized polygon edges.
|
|
|
|
Returns:
|
|
self
|
|
"""
|
|
for pat in self.values():
|
|
pat.manhattanize(grid_x, grid_y)
|
|
return self
|
|
|
|
def flatten(
|
|
self,
|
|
tops: str | Sequence[str],
|
|
flatten_ports: bool = False,
|
|
dangling_ok: bool = False,
|
|
) -> dict[str, 'Pattern']:
|
|
"""
|
|
Returns copies of all `tops` patterns with all refs
|
|
removed and replaced with equivalent shapes.
|
|
Also returns flattened copies of all referenced patterns.
|
|
The originals in the calling `Library` are not modified.
|
|
For an in-place variant, see `Pattern.flatten`.
|
|
|
|
Args:
|
|
tops: The pattern(s) to flatten.
|
|
flatten_ports: If `True`, keep ports from any referenced
|
|
patterns; otherwise discard them.
|
|
dangling_ok: If `True`, no error will be thrown if any
|
|
ref points to a name which is not present in the library.
|
|
Default False.
|
|
|
|
Returns:
|
|
{name: flat_pattern} mapping for all flattened patterns.
|
|
"""
|
|
if isinstance(tops, str):
|
|
tops = (tops,)
|
|
|
|
flattened: dict[str, Pattern | None] = {}
|
|
|
|
def flatten_single(name: str) -> None:
|
|
flattened[name] = None
|
|
pat = self[name].deepcopy()
|
|
refs_by_target = tuple((target, tuple(refs)) for target, refs in pat.refs.items())
|
|
|
|
for target, refs in refs_by_target:
|
|
if target is None:
|
|
continue
|
|
if dangling_ok and target not in self:
|
|
continue
|
|
if target not in flattened:
|
|
flatten_single(target)
|
|
|
|
target_pat = flattened[target]
|
|
if target_pat is None:
|
|
raise PatternError(f'Circular reference in {name} to {target}')
|
|
ports_only = flatten_ports and bool(target_pat.ports)
|
|
if target_pat.is_empty() and not ports_only: # avoid some extra allocations
|
|
continue
|
|
|
|
for ref in refs:
|
|
if flatten_ports and ref.repetition is not None and target_pat.ports:
|
|
raise PatternError(
|
|
f'Cannot flatten ports from repeated ref to {target!r}; '
|
|
'flatten with flatten_ports=False or expand/rename the ports manually first.'
|
|
)
|
|
p = ref.as_pattern(pattern=target_pat)
|
|
if not flatten_ports:
|
|
p.ports.clear()
|
|
pat.append(p)
|
|
|
|
for target in set(pat.refs.keys()) & set(self.keys()):
|
|
del pat.refs[target]
|
|
|
|
flattened[name] = pat
|
|
|
|
for top in tops:
|
|
flatten_single(top)
|
|
|
|
assert None not in flattened.values()
|
|
return cast('dict[str, Pattern]', flattened)
|
|
|
|
def get_name(
|
|
self,
|
|
name: str = SINGLE_USE_PREFIX * 2,
|
|
sanitize: bool = True,
|
|
max_length: int = 32,
|
|
quiet: bool | None = None,
|
|
) -> str:
|
|
"""
|
|
Find a unique name for the pattern.
|
|
|
|
This function may be overridden in a subclass or monkey-patched to fit the caller's requirements.
|
|
|
|
Args:
|
|
name: Preferred name for the pattern. Default is `SINGLE_USE_PREFIX * 2`.
|
|
sanitize: Allows only alphanumeric charaters and _?$. Replaces invalid characters with underscores.
|
|
max_length: Names longer than this will be truncated.
|
|
quiet: If `True`, suppress log messages. Default `None` suppresses messages only if
|
|
the name starts with `SINGLE_USE_PREFIX`.
|
|
|
|
Returns:
|
|
Name, unique within this library.
|
|
"""
|
|
if quiet is None:
|
|
quiet = name.startswith(SINGLE_USE_PREFIX)
|
|
|
|
if sanitize:
|
|
# Remove invalid characters
|
|
sanitized_name = re.compile(r'[^A-Za-z0-9_\?\$]').sub('_', name)
|
|
else:
|
|
sanitized_name = name
|
|
|
|
suffixed_name = sanitized_name
|
|
if sanitized_name in self:
|
|
ii = sum(1 for nn in self.keys() if nn.startswith(sanitized_name))
|
|
else:
|
|
ii = 0
|
|
while suffixed_name in self or suffixed_name == '':
|
|
suffixed_name = sanitized_name + b64suffix(ii)
|
|
ii += 1
|
|
|
|
if len(suffixed_name) > max_length:
|
|
if name == '':
|
|
raise LibraryError(f'No valid pattern names remaining within the specified {max_length=}')
|
|
|
|
cropped_name = self.get_name(sanitized_name[:-1], sanitize=sanitize, max_length=max_length, quiet=True)
|
|
else:
|
|
cropped_name = suffixed_name
|
|
|
|
if not quiet:
|
|
logger.info(f'Requested name "{name}" changed to "{cropped_name}"')
|
|
|
|
return cropped_name
|
|
|
|
def tops(self) -> list[str]:
|
|
"""
|
|
Return the list of all patterns that are not referenced by any other pattern in the library.
|
|
|
|
Returns:
|
|
A list of pattern names in which no pattern is referenced by any other pattern.
|
|
"""
|
|
names = set(self.keys())
|
|
not_toplevel: set[str | None] = set()
|
|
for name in names:
|
|
not_toplevel |= set(self[name].refs.keys())
|
|
|
|
toplevel = list(names - not_toplevel)
|
|
return toplevel
|
|
|
|
def top(self) -> str:
|
|
"""
|
|
Return the name of the topcell, or raise an exception if there isn't a single topcell
|
|
|
|
Raises:
|
|
LibraryError if there is not exactly one topcell.
|
|
"""
|
|
tops = self.tops()
|
|
if len(tops) != 1:
|
|
raise LibraryError(f'Asked for the single topcell, but found the following: {pformat(tops)}')
|
|
return tops[0]
|
|
|
|
def top_pattern(self) -> 'Pattern':
|
|
"""
|
|
Shorthand for self[self.top()]
|
|
|
|
Raises:
|
|
LibraryError if there is not exactly one topcell.
|
|
"""
|
|
return self[self.top()]
|
|
|
|
@staticmethod
|
|
def _dangling_refs_error(dangling: set[str], context: str) -> LibraryError:
|
|
dangling_list = sorted(dangling)
|
|
return LibraryError(f'Dangling refs found while {context}: ' + pformat(dangling_list))
|
|
|
|
def _raw_child_graph(self) -> tuple[dict[str, set[str]], set[str]]:
|
|
existing = set(self.keys())
|
|
graph: dict[str, set[str]] = {}
|
|
dangling: set[str] = set()
|
|
for name, pat in self.items():
|
|
children = {child for child, refs in pat.refs.items() if child is not None and refs}
|
|
graph[name] = children
|
|
dangling |= children - existing
|
|
return graph, dangling
|
|
|
|
def dfs(
|
|
self,
|
|
pattern: 'Pattern',
|
|
visit_before: visitor_function_t | None = None,
|
|
visit_after: visitor_function_t | None = None,
|
|
*,
|
|
hierarchy: tuple[str | None, ...] = (None,),
|
|
transform: ArrayLike | bool | None = False,
|
|
memo: dict | None = None,
|
|
) -> Self:
|
|
"""
|
|
Convenience function.
|
|
Performs a depth-first traversal of a pattern and its referenced patterns.
|
|
At each pattern in the tree, the following sequence is called:
|
|
```
|
|
current_pattern = visit_before(current_pattern, **vist_args)
|
|
for target in current_pattern.refs:
|
|
for ref in pattern.refs[target]:
|
|
self.dfs(target, visit_before, visit_after,
|
|
hierarchy + (sp.target,), updated_transform, memo)
|
|
current_pattern = visit_after(current_pattern, **visit_args)
|
|
```
|
|
where `visit_args` are
|
|
`hierarchy`: (top_pattern_or_None, L1_pattern, L2_pattern, ..., parent_pattern, target_pattern)
|
|
tuple of all parent-and-higher pattern names. Top pattern name may be
|
|
`None` if not provided in first call to .dfs()
|
|
`transform`: numpy.ndarray containing cumulative
|
|
[x_offset, y_offset, rotation (rad), mirror_x (0 or 1)]
|
|
for the instance being visited
|
|
`memo`: Arbitrary dict (not altered except by `visit_before()` and `visit_after()`)
|
|
|
|
Args:
|
|
pattern: Pattern object to start at ("top"/root node of the tree).
|
|
visit_before: Function to call before traversing refs.
|
|
Should accept a `Pattern` and `**visit_args`, and return the (possibly modified)
|
|
pattern. Default `None` (not called).
|
|
visit_after: Function to call after traversing refs.
|
|
Should accept a `Pattern` and `**visit_args`, and return the (possibly modified)
|
|
pattern. Default `None` (not called).
|
|
transform: Initial value for `visit_args['transform']`.
|
|
Can be `False`, in which case the transform is not calculated.
|
|
`True` or `None` is interpreted as `[0, 0, 0, 0]`.
|
|
memo: Arbitrary dict for use by `visit_*()` functions. Default `None` (empty dict).
|
|
hierarchy: Tuple of patterns specifying the hierarchy above the current pattern.
|
|
Default is (None,), which will be used as a placeholder for the top pattern's
|
|
name if not overridden.
|
|
|
|
Returns:
|
|
self
|
|
"""
|
|
if memo is None:
|
|
memo = {}
|
|
|
|
if transform is None or transform is True:
|
|
transform = numpy.array([0, 0, 0, 0, 1], dtype=float)
|
|
elif transform is not False:
|
|
transform = numpy.asarray(transform, dtype=float)
|
|
if transform.size == 4:
|
|
transform = numpy.append(transform, 1.0)
|
|
|
|
original_pattern = pattern
|
|
|
|
if visit_before is not None:
|
|
pattern = visit_before(pattern, hierarchy=hierarchy, memo=memo, transform=transform)
|
|
|
|
for target in pattern.refs:
|
|
if target is None:
|
|
continue
|
|
if target in hierarchy:
|
|
raise LibraryError(f'.dfs() called on pattern with circular reference to "{target}"')
|
|
|
|
for ref in pattern.refs[target]:
|
|
ref_transforms: list[bool] | NDArray[numpy.float64]
|
|
if transform is not False:
|
|
ref_transforms = apply_transforms(transform, ref.as_transforms())
|
|
else:
|
|
ref_transforms = [False]
|
|
|
|
for ref_transform in ref_transforms:
|
|
self.dfs(
|
|
pattern = self[target],
|
|
visit_before = visit_before,
|
|
visit_after = visit_after,
|
|
hierarchy = hierarchy + (target,),
|
|
transform = ref_transform,
|
|
memo = memo,
|
|
)
|
|
|
|
if visit_after is not None:
|
|
pattern = visit_after(pattern, hierarchy=hierarchy, memo=memo, transform=transform)
|
|
|
|
if pattern is not original_pattern:
|
|
name = hierarchy[-1]
|
|
if not isinstance(self, ILibrary):
|
|
raise LibraryError('visit_* functions returned a new `Pattern` object'
|
|
' but the library is immutable')
|
|
if name is None:
|
|
# The top pattern is not the original pattern, but we don't know what to call it!
|
|
raise LibraryError('visit_* functions returned a new `Pattern` object'
|
|
' but no top-level name was provided in `hierarchy`')
|
|
|
|
del cast('ILibrary', self)[name]
|
|
cast('ILibrary', self)[name] = pattern
|
|
|
|
return self
|
|
|
|
def child_graph(
|
|
self,
|
|
dangling: dangling_mode_t = 'error',
|
|
) -> dict[str, set[str]]:
|
|
"""
|
|
Return a mapping from pattern name to a set of all child patterns
|
|
(patterns it references).
|
|
|
|
Only non-empty ref lists with non-`None` targets are treated as graph edges.
|
|
|
|
Args:
|
|
dangling: How refs to missing targets are handled. `'error'` raises,
|
|
`'ignore'` drops those edges, and `'include'` exposes them as
|
|
synthetic leaf nodes.
|
|
|
|
Returns:
|
|
Mapping from pattern name to a set of all pattern names it references.
|
|
"""
|
|
graph, dangling_refs = self._raw_child_graph()
|
|
if dangling == 'error':
|
|
if dangling_refs:
|
|
raise self._dangling_refs_error(dangling_refs, 'building child graph')
|
|
return graph
|
|
if dangling == 'ignore':
|
|
existing = set(graph)
|
|
return {name: {child for child in children if child in existing} for name, children in graph.items()}
|
|
|
|
for target in dangling_refs:
|
|
graph.setdefault(target, set())
|
|
return graph
|
|
|
|
def parent_graph(
|
|
self,
|
|
dangling: dangling_mode_t = 'error',
|
|
) -> dict[str, set[str]]:
|
|
"""
|
|
Return a mapping from pattern name to a set of all parent patterns
|
|
(patterns which reference it).
|
|
|
|
Args:
|
|
dangling: How refs to missing targets are handled. `'error'` raises,
|
|
`'ignore'` drops those targets, and `'include'` adds them as
|
|
synthetic keys whose values are their existing parents.
|
|
|
|
Returns:
|
|
Mapping from pattern name to a set of all patterns which reference it.
|
|
"""
|
|
child_graph, dangling_refs = self._raw_child_graph()
|
|
if dangling == 'error' and dangling_refs:
|
|
raise self._dangling_refs_error(dangling_refs, 'building parent graph')
|
|
|
|
existing = set(child_graph)
|
|
igraph: dict[str, set[str]] = {name: set() for name in existing}
|
|
for parent, children in child_graph.items():
|
|
for child in children:
|
|
if child in existing:
|
|
igraph[child].add(parent)
|
|
elif dangling == 'include':
|
|
igraph.setdefault(child, set()).add(parent)
|
|
return igraph
|
|
|
|
def child_order(
|
|
self,
|
|
dangling: dangling_mode_t = 'error',
|
|
) -> list[str]:
|
|
"""
|
|
Return a topologically sorted list of graph node names.
|
|
Child (referenced) patterns will appear before their parents.
|
|
|
|
Args:
|
|
dangling: Passed to `child_graph()`.
|
|
|
|
Return:
|
|
Topologically sorted list of pattern names.
|
|
"""
|
|
try:
|
|
return cast('list[str]', list(TopologicalSorter(self.child_graph(dangling=dangling)).static_order()))
|
|
except CycleError as exc:
|
|
cycle = exc.args[1] if len(exc.args) > 1 else None
|
|
if cycle is None:
|
|
raise LibraryError('Cycle found while building child order') from exc
|
|
raise LibraryError(f'Cycle found while building child order: {cycle}') from exc
|
|
|
|
def find_refs_local(
|
|
self,
|
|
name: str,
|
|
parent_graph: dict[str, set[str]] | None = None,
|
|
dangling: dangling_mode_t = 'error',
|
|
) -> dict[str, list[NDArray[numpy.float64]]]:
|
|
"""
|
|
Find the location and orientation of all refs pointing to `name`.
|
|
Refs with a `repetition` are resolved into multiple instances (locations).
|
|
|
|
Args:
|
|
name: Name of the referenced pattern.
|
|
parent_graph: Mapping from pattern name to the set of patterns which
|
|
reference it. Default (`None`) calls `self.parent_graph()`.
|
|
The provided graph may be for a superset of `self` (i.e. it may
|
|
contain additional patterns which are not present in self; they
|
|
will be ignored).
|
|
dangling: How refs to missing targets are handled if `parent_graph`
|
|
is not provided. `'include'` also allows querying missing names.
|
|
|
|
Returns:
|
|
Mapping of {parent_name: transform_list}, where transform_list
|
|
is an Nx4 ndarray with rows
|
|
`(x_offset, y_offset, rotation_ccw_rad, mirror_across_x)`.
|
|
"""
|
|
instances = defaultdict(list)
|
|
if parent_graph is None:
|
|
graph_mode = 'ignore' if dangling == 'ignore' else 'include'
|
|
parent_graph = self.parent_graph(dangling=graph_mode)
|
|
|
|
if name not in self:
|
|
if name not in parent_graph:
|
|
return instances
|
|
if dangling == 'error':
|
|
raise self._dangling_refs_error({name}, f'finding local refs for {name!r}')
|
|
if dangling == 'ignore':
|
|
return instances
|
|
|
|
for parent in parent_graph.get(name, set()):
|
|
if parent not in self: # parent_graph may be a for a superset of self
|
|
continue
|
|
for ref in self[parent].refs[name]:
|
|
instances[parent].append(ref.as_transforms())
|
|
|
|
return instances
|
|
|
|
def find_refs_global(
|
|
self,
|
|
name: str,
|
|
order: list[str] | None = None,
|
|
parent_graph: dict[str, set[str]] | None = None,
|
|
dangling: dangling_mode_t = 'error',
|
|
) -> dict[tuple[str, ...], NDArray[numpy.float64]]:
|
|
"""
|
|
Find the absolute (top-level) location and orientation of all refs (including
|
|
repetitions) pointing to `name`.
|
|
|
|
Args:
|
|
name: Name of the referenced pattern.
|
|
order: List of pattern names in which children are guaranteed
|
|
to appear before their parents (i.e. topologically sorted).
|
|
Default (`None`) calls `self.child_order()`.
|
|
parent_graph: Passed to `find_refs_local`.
|
|
Mapping from pattern name to the set of patterns which
|
|
reference it. Default (`None`) calls `self.parent_graph()`.
|
|
The provided graph may be for a superset of `self` (i.e. it may
|
|
contain additional patterns which are not present in self; they
|
|
will be ignored).
|
|
dangling: How refs to missing targets are handled if `order` or
|
|
`parent_graph` are not provided. `'include'` also allows
|
|
querying missing names.
|
|
|
|
Returns:
|
|
Mapping of `{hierarchy: transform_list}`, where `hierarchy` is a tuple of the form
|
|
`(toplevel_pattern, lvl1_pattern, ..., name)` and `transform_list` is an Nx4 ndarray
|
|
with rows `(x_offset, y_offset, rotation_ccw_rad, mirror_across_x)`.
|
|
"""
|
|
graph_mode = 'ignore' if dangling == 'ignore' else 'include'
|
|
if order is None:
|
|
order = self.child_order(dangling=graph_mode)
|
|
if parent_graph is None:
|
|
parent_graph = self.parent_graph(dangling=graph_mode)
|
|
|
|
if name not in self:
|
|
if name not in parent_graph:
|
|
return {}
|
|
if dangling == 'error':
|
|
raise self._dangling_refs_error({name}, f'finding global refs for {name!r}')
|
|
if dangling == 'ignore':
|
|
return {}
|
|
|
|
self_keys = set(self.keys())
|
|
|
|
transforms: dict[str, list[tuple[
|
|
tuple[str, ...],
|
|
NDArray[numpy.float64]
|
|
]]]
|
|
transforms = defaultdict(list)
|
|
for parent, vals in self.find_refs_local(name, parent_graph=parent_graph, dangling=dangling).items():
|
|
transforms[parent] = [((name,), numpy.concatenate(vals))]
|
|
|
|
for next_name in order:
|
|
if next_name not in transforms:
|
|
continue
|
|
if not parent_graph.get(next_name, set()) & self_keys:
|
|
continue
|
|
|
|
outers = self.find_refs_local(next_name, parent_graph=parent_graph, dangling=dangling)
|
|
inners = transforms.pop(next_name)
|
|
for parent, outer in outers.items():
|
|
for path, inner in inners:
|
|
combined = apply_transforms(numpy.concatenate(outer), inner)
|
|
transforms[parent].append((
|
|
(next_name,) + path,
|
|
combined,
|
|
))
|
|
result = {}
|
|
for parent, targets in transforms.items():
|
|
for path, instances in targets:
|
|
full_path = (parent,) + path
|
|
assert full_path not in result
|
|
result[full_path] = instances
|
|
return result
|
|
|
|
|
|
|
|
class ILibrary(ILibraryView, MutableMapping[str, 'Pattern'], metaclass=ABCMeta):
|
|
"""
|
|
Interface for a writeable library.
|
|
|
|
A library is a mapping from unique names (str) to collections of geometry (`Pattern`).
|
|
"""
|
|
# inherited abstract functions
|
|
#def __getitem__(self, key: str) -> 'Pattern':
|
|
#def __iter__(self) -> Iterator[str]:
|
|
#def __len__(self) -> int:
|
|
#def __setitem__(self, key: str, value: 'Pattern | Callable[[], Pattern]') -> None:
|
|
#def __delitem__(self, key: str) -> None:
|
|
|
|
@abstractmethod
|
|
def __setitem__(
|
|
self,
|
|
key: str,
|
|
value: 'Pattern | Callable[[], Pattern]',
|
|
) -> None:
|
|
pass
|
|
|
|
@abstractmethod
|
|
def __delitem__(self, key: str) -> None:
|
|
pass
|
|
|
|
@abstractmethod
|
|
def _merge(self, key_self: str, other: Mapping[str, 'Pattern'], key_other: str) -> None:
|
|
pass
|
|
|
|
def resolve(
|
|
self,
|
|
other: 'Abstract | str | Pattern | TreeView',
|
|
append: bool = False,
|
|
) -> 'Abstract | Pattern':
|
|
"""
|
|
Resolve another device (name, Abstract, Pattern, or TreeView) into an Abstract or Pattern.
|
|
If it is a TreeView, it is first added into this library.
|
|
|
|
Args:
|
|
other: The device to resolve.
|
|
append: If True and `other` is an `Abstract`, returns the full `Pattern` from the library.
|
|
|
|
Returns:
|
|
An `Abstract` or `Pattern` object.
|
|
"""
|
|
from .pattern import Pattern #noqa: PLC0415
|
|
if not isinstance(other, (str, Abstract, Pattern)):
|
|
# We got a TreeView; add it into self and grab its topcell as an Abstract
|
|
other = self << other
|
|
|
|
if isinstance(other, str):
|
|
other = self.abstract(other)
|
|
if append and isinstance(other, Abstract):
|
|
other = self[other.name]
|
|
return other
|
|
|
|
def rename(
|
|
self,
|
|
old_name: str,
|
|
new_name: str,
|
|
move_references: bool = False,
|
|
) -> Self:
|
|
"""
|
|
Rename a pattern.
|
|
|
|
Args:
|
|
old_name: Current name for the pattern
|
|
new_name: New name for the pattern
|
|
move_references: If `True`, any refs in this library pointing to `old_name`
|
|
will be updated to point to `new_name`.
|
|
|
|
Returns:
|
|
self
|
|
"""
|
|
if old_name not in self:
|
|
raise LibraryError(f'"{old_name}" does not exist in the library.')
|
|
if old_name == new_name:
|
|
return self
|
|
|
|
self[new_name] = self[old_name]
|
|
del self[old_name]
|
|
if move_references:
|
|
self.move_references(old_name, new_name)
|
|
return self
|
|
|
|
def rename_top(self, name: str) -> Self:
|
|
"""
|
|
Rename the (single) top pattern
|
|
"""
|
|
self.rename(self.top(), name, move_references=True)
|
|
return self
|
|
|
|
def move_references(self, old_target: str, new_target: str) -> Self:
|
|
"""
|
|
Change all references pointing at `old_target` into references pointing at `new_target`.
|
|
|
|
Args:
|
|
old_target: Current reference target
|
|
new_target: New target for the reference
|
|
|
|
Returns:
|
|
self
|
|
"""
|
|
if old_target == new_target:
|
|
return self
|
|
|
|
for pattern in self.values():
|
|
if old_target in pattern.refs:
|
|
pattern.refs[new_target].extend(pattern.refs[old_target])
|
|
del pattern.refs[old_target]
|
|
return self
|
|
|
|
def map_layers(
|
|
self,
|
|
map_layer: Callable[[layer_t], layer_t],
|
|
) -> Self:
|
|
"""
|
|
Move all the elements in all patterns from one layer onto a different layer.
|
|
Can also handle multiple such mappings simultaneously.
|
|
|
|
Args:
|
|
map_layer: Callable which may be called with each layer present in `elements`,
|
|
and should return the new layer to which it will be mapped.
|
|
A simple example which maps `old_layer` to `new_layer` and leaves all others
|
|
as-is would look like `lambda layer: {old_layer: new_layer}.get(layer, layer)`
|
|
|
|
Returns:
|
|
self
|
|
"""
|
|
for pattern in self.values():
|
|
pattern.shapes = map_layers(pattern.shapes, map_layer)
|
|
pattern.labels = map_layers(pattern.labels, map_layer)
|
|
return self
|
|
|
|
def mkpat(self, name: str) -> tuple[str, 'Pattern']:
|
|
"""
|
|
Convenience method to create an empty pattern, add it to the library,
|
|
and return both the pattern and name.
|
|
|
|
Args:
|
|
name: Name for the pattern
|
|
|
|
Returns:
|
|
(name, pattern) tuple
|
|
"""
|
|
from .pattern import Pattern #noqa: PLC0415
|
|
pat = Pattern()
|
|
self[name] = pat
|
|
return name, pat
|
|
|
|
def add(
|
|
self,
|
|
other: Mapping[str, 'Pattern'],
|
|
rename_theirs: Callable[['ILibraryView', str], str] = _rename_patterns,
|
|
mutate_other: bool = False,
|
|
) -> dict[str, str]:
|
|
"""
|
|
Add items from another library into this one.
|
|
|
|
If any name in `other` is already present in `self`, `rename_theirs(self, name)` is called
|
|
to pick a new name for the newly-added pattern. If the new name still conflicts with a name
|
|
in `self` a `LibraryError` is raised. All references to the original name (within `other)`
|
|
are updated to the new name.
|
|
If `mutate_other=False` (default), all changes are made to a deepcopy of `other`.
|
|
|
|
By default, `rename_theirs` makes no changes to the name (causing a `LibraryError`) unless the
|
|
name starts with `SINGLE_USE_PREFIX`. Prefixed names are truncated to before their first
|
|
non-prefix '$' and then passed to `self.get_name()` to create a new unique name.
|
|
|
|
Args:
|
|
other: The library to insert keys from.
|
|
rename_theirs: Called as rename_theirs(self, name) for each duplicate name
|
|
encountered in `other`. Should return the new name for the pattern in
|
|
`other`. See above for default behavior.
|
|
mutate_other: If `True`, modify the original library and its contained patterns
|
|
(e.g. when renaming patterns and updating refs). Otherwise, operate on a deepcopy
|
|
(default).
|
|
|
|
Returns:
|
|
A mapping of `{old_name: new_name}` for all names in `other` which were
|
|
renamed while being added. Unchanged names are omitted.
|
|
|
|
Raises:
|
|
`LibraryError` if a duplicate name is encountered even after applying `rename_theirs()`.
|
|
"""
|
|
from .pattern import map_targets #noqa: PLC0415
|
|
duplicates = set(self.keys()) & set(other.keys())
|
|
|
|
if not duplicates:
|
|
if mutate_other:
|
|
temp = other
|
|
else:
|
|
temp = Library(copy.deepcopy(dict(other)))
|
|
|
|
for key in temp:
|
|
self._merge(key, temp, key)
|
|
return {}
|
|
|
|
if mutate_other:
|
|
if isinstance(other, Library):
|
|
temp = other
|
|
else:
|
|
temp = Library(dict(other))
|
|
else:
|
|
temp = Library(copy.deepcopy(dict(other)))
|
|
rename_map = {}
|
|
for old_name in temp:
|
|
if old_name in self:
|
|
new_name = rename_theirs(self, old_name)
|
|
if new_name in self:
|
|
raise LibraryError(f'Unresolved duplicate key encountered in library merge: {old_name} -> {new_name}')
|
|
rename_map[old_name] = new_name
|
|
else:
|
|
new_name = old_name
|
|
|
|
self._merge(new_name, temp, old_name)
|
|
|
|
# Update references in the newly-added cells
|
|
for old_name in temp:
|
|
new_name = rename_map.get(old_name, old_name)
|
|
pat = self[new_name]
|
|
pat.refs = map_targets(pat.refs, lambda tt: cast('dict[str | None, str | None]', rename_map).get(tt, tt))
|
|
|
|
return rename_map
|
|
|
|
def __lshift__(self, other: TreeView) -> str:
|
|
"""
|
|
`add()` items from a tree (single-topcell name: pattern mapping) into this one,
|
|
and return the name of the tree's topcell (in this library; it may have changed
|
|
based on `add()`'s default `rename_theirs` argument).
|
|
|
|
Raises:
|
|
LibraryError if there is more than one topcell in `other`.
|
|
"""
|
|
if len(other) == 1:
|
|
name = next(iter(other))
|
|
else:
|
|
if not isinstance(other, ILibraryView):
|
|
other = LibraryView(other)
|
|
|
|
tops = other.tops()
|
|
if len(tops) > 1:
|
|
raise LibraryError('Received a library containing multiple topcells!')
|
|
|
|
name = tops[0]
|
|
|
|
rename_map = self.add(other)
|
|
new_name = rename_map.get(name, name)
|
|
return new_name
|
|
|
|
def __le__(self, other: Mapping[str, 'Pattern']) -> Abstract:
|
|
"""
|
|
Perform the same operation as `__lshift__` / `<<`, but return an `Abstract` instead
|
|
of just the pattern's name.
|
|
|
|
Raises:
|
|
LibraryError if there is more than one topcell in `other`.
|
|
"""
|
|
new_name = self << other
|
|
return self.abstract(new_name)
|
|
|
|
def dedup(
|
|
self,
|
|
norm_value: int = int(1e6),
|
|
exclude_types: tuple[type] = (Polygon,),
|
|
label2name: Callable[[tuple], str] | None = None,
|
|
threshold: int = 2,
|
|
) -> Self:
|
|
"""
|
|
Iterates through all `Pattern`s. Within each `Pattern`, it iterates
|
|
over all shapes, calling `.normalized_form(norm_value)` on them to retrieve a scale-,
|
|
offset-, and rotation-independent form. Each shape whose normalized form appears
|
|
more than once is removed and re-added using `Ref` objects referencing a newly-created
|
|
`Pattern` containing only the normalized form of the shape.
|
|
|
|
Note:
|
|
The default norm_value was chosen to give a reasonable precision when using
|
|
integer values for coordinates.
|
|
|
|
Args:
|
|
norm_value: Passed to `shape.normalized_form(norm_value)`. Default `1e6` (see function
|
|
note)
|
|
exclude_types: Shape types passed in this argument are always left untouched, for
|
|
speed or convenience. Default: `(shapes.Polygon,)`
|
|
label2name: Given a label tuple as returned by `shape.normalized_form(...)`, pick
|
|
a name for the generated pattern.
|
|
Default `self.get_name(SINGLE_USE_PREIX + 'shape')`.
|
|
threshold: Only replace shapes with refs if there will be at least this many
|
|
instances.
|
|
|
|
Returns:
|
|
self
|
|
"""
|
|
# This currently simplifies globally (same shape in different patterns is
|
|
# merged into the same ref target).
|
|
|
|
from .pattern import Pattern #noqa: PLC0415
|
|
|
|
if exclude_types is None:
|
|
exclude_types = ()
|
|
|
|
if label2name is None:
|
|
def label2name(label: tuple) -> str: # noqa: ARG001
|
|
return self.get_name(SINGLE_USE_PREFIX + 'shape')
|
|
|
|
used_names = set(self.keys())
|
|
|
|
def reserve_target_name(label: tuple) -> str:
|
|
base_name = label2name(label)
|
|
name = base_name
|
|
ii = sum(1 for nn in used_names if nn.startswith(base_name)) if base_name in used_names else 0
|
|
while name in used_names or name == '':
|
|
name = base_name + b64suffix(ii)
|
|
ii += 1
|
|
used_names.add(name)
|
|
return name
|
|
|
|
shape_counts: MutableMapping[tuple, int] = defaultdict(int)
|
|
shape_funcs = {}
|
|
|
|
# ## First pass ##
|
|
# Using the label tuple from `.normalized_form()` as a key, check how many of each shape
|
|
# are present and store the shape function for each one
|
|
for pat in tuple(self.values()):
|
|
for layer, sseq in pat.shapes.items():
|
|
for shape in sseq:
|
|
if not any(isinstance(shape, t) for t in exclude_types):
|
|
base_label, _values, func = shape.normalized_form(norm_value)
|
|
label = (*base_label, layer)
|
|
shape_funcs[label] = func
|
|
shape_counts[label] += 1
|
|
|
|
shape_pats = {}
|
|
target_names = {}
|
|
for label, count in shape_counts.items():
|
|
if count < threshold:
|
|
continue
|
|
|
|
shape_func = shape_funcs[label]
|
|
shape_pat = Pattern()
|
|
shape_pat.shapes[label[-1]] += [shape_func()]
|
|
shape_pats[label] = shape_pat
|
|
target_names[label] = reserve_target_name(label)
|
|
|
|
# ## Second pass ##
|
|
for pat in tuple(self.values()):
|
|
# Store `[(index_in_shapes, values_from_normalized_form), ...]` for all shapes which
|
|
# are to be replaced.
|
|
# The `values` are `(offset, scale, rotation)`.
|
|
|
|
shape_table: dict[tuple, list] = defaultdict(list)
|
|
for layer, sseq in pat.shapes.items():
|
|
for ii, shape in enumerate(sseq):
|
|
if any(isinstance(shape, tt) for tt in exclude_types):
|
|
continue
|
|
|
|
base_label, values, _func = shape.normalized_form(norm_value)
|
|
label = (*base_label, layer)
|
|
|
|
if label not in shape_pats:
|
|
continue
|
|
|
|
shape_table[label].append((ii, values))
|
|
|
|
# For repeated shapes, create a `Pattern` holding a normalized shape object,
|
|
# and add `pat.refs` entries for each occurrence in pat. Also, note down that
|
|
# we should delete the `pat.shapes` entries for which we made `Ref`s.
|
|
for label, shape_entries in shape_table.items():
|
|
layer = label[-1]
|
|
target = target_names[label]
|
|
shapes_to_remove = []
|
|
for ii, values in shape_entries:
|
|
offset, scale, rotation, mirror_x = values
|
|
pat.ref(target=target, offset=offset, scale=scale,
|
|
rotation=rotation, mirrored=mirror_x)
|
|
shapes_to_remove.append(ii)
|
|
|
|
# Remove any shapes for which we have created refs.
|
|
for ii in sorted(shapes_to_remove, reverse=True):
|
|
del pat.shapes[layer][ii]
|
|
|
|
for ll, pp in shape_pats.items():
|
|
self[target_names[ll]] = pp
|
|
|
|
return self
|
|
|
|
def wrap_repeated_shapes(
|
|
self,
|
|
name_func: Callable[['Pattern', Shape | Label], str] | None = None,
|
|
) -> Self:
|
|
"""
|
|
Wraps all shapes and labels with a non-`None` `repetition` attribute
|
|
into a `Ref`/`Pattern` combination, and applies the `repetition`
|
|
to each `Ref` instead of its contained shape.
|
|
|
|
Args:
|
|
name_func: Function f(this_pattern, shape) which generates a name for the
|
|
wrapping pattern.
|
|
Default is `self.get_name(SINGLE_USE_PREFIX + 'rep')`.
|
|
|
|
Returns:
|
|
self
|
|
"""
|
|
from .pattern import Pattern #noqa: PLC0415
|
|
|
|
if name_func is None:
|
|
def name_func(_pat: Pattern, _shape: Shape | Label) -> str:
|
|
return self.get_name(SINGLE_USE_PREFIX + 'rep')
|
|
|
|
for pat in tuple(self.values()):
|
|
for layer in pat.shapes:
|
|
new_shapes = []
|
|
for shape in pat.shapes[layer]:
|
|
if shape.repetition is None:
|
|
new_shapes.append(shape)
|
|
continue
|
|
|
|
name = name_func(pat, shape)
|
|
self[name] = Pattern(shapes={layer: [shape]})
|
|
pat.ref(name, repetition=shape.repetition)
|
|
shape.repetition = None
|
|
pat.shapes[layer] = new_shapes
|
|
|
|
for layer in pat.labels:
|
|
new_labels = []
|
|
for label in pat.labels[layer]:
|
|
if label.repetition is None:
|
|
new_labels.append(label)
|
|
continue
|
|
name = name_func(pat, label)
|
|
self[name] = Pattern(labels={layer: [label]})
|
|
pat.ref(name, repetition=label.repetition)
|
|
label.repetition = None
|
|
pat.labels[layer] = new_labels
|
|
|
|
return self
|
|
|
|
def resolve_repeated_refs(self, name: str | None = None) -> Self:
|
|
"""
|
|
Expand all repeated references into multiple individual references.
|
|
Alters the library in-place.
|
|
|
|
Args:
|
|
name: If specified, only resolve repeated refs in this pattern.
|
|
Otherwise, resolve in all patterns.
|
|
|
|
Returns:
|
|
self
|
|
"""
|
|
if name is not None:
|
|
self[name].resolve_repeated_refs()
|
|
else:
|
|
for pat in self.values():
|
|
pat.resolve_repeated_refs()
|
|
return self
|
|
|
|
def subtree(
|
|
self,
|
|
tops: str | Sequence[str],
|
|
) -> Self:
|
|
"""
|
|
Return a new `ILibraryView`, containing only the specified patterns and the patterns they
|
|
reference (recursively).
|
|
Dangling references do not cause an error.
|
|
|
|
Args:
|
|
tops: Name(s) of patterns to keep
|
|
|
|
Returns:
|
|
An object of the same type as `self` containing only `tops` and the patterns they reference.
|
|
"""
|
|
if isinstance(tops, str):
|
|
tops = (tops,)
|
|
|
|
keep = cast('set[str]', self.referenced_patterns(tops) - {None})
|
|
keep |= set(tops)
|
|
|
|
new = type(self)()
|
|
for key in keep & set(self.keys()):
|
|
new._merge(key, self, key)
|
|
return new
|
|
|
|
def prune_empty(
|
|
self,
|
|
repeat: bool = True,
|
|
dangling: dangling_mode_t = 'error',
|
|
) -> set[str]:
|
|
"""
|
|
Delete any empty patterns (i.e. where `Pattern.is_empty` returns `True`).
|
|
|
|
Args:
|
|
repeat: Also recursively delete any patterns which only contain(ed) empty patterns.
|
|
dangling: Passed to `parent_graph()`.
|
|
|
|
Returns:
|
|
A set containing the names of all deleted patterns
|
|
"""
|
|
parent_graph = self.parent_graph(dangling=dangling)
|
|
empty = {name for name, pat in self.items() if pat.is_empty()}
|
|
trimmed = set()
|
|
while empty:
|
|
parents = set()
|
|
for name in empty:
|
|
del self[name]
|
|
for parent in parent_graph[name]:
|
|
del self[parent].refs[name]
|
|
parents |= parent_graph[name]
|
|
|
|
trimmed |= empty
|
|
if not repeat:
|
|
break
|
|
|
|
empty = {parent for parent in parents if self[parent].is_empty()}
|
|
return trimmed
|
|
|
|
def delete(
|
|
self,
|
|
key: str,
|
|
delete_refs: bool = True,
|
|
) -> Self:
|
|
"""
|
|
Delete a pattern and (optionally) all refs pointing to that pattern.
|
|
|
|
Args:
|
|
key: Name of the pattern to be deleted.
|
|
delete_refs: If `True` (default), also delete all refs pointing to the pattern.
|
|
"""
|
|
del self[key]
|
|
if delete_refs:
|
|
for pat in self.values():
|
|
if key in pat.refs:
|
|
del pat.refs[key]
|
|
return self
|
|
|
|
|
|
class LibraryView(ILibraryView):
|
|
"""
|
|
Default implementation for a read-only library.
|
|
|
|
A library is a mapping from unique names (str) to collections of geometry (`Pattern`).
|
|
This library is backed by an arbitrary python object which implements the `Mapping` interface.
|
|
"""
|
|
mapping: Mapping[str, 'Pattern']
|
|
|
|
def __init__(
|
|
self,
|
|
mapping: Mapping[str, 'Pattern'],
|
|
) -> None:
|
|
self.mapping = mapping
|
|
|
|
def __getitem__(self, key: str) -> 'Pattern':
|
|
return self.mapping[key]
|
|
|
|
def __iter__(self) -> Iterator[str]:
|
|
return iter(self.mapping)
|
|
|
|
def __len__(self) -> int:
|
|
return len(self.mapping)
|
|
|
|
def __contains__(self, key: object) -> bool:
|
|
return key in self.mapping
|
|
|
|
def __repr__(self) -> str:
|
|
return f'<LibraryView ({type(self.mapping)}) with keys\n' + pformat(list(self.keys())) + '>'
|
|
|
|
|
|
class Library(ILibrary):
|
|
"""
|
|
Default implementation for a writeable library.
|
|
|
|
A library is a mapping from unique names (str) to collections of geometry (`Pattern`).
|
|
This library is backed by an arbitrary python object which implements the `MutableMapping` interface.
|
|
"""
|
|
mapping: MutableMapping[str, 'Pattern']
|
|
|
|
def __init__(
|
|
self,
|
|
mapping: MutableMapping[str, 'Pattern'] | None = None,
|
|
) -> None:
|
|
if mapping is None:
|
|
self.mapping = {}
|
|
else:
|
|
self.mapping = mapping
|
|
|
|
def __getitem__(self, key: str) -> 'Pattern':
|
|
return self.mapping[key]
|
|
|
|
def __iter__(self) -> Iterator[str]:
|
|
return iter(self.mapping)
|
|
|
|
def __len__(self) -> int:
|
|
return len(self.mapping)
|
|
|
|
def __contains__(self, key: object) -> bool:
|
|
return key in self.mapping
|
|
|
|
def __setitem__(
|
|
self,
|
|
key: str,
|
|
value: 'Pattern | Callable[[], Pattern]',
|
|
) -> None:
|
|
if key in self.mapping:
|
|
raise LibraryError(f'"{key}" already exists in the library. Overwriting is not allowed!')
|
|
|
|
value = value() if callable(value) else value
|
|
self.mapping[key] = value
|
|
|
|
def __delitem__(self, key: str) -> None:
|
|
del self.mapping[key]
|
|
|
|
def _merge(self, key_self: str, other: Mapping[str, 'Pattern'], key_other: str) -> None:
|
|
self[key_self] = other[key_other]
|
|
|
|
def __repr__(self) -> str:
|
|
return f'<Library ({type(self.mapping)}) with keys\n' + pformat(list(self.keys())) + '>'
|
|
|
|
@classmethod
|
|
def mktree(cls: type[Self], name: str) -> tuple[Self, 'Pattern']:
|
|
"""
|
|
Create a new Library and immediately add a pattern
|
|
|
|
Args:
|
|
name: The name for the new pattern (usually the name of the topcell).
|
|
|
|
Returns:
|
|
The newly created `Library` and the newly created `Pattern`
|
|
"""
|
|
from .pattern import Pattern #noqa: PLC0415
|
|
tree = cls()
|
|
pat = Pattern()
|
|
tree[name] = pat
|
|
return tree, pat
|
|
|
|
|
|
@dataclass
|
|
class _BuildRecipe:
|
|
""" Captured deferred call to a pattern factory. """
|
|
func: Callable[..., 'Pattern']
|
|
args: tuple[Any, ...]
|
|
kwargs: dict[str, Any]
|
|
explicit_dependencies: tuple[str, ...] = ()
|
|
|
|
def depends_on(self, *names: str) -> '_BuildRecipe':
|
|
self.explicit_dependencies += tuple(names)
|
|
return self
|
|
|
|
|
|
def cell(func: Callable[..., 'Pattern']) -> Callable[..., _BuildRecipe]:
|
|
"""
|
|
Wrap a plain pattern factory so calls return deferred build recipes.
|
|
|
|
Use as either `cell(fn)(...)` or `@cell`.
|
|
"""
|
|
@wraps(func)
|
|
def wrapper(*args: Any, **kwargs: Any) -> _BuildRecipe:
|
|
return _BuildRecipe(func=func, args=args, kwargs=kwargs)
|
|
|
|
return wrapper
|
|
|
|
|
|
class BuildCellsView:
|
|
"""
|
|
Attribute-based declaration namespace for `BuildLibrary`.
|
|
|
|
This is the ergonomic authoring surface exposed as `builder.cells`. It is
|
|
intentionally write-focused: attribute assignment and deletion register
|
|
declarations, while attribute reads fail with guidance to build first and
|
|
use the returned library.
|
|
"""
|
|
|
|
def __init__(self, library: 'BuildLibrary') -> None:
|
|
object.__setattr__(self, '_library', library)
|
|
|
|
def __getattr__(self, name: str) -> 'Pattern':
|
|
raise BuildError(
|
|
f'BuildLibrary.cells.{name} is write-only during authoring. '
|
|
'Call build() and index the returned library instead.'
|
|
)
|
|
|
|
def __setattr__(self, name: str, value: 'Pattern | _BuildRecipe') -> None:
|
|
if name.startswith('_'):
|
|
object.__setattr__(self, name, value)
|
|
return
|
|
self._library[name] = value
|
|
|
|
def __delattr__(self, name: str) -> None:
|
|
if name.startswith('_'):
|
|
raise AttributeError(name)
|
|
del self._library[name]
|
|
|
|
|
|
class BuildLibrary(ILibrary):
|
|
"""
|
|
Two-phase declaration surface for mixed imported/generated libraries.
|
|
|
|
A `BuildLibrary` collects three kinds of inputs:
|
|
- direct declared `Pattern` objects
|
|
- deferred recipes created with `cell(...)`
|
|
- imported source-backed library views added with `add_source(...)`
|
|
|
|
The builder itself is not a normal readable library during authoring.
|
|
Instead, `validate()` and `build()` create a temporary build-session library
|
|
that recipes can read from and write helper cells into while dependencies
|
|
are resolved. `build()` then freezes the builder on success and returns the
|
|
built library plus a `BuildReport`.
|
|
"""
|
|
|
|
def __init__(self) -> None:
|
|
self.cells = BuildCellsView(self)
|
|
self._frozen = False
|
|
self._declarations: dict[str, Pattern | _BuildRecipe] = {}
|
|
self._sources: list[tuple[ILibraryView, dict[str, str]]] = []
|
|
self._names: dict[str, None] = {}
|
|
|
|
def _active_session(self) -> '_BuildSessionLibrary | None':
|
|
sessions = _ACTIVE_BUILD_SESSIONS.get()
|
|
if sessions is None:
|
|
return None
|
|
return sessions.get(id(self))
|
|
|
|
def _require_active_session(self, operation: str) -> '_BuildSessionLibrary':
|
|
session = self._active_session()
|
|
if session is None:
|
|
raise BuildError(
|
|
f'BuildLibrary.{operation}() is only available while validate() or build() is running. '
|
|
'Use the built output library for reads.'
|
|
)
|
|
return session
|
|
|
|
def _assert_editable(self) -> None:
|
|
if self._frozen:
|
|
raise BuildError('This BuildLibrary has already been built successfully and is now frozen.')
|
|
|
|
def __iter__(self) -> Iterator[str]:
|
|
session = self._active_session()
|
|
if session is not None:
|
|
return iter(session)
|
|
return iter(self._names)
|
|
|
|
def __len__(self) -> int:
|
|
session = self._active_session()
|
|
if session is not None:
|
|
return len(session)
|
|
return len(self._names)
|
|
|
|
def __contains__(self, key: object) -> bool:
|
|
session = self._active_session()
|
|
if session is not None:
|
|
return key in session
|
|
return key in self._names
|
|
|
|
def __getitem__(self, key: str) -> 'Pattern':
|
|
return self._require_active_session('__getitem__')[key]
|
|
|
|
def __setitem__(
|
|
self,
|
|
key: str,
|
|
value: 'Pattern | _BuildRecipe',
|
|
) -> None:
|
|
session = self._active_session()
|
|
if session is not None:
|
|
session[key] = value
|
|
return
|
|
|
|
self._assert_editable()
|
|
if key in self._names:
|
|
raise LibraryError(f'"{key}" already exists in the builder. Overwriting is not allowed!')
|
|
|
|
if isinstance(value, _BuildRecipe):
|
|
declaration = value
|
|
else:
|
|
if callable(value):
|
|
raise TypeError('BuildLibrary recipes must be wrapped with cell(fn)(...) or @cell.')
|
|
declaration = value
|
|
|
|
self._declarations[key] = declaration
|
|
self._names[key] = None
|
|
|
|
def __delitem__(self, key: str) -> None:
|
|
session = self._active_session()
|
|
if session is not None:
|
|
del session[key]
|
|
return
|
|
|
|
self._assert_editable()
|
|
if key not in self._declarations:
|
|
raise KeyError(key)
|
|
del self._declarations[key]
|
|
del self._names[key]
|
|
|
|
def _merge(self, key_self: str, other: Mapping[str, 'Pattern'], key_other: str) -> None:
|
|
session = self._active_session()
|
|
if session is not None:
|
|
session._merge(key_self, other, key_other)
|
|
return
|
|
self[key_self] = copy.deepcopy(other[key_other])
|
|
|
|
def add(
|
|
self,
|
|
other: Mapping[str, 'Pattern'],
|
|
rename_theirs: Callable[['ILibraryView', str], str] = _rename_patterns,
|
|
mutate_other: bool = False,
|
|
) -> dict[str, str]:
|
|
from .pattern import map_targets # noqa: PLC0415
|
|
|
|
session = self._active_session()
|
|
if session is not None:
|
|
return session.add(other, rename_theirs=rename_theirs, mutate_other=mutate_other)
|
|
|
|
self._assert_editable()
|
|
|
|
source_backed = isinstance(other, ILibraryView) and not isinstance(other, Library | LibraryView)
|
|
if source_backed:
|
|
if mutate_other:
|
|
raise BuildError('BuildLibrary.add(..., mutate_other=True) is not supported for source-backed inputs.')
|
|
return self.add_source(
|
|
other,
|
|
rename_theirs = rename_theirs,
|
|
rename_when = 'conflict',
|
|
)
|
|
|
|
source_order = tuple(other.keys())
|
|
source_to_visible = _plan_source_names(
|
|
self,
|
|
source_order,
|
|
self._names,
|
|
rename_theirs = rename_theirs,
|
|
rename_when = 'conflict',
|
|
)
|
|
rename_map = _source_rename_map(source_to_visible)
|
|
|
|
if mutate_other:
|
|
temp = other
|
|
else:
|
|
temp = Library(copy.deepcopy(dict(other)))
|
|
|
|
for source_name in source_order:
|
|
visible_name = source_to_visible[source_name]
|
|
pattern = temp[source_name]
|
|
if rename_map:
|
|
pattern.refs = map_targets(
|
|
pattern.refs,
|
|
lambda target: cast('dict[str | None, str | None]', rename_map).get(target, target),
|
|
)
|
|
self[visible_name] = pattern
|
|
|
|
return rename_map
|
|
|
|
def __lshift__(self, other: TreeView) -> str:
|
|
session = self._active_session()
|
|
if session is not None:
|
|
return session << other
|
|
|
|
self._assert_editable()
|
|
if len(other) == 1:
|
|
name = next(iter(other))
|
|
elif isinstance(other, ILibraryView) and not isinstance(other, Library | LibraryView):
|
|
source_order = other.source_order()
|
|
child_graph = other.child_graph(dangling='include')
|
|
referenced = set().union(*child_graph.values()) if child_graph else set()
|
|
tops = [candidate for candidate in source_order if candidate not in referenced]
|
|
if len(tops) != 1:
|
|
raise LibraryError(f'Asked for the single topcell, but found the following: {pformat(tops)}')
|
|
name = tops[0]
|
|
else:
|
|
return super().__lshift__(other)
|
|
|
|
rename_map = self.add(other)
|
|
return rename_map.get(name, name)
|
|
|
|
def __le__(self, other: Mapping[str, 'Pattern']) -> Abstract:
|
|
if self._active_session() is not None:
|
|
return super().__le__(other)
|
|
raise BuildError('BuildLibrary.__le__() is only available while validate() or build() is running.')
|
|
|
|
def rename(
|
|
self,
|
|
old_name: str,
|
|
new_name: str,
|
|
move_references: bool = False,
|
|
) -> Self:
|
|
"""
|
|
Rename a helper cell during an active build session.
|
|
|
|
During authoring, declared cells must be registered under their
|
|
intended final names and imported source cells must be renamed through
|
|
`add_source(...)`.
|
|
"""
|
|
session = self._active_session()
|
|
if session is not None:
|
|
session.rename(old_name, new_name, move_references=move_references)
|
|
return self
|
|
|
|
self._assert_editable()
|
|
if old_name == new_name:
|
|
return self
|
|
if old_name in self._declarations:
|
|
raise BuildError(
|
|
f'Cannot rename declared build cell "{old_name}" during authoring. '
|
|
'Register it under the intended final name instead.'
|
|
)
|
|
if old_name not in self._names:
|
|
raise LibraryError(f'"{old_name}" does not exist in the builder.')
|
|
raise BuildError(
|
|
f'Cannot rename imported source cell "{old_name}" during authoring. '
|
|
'Choose visible source names with add_source(..., rename_theirs=..., rename_when=...).'
|
|
)
|
|
|
|
def abstract(self, name: str) -> Abstract:
|
|
return self._require_active_session('abstract').abstract(name)
|
|
|
|
def resolve(
|
|
self,
|
|
other: 'Abstract | str | Pattern | TreeView',
|
|
append: bool = False,
|
|
) -> 'Abstract | Pattern':
|
|
return self._require_active_session('resolve').resolve(other, append=append)
|
|
|
|
def add_source(
|
|
self,
|
|
source: Mapping[str, 'Pattern'] | ILibraryView,
|
|
*,
|
|
rename_theirs: Callable[[ILibraryView, str], str] | None = None,
|
|
rename_when: Literal['conflict', 'always'] = 'conflict',
|
|
) -> dict[str, str]:
|
|
"""
|
|
Register an imported source-backed library with the builder.
|
|
|
|
The source is not materialized immediately. Its names are scanned once
|
|
to reserve visible builder names, then the source is read again when a
|
|
build session starts. The source's cell membership must not be
|
|
structurally mutated between `add_source()` and `build()`/`validate()`.
|
|
Source cells may be renamed on entry to avoid collisions with existing
|
|
declarations or other imported sources.
|
|
|
|
Args:
|
|
rename_theirs: Function used to choose visible names for imported
|
|
source cells.
|
|
rename_when: If `'conflict'`, only conflicting names are renamed.
|
|
If `'always'`, every imported source name is passed through
|
|
`rename_theirs`.
|
|
|
|
Returns:
|
|
Mapping of `{source_name: visible_name}` for imported names that
|
|
were renamed while being added.
|
|
"""
|
|
if self._active_session() is not None:
|
|
raise BuildError('BuildLibrary.add_source() is only available while authoring, not during validate() or build().')
|
|
self._assert_editable()
|
|
|
|
view = source if isinstance(source, ILibraryView) else LibraryView(source)
|
|
source_order = tuple(view.source_order())
|
|
source_to_visible = _plan_source_names(
|
|
self,
|
|
source_order,
|
|
self._names,
|
|
rename_theirs = rename_theirs,
|
|
rename_when = rename_when,
|
|
)
|
|
|
|
self._sources.append((view, dict(source_to_visible)))
|
|
for source_name in source_order:
|
|
visible = source_to_visible[source_name]
|
|
self._names[visible] = None
|
|
return _source_rename_map(source_to_visible)
|
|
|
|
def validate(
|
|
self,
|
|
names: Sequence[str] | None = None,
|
|
*,
|
|
allow_dangling: bool = False,
|
|
) -> BuildReport:
|
|
"""
|
|
Run the full build logic and return a `BuildReport` without producing output.
|
|
|
|
This is a dry run over the same dependency resolution and recipe
|
|
execution path used by `build()`. Any generated library is discarded
|
|
after validation completes.
|
|
"""
|
|
_session, report = self._run_build(names=names, allow_dangling=allow_dangling)
|
|
return report
|
|
|
|
def build(
|
|
self,
|
|
*,
|
|
output: Literal['overlay', 'library'] = 'overlay',
|
|
allow_dangling: bool = False,
|
|
) -> tuple[ILibrary, BuildReport]:
|
|
"""
|
|
Materialize declarations and return a usable output library plus report.
|
|
|
|
Args:
|
|
output: `'overlay'` preserves imported source-backed cells where
|
|
possible, while `'library'` eagerly materializes the full
|
|
result.
|
|
allow_dangling: If `False`, fail the build when the completed
|
|
library still contains dangling references.
|
|
"""
|
|
if output not in ('overlay', 'library'):
|
|
raise ValueError(f'Unknown build output mode: {output!r}')
|
|
self._assert_editable()
|
|
session, report = self._run_build(names=None, allow_dangling=allow_dangling)
|
|
if output == 'library':
|
|
built_output = session.to_library()
|
|
else:
|
|
built_output = session.to_overlay()
|
|
self._frozen = True
|
|
return built_output, report
|
|
|
|
def _run_build(
|
|
self,
|
|
*,
|
|
names: Sequence[str] | None,
|
|
allow_dangling: bool,
|
|
) -> tuple['_BuildSessionLibrary', BuildReport]:
|
|
roots = tuple(dict.fromkeys(names if names is not None else self._declarations.keys()))
|
|
unknown = [name for name in roots if name not in self._names]
|
|
if unknown:
|
|
raise BuildError(f'Unknown build roots requested: {unknown}')
|
|
|
|
session = _BuildSessionLibrary(self)
|
|
sessions = dict(_ACTIVE_BUILD_SESSIONS.get() or {})
|
|
sessions[id(self)] = session
|
|
token = _ACTIVE_BUILD_SESSIONS.set(sessions)
|
|
try:
|
|
session.materialize_many(roots)
|
|
if not allow_dangling:
|
|
session.child_graph(dangling='error')
|
|
finally:
|
|
_ACTIVE_BUILD_SESSIONS.reset(token)
|
|
|
|
report = session.build_report(roots)
|
|
return session, report
|
|
|
|
|
|
class _BuildSessionLibrary(ILibrary):
|
|
"""
|
|
Internal overlay-backed library used while a `BuildLibrary` is executing.
|
|
|
|
This object provides the mutable-library surface that recipes expect while
|
|
also tracking declared-cell dependencies, helper-cell provenance, and
|
|
imported source cells. It exists only for the duration of a validation or
|
|
build run.
|
|
"""
|
|
|
|
def __init__(self, builder: BuildLibrary) -> None:
|
|
from .file.gdsii_lazy_core import OverlayLibrary # noqa: PLC0415
|
|
|
|
self._builder = builder
|
|
self._overlay = OverlayLibrary()
|
|
self._built: set[str] = set()
|
|
self._declared_stack: list[str] = []
|
|
self._names = dict(builder._names)
|
|
self._provenance: dict[str, CellProvenance] = {}
|
|
self._dependency_graph: defaultdict[str, set[str]] = defaultdict(set)
|
|
self._install_sources()
|
|
|
|
def _install_sources(self) -> None:
|
|
for source_library, source_to_visible in self._builder._sources:
|
|
source_order = source_library.source_order()
|
|
expected_names = set(source_to_visible)
|
|
actual_names = set(source_order)
|
|
if actual_names != expected_names:
|
|
added_names = sorted(actual_names - expected_names)
|
|
removed_names = sorted(expected_names - actual_names)
|
|
detail = []
|
|
if added_names:
|
|
detail.append(f'added={added_names}')
|
|
if removed_names:
|
|
detail.append(f'removed={removed_names}')
|
|
raise BuildError(
|
|
'Imported source library changed after add_source() was called '
|
|
f'({", ".join(detail)}). '
|
|
'Do not structurally mutate source libraries between add_source() and build()/validate().'
|
|
)
|
|
|
|
def rename_source(_lib: ILibraryView, name: str, *, mapping: Mapping[str, str] = source_to_visible) -> str:
|
|
return mapping[name]
|
|
|
|
self._overlay.add_source(
|
|
source_library,
|
|
rename_theirs = rename_source,
|
|
rename_when = 'always',
|
|
)
|
|
|
|
for source_name in source_order:
|
|
visible_name = source_to_visible[source_name]
|
|
self._provenance[visible_name] = CellProvenance(
|
|
requested_name = source_name,
|
|
kind = 'source',
|
|
owner_declared_name = None,
|
|
build_chain = (),
|
|
)
|
|
|
|
def __iter__(self) -> Iterator[str]:
|
|
return iter(self._names)
|
|
|
|
def __len__(self) -> int:
|
|
return len(self._names)
|
|
|
|
def __contains__(self, key: object) -> bool:
|
|
return key in self._names
|
|
|
|
def _current_declared(self) -> str | None:
|
|
if not self._declared_stack:
|
|
return None
|
|
return self._declared_stack[-1]
|
|
|
|
def _record_dependency(self, target: str) -> None:
|
|
current = self._current_declared()
|
|
if current is None or current == target or target not in self._builder._declarations:
|
|
return
|
|
self._dependency_graph[current].add(target)
|
|
|
|
def _guard_mutable_output_name(self, key: str, *, operation: str) -> None:
|
|
if key in self._builder._declarations:
|
|
raise BuildError(f'Cannot {operation} declared build cell "{key}" during an active build session.')
|
|
|
|
provenance = self._provenance.get(key)
|
|
if provenance is not None and provenance.kind == 'source':
|
|
raise BuildError(f'Cannot {operation} imported source cell "{key}" during an active build session.')
|
|
|
|
def rename(
|
|
self,
|
|
old_name: str,
|
|
new_name: str,
|
|
move_references: bool = False,
|
|
) -> Self:
|
|
if old_name == new_name:
|
|
return self
|
|
if old_name not in self._overlay:
|
|
if old_name in self._builder._declarations:
|
|
self._guard_mutable_output_name(old_name, operation='rename')
|
|
raise LibraryError(f'"{old_name}" does not exist in the library.')
|
|
|
|
self._guard_mutable_output_name(old_name, operation='rename')
|
|
if new_name in self._names:
|
|
raise LibraryError(f'"{new_name}" already exists in the library.')
|
|
|
|
self._overlay.rename(old_name, new_name, move_references=move_references)
|
|
self._names = {
|
|
new_name if name == old_name else name: None
|
|
for name in self._names
|
|
}
|
|
|
|
provenance = self._provenance.pop(old_name)
|
|
self._provenance[new_name] = provenance
|
|
return self
|
|
|
|
def __getitem__(self, key: str) -> 'Pattern':
|
|
if key in self._builder._declarations:
|
|
self._record_dependency(key)
|
|
self._ensure_declared(key)
|
|
return self._overlay[key]
|
|
|
|
def __setitem__(
|
|
self,
|
|
key: str,
|
|
value: 'Pattern | Callable[[], Pattern]',
|
|
) -> None:
|
|
if key in self._overlay:
|
|
raise LibraryError(f'"{key}" already exists in the library. Overwriting is not allowed!')
|
|
current = self._current_declared()
|
|
if key in self._builder._declarations and key != current:
|
|
raise LibraryError(f'"{key}" is reserved for a declared cell and cannot be used as a helper name.')
|
|
|
|
pattern = value() if callable(value) else value
|
|
self._overlay[key] = pattern
|
|
self._names.setdefault(key, None)
|
|
|
|
kind: Literal['declared', 'helper']
|
|
if current is not None and key == current:
|
|
kind = 'declared'
|
|
else:
|
|
kind = 'helper'
|
|
|
|
self._provenance[key] = CellProvenance(
|
|
requested_name = key,
|
|
kind = kind,
|
|
owner_declared_name = current if kind == 'helper' else key,
|
|
build_chain = tuple(self._declared_stack),
|
|
)
|
|
|
|
def __delitem__(self, key: str) -> None:
|
|
if key not in self._overlay:
|
|
if key in self._builder._declarations:
|
|
self._guard_mutable_output_name(key, operation='delete')
|
|
raise KeyError(key)
|
|
|
|
self._guard_mutable_output_name(key, operation='delete')
|
|
if key in self._overlay:
|
|
del self._overlay[key]
|
|
self._names.pop(key, None)
|
|
self._provenance.pop(key, None)
|
|
|
|
def _merge(self, key_self: str, other: Mapping[str, 'Pattern'], key_other: str) -> None:
|
|
self[key_self] = copy.deepcopy(other[key_other])
|
|
|
|
def add(
|
|
self,
|
|
other: Mapping[str, 'Pattern'],
|
|
rename_theirs: Callable[['ILibraryView', str], str] = _rename_patterns,
|
|
mutate_other: bool = False,
|
|
) -> dict[str, str]:
|
|
rename_map = super().add(other, rename_theirs=rename_theirs, mutate_other=mutate_other)
|
|
|
|
current = self._current_declared()
|
|
for old_name, new_name in rename_map.items():
|
|
if new_name in self._provenance:
|
|
self._provenance[new_name] = replace(
|
|
self._provenance[new_name],
|
|
requested_name = old_name,
|
|
owner_declared_name = current if current is not None else self._provenance[new_name].owner_declared_name,
|
|
)
|
|
return rename_map
|
|
|
|
def _wrap_error(self, name: str, exc: Exception) -> BuildError:
|
|
chain = tuple(self._declared_stack)
|
|
msg = [f'Failed while building declared cell "{name}"']
|
|
if chain:
|
|
msg.append(f'Dependency chain: {" -> ".join(chain)}')
|
|
msg.append(f'Cause: {exc}')
|
|
return BuildError('\n'.join(msg))
|
|
|
|
def _ensure_named(self, name: str) -> None:
|
|
if name in self._builder._declarations:
|
|
self._record_dependency(name)
|
|
self._ensure_declared(name)
|
|
return
|
|
if name in self._overlay:
|
|
return
|
|
raise BuildError(f'Missing dependency "{name}"')
|
|
|
|
def _ensure_declared(self, name: str) -> None:
|
|
from .pattern import Pattern # noqa: PLC0415
|
|
|
|
if name in self._built:
|
|
return
|
|
if name in self._declared_stack:
|
|
chain = ' -> '.join(self._declared_stack + [name])
|
|
raise BuildError(f'Cycle detected while building declared cells: {chain}')
|
|
|
|
declaration = self._builder._declarations[name]
|
|
self._declared_stack.append(name)
|
|
try:
|
|
if isinstance(declaration, _BuildRecipe):
|
|
for dep in declaration.explicit_dependencies:
|
|
self._ensure_named(dep)
|
|
pattern = declaration.func(*declaration.args, **declaration.kwargs)
|
|
if not isinstance(pattern, Pattern):
|
|
raise BuildError(f'Recipe for "{name}" returned {type(pattern).__name__}, expected Pattern')
|
|
else:
|
|
pattern = declaration.deepcopy()
|
|
|
|
if name in self._overlay:
|
|
if self._overlay[name] is not pattern:
|
|
raise BuildError(
|
|
f'Recipe for "{name}" wrote a different pattern into the session under its own name.'
|
|
)
|
|
else:
|
|
self[name] = pattern
|
|
self._built.add(name)
|
|
except Exception as exc:
|
|
raise self._wrap_error(name, exc) from exc
|
|
finally:
|
|
self._declared_stack.pop()
|
|
|
|
def materialize_many(self, names: Sequence[str]) -> None:
|
|
for name in dict.fromkeys(names):
|
|
self._ensure_named(name)
|
|
|
|
def source_order(self) -> tuple[str, ...]:
|
|
return self._overlay.source_order()
|
|
|
|
def child_graph(
|
|
self,
|
|
dangling: dangling_mode_t = 'error',
|
|
) -> dict[str, set[str]]:
|
|
return self._overlay.child_graph(dangling=dangling)
|
|
|
|
def parent_graph(
|
|
self,
|
|
dangling: dangling_mode_t = 'error',
|
|
) -> dict[str, set[str]]:
|
|
return self._overlay.parent_graph(dangling=dangling)
|
|
|
|
def build_report(self, requested_roots: Sequence[str]) -> BuildReport:
|
|
dependency_graph = {
|
|
name: frozenset(self._dependency_graph.get(name, set()))
|
|
for name in self._builder._declarations
|
|
if name in self._dependency_graph or name in requested_roots
|
|
}
|
|
return BuildReport(
|
|
requested_roots = tuple(dict.fromkeys(requested_roots)),
|
|
provenance = dict(self._provenance),
|
|
dependency_graph = dependency_graph,
|
|
)
|
|
|
|
def to_overlay(self) -> ILibrary:
|
|
return self._overlay
|
|
|
|
def to_library(self) -> Library:
|
|
mapping = {name: self._overlay[name] for name in self._overlay.source_order()}
|
|
return Library(mapping)
|
|
|
|
|
|
class LazyLibrary(ILibrary):
|
|
"""
|
|
This class is usually used to create a library of Patterns by mapping names to
|
|
functions which generate or load the relevant `Pattern` object as-needed.
|
|
|
|
TODO: lots of stuff causes recursive loads (e.g. data_to_ports?). What should you avoid?
|
|
"""
|
|
mapping: dict[str, Callable[[], 'Pattern']]
|
|
cache: dict[str, 'Pattern']
|
|
_lookups_in_progress: list[str]
|
|
|
|
def __init__(self) -> None:
|
|
self.mapping = {}
|
|
self.cache = {}
|
|
self._lookups_in_progress = []
|
|
|
|
def __setitem__(
|
|
self,
|
|
key: str,
|
|
value: 'Pattern | Callable[[], Pattern]',
|
|
) -> None:
|
|
if key in self.mapping:
|
|
raise LibraryError(f'"{key}" already exists in the library. Overwriting is not allowed!')
|
|
|
|
if callable(value):
|
|
value_func = value
|
|
else:
|
|
value_func = lambda: cast('Pattern', value) # noqa: E731
|
|
|
|
self.mapping[key] = value_func
|
|
if key in self.cache:
|
|
del self.cache[key]
|
|
|
|
def __delitem__(self, key: str) -> None:
|
|
del self.mapping[key]
|
|
if key in self.cache:
|
|
del self.cache[key]
|
|
|
|
def __getitem__(self, key: str) -> 'Pattern':
|
|
logger.debug(f'loading {key}')
|
|
if key in self.cache:
|
|
logger.debug(f'found {key} in cache')
|
|
return self.cache[key]
|
|
|
|
if key in self._lookups_in_progress:
|
|
chain = ' -> '.join(self._lookups_in_progress + [key])
|
|
raise LibraryError(
|
|
f'Detected circular reference or recursive lookup of "{key}".\n'
|
|
f'Lookup chain: {chain}\n'
|
|
'This may be caused by an invalid (cyclical) reference, or buggy code.\n'
|
|
'If you are lazy-loading a file, try a non-lazy load and check for reference cycles.'
|
|
)
|
|
|
|
self._lookups_in_progress.append(key)
|
|
try:
|
|
func = self.mapping[key]
|
|
pat = func()
|
|
finally:
|
|
self._lookups_in_progress.pop()
|
|
self.cache[key] = pat
|
|
return pat
|
|
|
|
def __iter__(self) -> Iterator[str]:
|
|
return iter(self.mapping)
|
|
|
|
def __len__(self) -> int:
|
|
return len(self.mapping)
|
|
|
|
def __contains__(self, key: object) -> bool:
|
|
return key in self.mapping
|
|
|
|
def _merge(self, key_self: str, other: Mapping[str, 'Pattern'], key_other: str) -> None:
|
|
if isinstance(other, LazyLibrary):
|
|
self.mapping[key_self] = other.mapping[key_other]
|
|
if key_other in other.cache:
|
|
self.cache[key_self] = other.cache[key_other]
|
|
else:
|
|
self[key_self] = other[key_other]
|
|
|
|
def __repr__(self) -> str:
|
|
return '<LazyLibrary with keys\n' + pformat(list(self.keys())) + '>'
|
|
|
|
def rename(
|
|
self,
|
|
old_name: str,
|
|
new_name: str,
|
|
move_references: bool = False,
|
|
) -> Self:
|
|
"""
|
|
Rename a pattern.
|
|
|
|
Args:
|
|
old_name: Current name for the pattern
|
|
new_name: New name for the pattern
|
|
move_references: Whether to scan all refs in the pattern and
|
|
move them to point to `new_name` as necessary.
|
|
Default `False`.
|
|
|
|
Returns:
|
|
self
|
|
"""
|
|
if old_name not in self.mapping:
|
|
raise LibraryError(f'"{old_name}" does not exist in the library.')
|
|
if old_name == new_name:
|
|
return self
|
|
|
|
self[new_name] = self.mapping[old_name] # copy over function
|
|
if old_name in self.cache:
|
|
self.cache[new_name] = self.cache[old_name]
|
|
del self[old_name]
|
|
|
|
if move_references:
|
|
self.move_references(old_name, new_name)
|
|
|
|
return self
|
|
|
|
def move_references(self, old_target: str, new_target: str) -> Self:
|
|
"""
|
|
Change all references pointing at `old_target` into references pointing at `new_target`.
|
|
|
|
Args:
|
|
old_target: Current reference target
|
|
new_target: New target for the reference
|
|
|
|
Returns:
|
|
self
|
|
"""
|
|
if old_target == new_target:
|
|
return self
|
|
|
|
self.precache()
|
|
for pattern in self.cache.values():
|
|
if old_target in pattern.refs:
|
|
pattern.refs[new_target].extend(pattern.refs[old_target])
|
|
del pattern.refs[old_target]
|
|
return self
|
|
|
|
def precache(self) -> Self:
|
|
"""
|
|
Force all patterns into the cache
|
|
|
|
Returns:
|
|
self
|
|
"""
|
|
for key in self.mapping:
|
|
_ = self[key] # want to trigger our own __getitem__
|
|
return self
|
|
|
|
def __deepcopy__(self, memo: dict | None = None) -> 'LazyLibrary':
|
|
raise LibraryError('LazyLibrary cannot be deepcopied (deepcopy doesn\'t descend into closures)')
|
|
|
|
|
|
class AbstractView(Mapping[str, Abstract]):
|
|
"""
|
|
A read-only mapping from names to `Abstract` objects.
|
|
|
|
This is usually just used as a shorthand for repeated calls to `library.abstract()`.
|
|
"""
|
|
library: ILibraryView
|
|
|
|
def __init__(self, library: ILibraryView) -> None:
|
|
self.library = library
|
|
|
|
def __getitem__(self, key: str) -> Abstract:
|
|
return self.library.abstract(key)
|
|
|
|
def __iter__(self) -> Iterator[str]:
|
|
return self.library.__iter__()
|
|
|
|
def __len__(self) -> int:
|
|
return self.library.__len__()
|
|
|
|
|
|
def b64suffix(ii: int) -> str:
|
|
"""
|
|
Turn an integer into a base64-equivalent suffix.
|
|
|
|
This could be done with base64.b64encode, but this way is faster for many small `ii`.
|
|
"""
|
|
def i2a(nn: int) -> str:
|
|
return 'ABCDEFGHIJKLMNOPQRSTUVWXYZabcdefghijklmnopqrstuvwxyz0123456789$?'[nn]
|
|
|
|
parts = ['$', i2a(ii % 64)]
|
|
ii >>= 6
|
|
while ii:
|
|
parts.append(i2a(ii % 64))
|
|
ii >>= 6
|
|
return ''.join(parts)
|