[OverlayLibrary / PortsLibraryView] move into masque.library
This commit is contained in:
parent
9a39a436b2
commit
4c64352152
7 changed files with 557 additions and 573 deletions
|
|
@ -63,6 +63,8 @@ from .library import (
|
|||
ILibrary as ILibrary,
|
||||
LibraryView as LibraryView,
|
||||
Library as Library,
|
||||
OverlayLibrary as OverlayLibrary,
|
||||
PortsLibraryView as PortsLibraryView,
|
||||
BuildLibrary as BuildLibrary,
|
||||
BuildReport as BuildReport,
|
||||
CellProvenance as CellProvenance,
|
||||
|
|
|
|||
|
|
@ -8,7 +8,7 @@ This module provides the non-Arrow half of Masque's lazy GDS architecture:
|
|||
- cells are materialized on demand through the classic `gdsii` decoder
|
||||
whenever a caller indexes the lazy view
|
||||
- the source can be wrapped in `PortsLibraryView` or merged through
|
||||
`OverlayLibrary`, both of which live in `gdsii_lazy_core`
|
||||
`OverlayLibrary`
|
||||
|
||||
The public surface intentionally parallels `gdsii_lazy_arrow` closely so that
|
||||
callers can swap between the classic and Arrow-backed implementations with
|
||||
|
|
@ -17,9 +17,8 @@ minimal changes.
|
|||
from __future__ import annotations
|
||||
|
||||
from dataclasses import dataclass
|
||||
from typing import IO, Any, cast
|
||||
from typing import IO, TYPE_CHECKING, Any, cast
|
||||
from collections import defaultdict
|
||||
from collections.abc import Iterator, Sequence
|
||||
import gzip
|
||||
import io
|
||||
import logging
|
||||
|
|
@ -28,17 +27,27 @@ import pathlib
|
|||
|
||||
import klamath
|
||||
import numpy
|
||||
from numpy.typing import NDArray
|
||||
from klamath import records
|
||||
|
||||
from . import gdsii
|
||||
from .utils import is_gzipped
|
||||
from .gdsii_lazy_core import OverlayLibrary, PortsLibraryView, _pattern_children, write, writefile
|
||||
from .gdsii_lazy_core import write as write, writefile as writefile
|
||||
from ..error import LibraryError
|
||||
from ..library import ILibraryView, LibraryView, dangling_mode_t
|
||||
from ..pattern import Pattern
|
||||
from ..library import (
|
||||
ILibraryView,
|
||||
LibraryView,
|
||||
PortsLibraryView,
|
||||
dangling_mode_t,
|
||||
)
|
||||
from ..utils import apply_transforms
|
||||
|
||||
if TYPE_CHECKING:
|
||||
from collections.abc import Iterator, Sequence
|
||||
|
||||
from numpy.typing import NDArray
|
||||
|
||||
from ..pattern import Pattern
|
||||
|
||||
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
|
|
@ -76,7 +85,7 @@ def _open_source_stream(
|
|||
with gzip.open(path, mode='rb') as stream:
|
||||
data = stream.read()
|
||||
return _SourceHandle(path=path, stream=io.BytesIO(data))
|
||||
stream = cast('IO[bytes]', gzip.open(path, mode='rb'))
|
||||
stream = cast('IO[bytes]', gzip.open(path, mode='rb')) # noqa: SIM115
|
||||
return _SourceHandle(path=path, stream=stream)
|
||||
|
||||
if use_mmap:
|
||||
|
|
@ -216,7 +225,7 @@ class GdsLibrarySource(ILibraryView):
|
|||
graph: dict[str, set[str]] = {}
|
||||
for name in self._cell_order:
|
||||
if name in self._cache:
|
||||
graph[name] = _pattern_children(self._cache[name])
|
||||
graph[name] = {child for child, refs in self._cache[name].refs.items() if child is not None and refs}
|
||||
else:
|
||||
graph[name] = self._raw_children(name)
|
||||
|
||||
|
|
|
|||
|
|
@ -7,25 +7,34 @@ keeps its current behavior and performance profile.
|
|||
from __future__ import annotations
|
||||
|
||||
from dataclasses import dataclass
|
||||
from typing import IO, Any, cast
|
||||
from typing import IO, TYPE_CHECKING, Any, cast
|
||||
from collections import defaultdict
|
||||
from collections.abc import Iterator, Sequence
|
||||
import gzip
|
||||
import logging
|
||||
import mmap
|
||||
import pathlib
|
||||
|
||||
import numpy
|
||||
from numpy.typing import NDArray
|
||||
import pyarrow
|
||||
|
||||
from . import gdsii_arrow
|
||||
from .utils import is_gzipped
|
||||
from .gdsii_lazy_core import OverlayLibrary, PortsLibraryView, _pattern_children, write, writefile
|
||||
from ..library import ILibraryView, LibraryView, dangling_mode_t
|
||||
from ..pattern import Pattern
|
||||
from .gdsii_lazy_core import write as write, writefile as writefile
|
||||
from ..library import (
|
||||
ILibraryView,
|
||||
LibraryView,
|
||||
PortsLibraryView,
|
||||
dangling_mode_t,
|
||||
)
|
||||
from ..utils import apply_transforms
|
||||
|
||||
if TYPE_CHECKING:
|
||||
from collections.abc import Iterator, Sequence
|
||||
|
||||
from numpy.typing import NDArray
|
||||
import pyarrow
|
||||
|
||||
from ..pattern import Pattern
|
||||
|
||||
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
|
|
@ -343,7 +352,7 @@ class ArrowLibrary(ILibraryView):
|
|||
graph: dict[str, set[str]] = {}
|
||||
for name in self._payload.cell_order:
|
||||
if name in self._cache:
|
||||
graph[name] = _pattern_children(self._cache[name])
|
||||
graph[name] = {child for child, refs in self._cache[name].refs.items() if child is not None and refs}
|
||||
else:
|
||||
graph[name] = self._raw_children(name)
|
||||
|
||||
|
|
|
|||
|
|
@ -1,558 +1,34 @@
|
|||
"""
|
||||
Shared helpers for source-backed lazy GDS views.
|
||||
GDS write helpers for source-backed lazy GDS views.
|
||||
|
||||
This module contains the reusable pieces that sit between lazy source readers
|
||||
and ordinary mutable library usage:
|
||||
|
||||
- `PortsLibraryView` layers a processed, ports-importing cache on top of a raw
|
||||
source view without mutating the source itself
|
||||
- `OverlayLibrary` exposes a mutable library surface that can mix source-backed
|
||||
cells with overlay-owned materialized patterns
|
||||
- the write helpers preserve source-backed copy-through behavior where
|
||||
possible, falling back to normal pattern serialization when a cell has been
|
||||
materialized or remapped
|
||||
|
||||
Both the classic and Arrow-backed lazy GDS readers rely on these helpers.
|
||||
The generic mutable overlay and ports-importing view live in `masque.library`.
|
||||
This module preserves source-backed GDS copy-through behavior where possible,
|
||||
falling back to normal pattern serialization when a cell has been materialized
|
||||
or remapped.
|
||||
"""
|
||||
from __future__ import annotations
|
||||
|
||||
from dataclasses import dataclass
|
||||
from typing import IO, Any, Literal, cast
|
||||
from collections import defaultdict
|
||||
from collections.abc import Callable, Iterator, Mapping, Sequence
|
||||
import copy
|
||||
from typing import IO, TYPE_CHECKING, Any, cast
|
||||
import gzip
|
||||
import logging
|
||||
import pathlib
|
||||
|
||||
import klamath
|
||||
import numpy
|
||||
from numpy.typing import NDArray
|
||||
|
||||
from . import gdsii
|
||||
from .utils import tmpfile
|
||||
from ..error import LibraryError
|
||||
from ..library import ILibrary, ILibraryView, LibraryView, _plan_source_names, _source_rename_map, dangling_mode_t
|
||||
from ..pattern import Pattern, map_targets
|
||||
from ..utils import apply_transforms
|
||||
from ..utils.ports2data import data_to_ports
|
||||
from ..library import ILibraryView, OverlayLibrary, _SourceEntry, _materialize_detached_pattern
|
||||
|
||||
if TYPE_CHECKING:
|
||||
from collections.abc import Iterator, Mapping
|
||||
|
||||
from ..pattern import Pattern
|
||||
|
||||
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
|
||||
@dataclass
|
||||
class _SourceLayer:
|
||||
""" One imported source layer tracked by an `OverlayLibrary`. """
|
||||
library: ILibraryView
|
||||
source_to_visible: dict[str, str]
|
||||
visible_to_source: dict[str, str]
|
||||
child_graph: dict[str, set[str]]
|
||||
order: list[str]
|
||||
|
||||
|
||||
@dataclass(frozen=True)
|
||||
class _SourceEntry:
|
||||
""" Reference to a single visible source-backed cell in an overlay. """
|
||||
layer_index: int
|
||||
source_name: str
|
||||
|
||||
|
||||
def _pattern_children(pat: Pattern) -> set[str]:
|
||||
return {child for child, refs in pat.refs.items() if child is not None and refs}
|
||||
|
||||
|
||||
def _remap_pattern_targets(pat: Pattern, remap: Callable[[str | None], str | None]) -> Pattern:
|
||||
if not pat.refs:
|
||||
return pat
|
||||
pat.refs = map_targets(pat.refs, remap)
|
||||
return pat
|
||||
|
||||
|
||||
def _coerce_library_view(source: Mapping[str, Pattern] | ILibraryView) -> ILibraryView:
|
||||
if isinstance(source, ILibraryView):
|
||||
return source
|
||||
return LibraryView(source)
|
||||
|
||||
|
||||
def _materialize_detached_pattern(view: ILibraryView, name: str) -> Pattern:
|
||||
func = getattr(view, '_materialize_pattern', None)
|
||||
if callable(func):
|
||||
return cast('Pattern', func(name, persist=False))
|
||||
return view[name].deepcopy()
|
||||
|
||||
|
||||
class PortsLibraryView(ILibraryView):
|
||||
"""
|
||||
Read-only view which imports ports into cells on first materialization.
|
||||
|
||||
The wrapped source remains untouched; this view owns a separate processed
|
||||
cache so direct-copy workflows can continue to use the raw source view.
|
||||
|
||||
Graph queries, source ordering, and copy-through capabilities are delegated
|
||||
to the wrapped source whenever possible, while `__getitem__` and
|
||||
`materialize_many()` return port-imported patterns.
|
||||
"""
|
||||
|
||||
def __init__(
|
||||
self,
|
||||
source: ILibraryView,
|
||||
*,
|
||||
layers: Sequence[gdsii.layer_t],
|
||||
max_depth: int = 0,
|
||||
skip_subcells: bool = True,
|
||||
) -> None:
|
||||
self._source = source
|
||||
self._layers = tuple(layers)
|
||||
self._max_depth = max_depth
|
||||
self._skip_subcells = skip_subcells
|
||||
self._cache: dict[str, Pattern] = {}
|
||||
self._lookups_in_progress: list[str] = []
|
||||
if hasattr(source, 'library_info'):
|
||||
self.library_info = cast('dict[str, Any]', getattr(source, 'library_info'))
|
||||
|
||||
def __getitem__(self, key: str) -> Pattern:
|
||||
return self._materialize_pattern(key, persist=True)
|
||||
|
||||
def __iter__(self) -> Iterator[str]:
|
||||
return iter(self._source)
|
||||
|
||||
def __len__(self) -> int:
|
||||
return len(self._source)
|
||||
|
||||
def __contains__(self, key: object) -> bool:
|
||||
return key in self._source
|
||||
|
||||
def _materialize_pattern(self, name: str, *, persist: bool) -> Pattern:
|
||||
if name in self._cache:
|
||||
return self._cache[name]
|
||||
|
||||
if name in self._lookups_in_progress:
|
||||
chain = ' -> '.join(self._lookups_in_progress + [name])
|
||||
raise LibraryError(
|
||||
f'Detected circular reference or recursive lookup of "{name}".\n'
|
||||
f'Lookup chain: {chain}\n'
|
||||
'This may be caused by an invalid (cyclical) reference, or buggy code.'
|
||||
)
|
||||
|
||||
self._lookups_in_progress.append(name)
|
||||
try:
|
||||
pat = _materialize_detached_pattern(self._source, name)
|
||||
pat = data_to_ports(
|
||||
layers=self._layers,
|
||||
library=self,
|
||||
pattern=pat,
|
||||
name=name,
|
||||
max_depth=self._max_depth,
|
||||
skip_subcells=self._skip_subcells,
|
||||
)
|
||||
finally:
|
||||
self._lookups_in_progress.pop()
|
||||
|
||||
if persist:
|
||||
self._cache[name] = pat
|
||||
return pat
|
||||
|
||||
def materialize_many(
|
||||
self,
|
||||
names: Sequence[str],
|
||||
*,
|
||||
persist: bool = True,
|
||||
) -> LibraryView:
|
||||
mats = {
|
||||
name: self._materialize_pattern(name, persist=persist)
|
||||
for name in dict.fromkeys(names)
|
||||
}
|
||||
return LibraryView(mats)
|
||||
|
||||
def source_order(self) -> tuple[str, ...]:
|
||||
return self._source.source_order()
|
||||
|
||||
def child_graph(
|
||||
self,
|
||||
dangling: dangling_mode_t = 'error',
|
||||
) -> dict[str, set[str]]:
|
||||
return self._source.child_graph(dangling=dangling)
|
||||
|
||||
def parent_graph(
|
||||
self,
|
||||
dangling: dangling_mode_t = 'error',
|
||||
) -> dict[str, set[str]]:
|
||||
return self._source.parent_graph(dangling=dangling)
|
||||
|
||||
def subtree(
|
||||
self,
|
||||
tops: str | Sequence[str],
|
||||
) -> ILibraryView:
|
||||
if isinstance(tops, str):
|
||||
tops = (tops,)
|
||||
keep = cast('set[str]', self._source.referenced_patterns(tops) - {None})
|
||||
keep |= set(tops)
|
||||
return self.materialize_many(tuple(keep), persist=True)
|
||||
|
||||
def tops(self) -> list[str]:
|
||||
return self._source.tops()
|
||||
|
||||
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]]]:
|
||||
finder = getattr(self._source, 'find_refs_local', None)
|
||||
if callable(finder):
|
||||
return cast('dict[str, list[NDArray[numpy.float64]]]', finder(name, parent_graph=parent_graph, dangling=dangling))
|
||||
return super().find_refs_local(name, parent_graph=parent_graph, dangling=dangling)
|
||||
|
||||
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]]:
|
||||
finder = getattr(self._source, 'find_refs_global', None)
|
||||
if callable(finder):
|
||||
return cast(
|
||||
'dict[tuple[str, ...], NDArray[numpy.float64]]',
|
||||
finder(name, order=order, parent_graph=parent_graph, dangling=dangling),
|
||||
)
|
||||
return super().find_refs_global(name, order=order, parent_graph=parent_graph, dangling=dangling)
|
||||
|
||||
def raw_struct_bytes(self, name: str) -> bytes:
|
||||
reader = getattr(self._source, 'raw_struct_bytes', None)
|
||||
if not callable(reader):
|
||||
raise AttributeError('raw_struct_bytes')
|
||||
return cast('bytes', reader(name))
|
||||
|
||||
def can_copy_raw_struct(self, name: str) -> bool:
|
||||
can_copy = getattr(self._source, 'can_copy_raw_struct', None)
|
||||
if not callable(can_copy):
|
||||
return False
|
||||
return bool(can_copy(name))
|
||||
|
||||
def close(self) -> None:
|
||||
closer = getattr(self._source, 'close', None)
|
||||
if callable(closer):
|
||||
closer()
|
||||
|
||||
def __enter__(self) -> PortsLibraryView:
|
||||
return self
|
||||
|
||||
def __exit__(self, *_args: object) -> None:
|
||||
self.close()
|
||||
|
||||
|
||||
class OverlayLibrary(ILibrary):
|
||||
"""
|
||||
Mutable overlay over one or more source libraries.
|
||||
|
||||
Source-backed cells remain lazy until accessed through `__getitem__`, at
|
||||
which point that visible cell is promoted into an overlay-owned materialized
|
||||
`Pattern`.
|
||||
|
||||
This is the main mutable integration surface for lazy GDS content. It lets
|
||||
callers:
|
||||
- expose one or more source-backed libraries behind a normal `ILibrary`
|
||||
interface
|
||||
- add or replace cells with overlay-owned patterns
|
||||
- rename visible source cells
|
||||
- remap references without immediately rewriting untouched source structs
|
||||
"""
|
||||
|
||||
def __init__(self) -> None:
|
||||
self._layers: list[_SourceLayer] = []
|
||||
self._entries: dict[str, Pattern | _SourceEntry] = {}
|
||||
self._order: list[str] = []
|
||||
self._target_remap: dict[str, str] = {}
|
||||
|
||||
def __iter__(self) -> Iterator[str]:
|
||||
return (name for name in self._order if name in self._entries)
|
||||
|
||||
def __len__(self) -> int:
|
||||
return len(self._entries)
|
||||
|
||||
def __contains__(self, key: object) -> bool:
|
||||
return key in self._entries
|
||||
|
||||
def __getitem__(self, key: str) -> Pattern:
|
||||
return self._materialize_pattern(key, persist=True)
|
||||
|
||||
def __setitem__(
|
||||
self,
|
||||
key: str,
|
||||
value: Pattern | Callable[[], Pattern],
|
||||
) -> None:
|
||||
if key in self._entries:
|
||||
raise LibraryError(f'"{key}" already exists in the library. Overwriting is not allowed!')
|
||||
pattern = value() if callable(value) else value
|
||||
self._entries[key] = pattern
|
||||
if key not in self._order:
|
||||
self._order.append(key)
|
||||
|
||||
def __delitem__(self, key: str) -> None:
|
||||
if key not in self._entries:
|
||||
raise KeyError(key)
|
||||
del self._entries[key]
|
||||
|
||||
def _merge(self, key_self: str, other: Mapping[str, Pattern], key_other: str) -> None:
|
||||
self[key_self] = copy.deepcopy(other[key_other])
|
||||
|
||||
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]:
|
||||
"""
|
||||
Add a source-backed library layer.
|
||||
|
||||
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`.
|
||||
"""
|
||||
view = _coerce_library_view(source)
|
||||
source_order = list(view.source_order())
|
||||
child_graph = view.child_graph(dangling='include')
|
||||
|
||||
source_to_visible = _plan_source_names(
|
||||
self,
|
||||
source_order,
|
||||
self._entries,
|
||||
rename_theirs = rename_theirs,
|
||||
rename_when = rename_when,
|
||||
)
|
||||
visible_to_source = {visible: source_name for source_name, visible in source_to_visible.items()}
|
||||
|
||||
layer = _SourceLayer(
|
||||
library=view,
|
||||
source_to_visible=source_to_visible,
|
||||
visible_to_source=visible_to_source,
|
||||
child_graph=child_graph,
|
||||
order=[source_to_visible[name] for name in source_order],
|
||||
)
|
||||
layer_index = len(self._layers)
|
||||
self._layers.append(layer)
|
||||
|
||||
for source_name, visible_name in source_to_visible.items():
|
||||
self._entries[visible_name] = _SourceEntry(layer_index=layer_index, source_name=source_name)
|
||||
if visible_name not in self._order:
|
||||
self._order.append(visible_name)
|
||||
|
||||
return _source_rename_map(source_to_visible)
|
||||
|
||||
def rename(
|
||||
self,
|
||||
old_name: str,
|
||||
new_name: str,
|
||||
move_references: bool = False,
|
||||
) -> OverlayLibrary:
|
||||
if old_name not in self._entries:
|
||||
raise LibraryError(f'"{old_name}" does not exist in the library.')
|
||||
if old_name == new_name:
|
||||
return self
|
||||
if new_name in self._entries:
|
||||
raise LibraryError(f'"{new_name}" already exists in the library.')
|
||||
|
||||
entry = self._entries.pop(old_name)
|
||||
self._entries[new_name] = entry
|
||||
if isinstance(entry, _SourceEntry):
|
||||
layer = self._layers[entry.layer_index]
|
||||
layer.source_to_visible[entry.source_name] = new_name
|
||||
del layer.visible_to_source[old_name]
|
||||
layer.visible_to_source[new_name] = entry.source_name
|
||||
|
||||
idx = self._order.index(old_name)
|
||||
self._order[idx] = new_name
|
||||
|
||||
if move_references:
|
||||
self.move_references(old_name, new_name)
|
||||
return self
|
||||
|
||||
def _resolve_target(self, target: str) -> str:
|
||||
seen: set[str] = set()
|
||||
current = target
|
||||
while current in self._target_remap:
|
||||
if current in seen:
|
||||
raise LibraryError(f'Cycle encountered while resolving target remap for {target!r}')
|
||||
seen.add(current)
|
||||
current = self._target_remap[current]
|
||||
return current
|
||||
|
||||
def _set_target_remap(self, old_target: str, new_target: str) -> None:
|
||||
resolved_new = self._resolve_target(new_target)
|
||||
if resolved_new == old_target:
|
||||
raise LibraryError(f'Ref target remap would create a cycle: {old_target!r} -> {new_target!r}')
|
||||
self._target_remap[old_target] = resolved_new
|
||||
for key in list(self._target_remap):
|
||||
self._target_remap[key] = self._resolve_target(self._target_remap[key])
|
||||
|
||||
def move_references(self, old_target: str, new_target: str) -> OverlayLibrary:
|
||||
if old_target == new_target:
|
||||
return self
|
||||
self._set_target_remap(old_target, new_target)
|
||||
for entry in list(self._entries.values()):
|
||||
if isinstance(entry, Pattern) and old_target in entry.refs:
|
||||
entry.refs[new_target].extend(entry.refs[old_target])
|
||||
del entry.refs[old_target]
|
||||
return self
|
||||
|
||||
def _effective_target(self, layer: _SourceLayer, target: str) -> str:
|
||||
visible = layer.source_to_visible.get(target, target)
|
||||
return self._resolve_target(visible)
|
||||
|
||||
def _materialize_pattern(self, name: str, *, persist: bool) -> Pattern:
|
||||
if name not in self._entries:
|
||||
raise KeyError(name)
|
||||
entry = self._entries[name]
|
||||
if isinstance(entry, Pattern):
|
||||
return entry
|
||||
|
||||
layer = self._layers[entry.layer_index]
|
||||
source_pat = layer.library[entry.source_name].deepcopy()
|
||||
remap = lambda target: None if target is None else self._effective_target(layer, target)
|
||||
pat = _remap_pattern_targets(source_pat, remap)
|
||||
if persist:
|
||||
self._entries[name] = pat
|
||||
return pat
|
||||
|
||||
def child_graph(
|
||||
self,
|
||||
dangling: dangling_mode_t = 'error',
|
||||
) -> dict[str, set[str]]:
|
||||
graph: dict[str, set[str]] = {}
|
||||
for name in self._order:
|
||||
if name not in self._entries:
|
||||
continue
|
||||
entry = self._entries[name]
|
||||
if isinstance(entry, Pattern):
|
||||
graph[name] = _pattern_children(entry)
|
||||
continue
|
||||
layer = self._layers[entry.layer_index]
|
||||
children = {self._effective_target(layer, child) for child in layer.child_graph.get(entry.source_name, set())}
|
||||
graph[name] = children
|
||||
|
||||
existing = set(graph)
|
||||
dangling_refs = set().union(*(children - existing for children in graph.values()))
|
||||
if dangling == 'error':
|
||||
if dangling_refs:
|
||||
raise self._dangling_refs_error(cast('set[str]', dangling_refs), 'building child graph')
|
||||
return graph
|
||||
if dangling == 'ignore':
|
||||
return {name: {child for child in children if child in existing} for name, children in graph.items()}
|
||||
|
||||
for child in dangling_refs:
|
||||
graph.setdefault(cast('str', child), set())
|
||||
return graph
|
||||
|
||||
def parent_graph(
|
||||
self,
|
||||
dangling: dangling_mode_t = 'error',
|
||||
) -> dict[str, set[str]]:
|
||||
child_graph = self.child_graph(dangling='include' if dangling == 'include' else 'ignore')
|
||||
existing = set(self.keys())
|
||||
igraph: dict[str, set[str]] = {name: set() for name in child_graph}
|
||||
for parent, children in child_graph.items():
|
||||
for child in children:
|
||||
if child in existing or dangling == 'include':
|
||||
igraph.setdefault(child, set()).add(parent)
|
||||
if dangling == 'error':
|
||||
raw = self.child_graph(dangling='include')
|
||||
dangling_refs = set().union(*(children - existing for children in raw.values()))
|
||||
if dangling_refs:
|
||||
raise self._dangling_refs_error(cast('set[str]', dangling_refs), 'building parent graph')
|
||||
return igraph
|
||||
|
||||
def subtree(
|
||||
self,
|
||||
tops: str | Sequence[str],
|
||||
) -> ILibraryView:
|
||||
if isinstance(tops, str):
|
||||
tops = (tops,)
|
||||
keep = cast('set[str]', self.referenced_patterns(tops) - {None})
|
||||
keep |= set(tops)
|
||||
return LibraryView({name: self[name] for name in keep})
|
||||
|
||||
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]]]:
|
||||
instances: dict[str, list[NDArray[numpy.float64]]] = 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()):
|
||||
pat = self._materialize_pattern(parent, persist=False)
|
||||
for ref in pat.refs.get(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]]:
|
||||
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():
|
||||
outer_tf = numpy.concatenate(outer)
|
||||
for path, inner in inners:
|
||||
combined = apply_transforms(outer_tf, inner)
|
||||
transforms[parent].append(((next_name,) + path, combined))
|
||||
|
||||
result = {}
|
||||
for parent, targets in transforms.items():
|
||||
for path, instances in targets:
|
||||
result[(parent,) + path] = instances
|
||||
return result
|
||||
|
||||
def source_order(self) -> tuple[str, ...]:
|
||||
return tuple(name for name in self._order if name in self._entries)
|
||||
|
||||
|
||||
def _iter_library_infos(library: Mapping[str, Pattern] | ILibraryView) -> Iterator[dict[str, Any]]:
|
||||
info = getattr(library, 'library_info', None)
|
||||
if isinstance(info, dict):
|
||||
|
|
@ -598,7 +74,7 @@ def _can_copy_raw_cell(library: Mapping[str, Pattern] | ILibraryView, name: str)
|
|||
def _raw_struct_bytes(library: Mapping[str, Pattern] | ILibraryView, name: str) -> bytes:
|
||||
reader = getattr(library, 'raw_struct_bytes', None)
|
||||
if not callable(reader):
|
||||
raise AttributeError('raw_struct_bytes')
|
||||
raise TypeError('raw_struct_bytes')
|
||||
return cast('bytes', reader(name))
|
||||
|
||||
|
||||
|
|
|
|||
|
|
@ -14,7 +14,7 @@ Classes include:
|
|||
- `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 typing import Self, Any, cast, TypeAlias, Protocol, Literal
|
||||
from collections.abc import Container, Iterator, Mapping, MutableMapping, Sequence, Callable
|
||||
import logging
|
||||
import re
|
||||
|
|
@ -35,11 +35,7 @@ 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
|
||||
|
||||
from .pattern import Pattern, map_layers, map_targets
|
||||
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
|
|
@ -903,7 +899,7 @@ class ILibrary(ILibraryView, MutableMapping[str, 'Pattern'], metaclass=ABCMeta):
|
|||
An `Abstract` or `Pattern` object.
|
||||
"""
|
||||
from .pattern import Pattern #noqa: PLC0415
|
||||
if not isinstance(other, (str, Abstract, Pattern)):
|
||||
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
|
||||
|
||||
|
|
@ -1491,6 +1487,500 @@ class Library(ILibrary):
|
|||
return tree, pat
|
||||
|
||||
|
||||
@dataclass
|
||||
class _SourceLayer:
|
||||
""" One imported source layer tracked by an `OverlayLibrary`. """
|
||||
library: ILibraryView
|
||||
source_to_visible: dict[str, str]
|
||||
visible_to_source: dict[str, str]
|
||||
child_graph: dict[str, set[str]]
|
||||
order: list[str]
|
||||
|
||||
|
||||
@dataclass(frozen=True)
|
||||
class _SourceEntry:
|
||||
""" Reference to a single visible source-backed cell in an overlay. """
|
||||
layer_index: int
|
||||
source_name: str
|
||||
|
||||
|
||||
def _materialize_detached_pattern(view: ILibraryView, name: str) -> Pattern:
|
||||
func = getattr(view, '_materialize_pattern', None)
|
||||
if callable(func):
|
||||
return cast('Pattern', func(name, persist=False))
|
||||
return view[name].deepcopy()
|
||||
|
||||
|
||||
class PortsLibraryView(ILibraryView):
|
||||
"""
|
||||
Read-only view which imports ports into cells on first materialization.
|
||||
|
||||
The wrapped source remains untouched; this view owns a separate processed
|
||||
cache so direct-copy workflows can continue to use the raw source view.
|
||||
|
||||
Graph queries, source ordering, and copy-through capabilities are delegated
|
||||
to the wrapped source whenever possible, while `__getitem__` and
|
||||
`materialize_many()` return port-imported patterns.
|
||||
"""
|
||||
|
||||
def __init__(
|
||||
self,
|
||||
source: ILibraryView,
|
||||
*,
|
||||
layers: Sequence[layer_t],
|
||||
max_depth: int = 0,
|
||||
skip_subcells: bool = True,
|
||||
) -> None:
|
||||
self._source = source
|
||||
self._layers = tuple(layers)
|
||||
self._max_depth = max_depth
|
||||
self._skip_subcells = skip_subcells
|
||||
self._cache: dict[str, Pattern] = {}
|
||||
self._lookups_in_progress: list[str] = []
|
||||
if hasattr(source, 'library_info'):
|
||||
self.library_info = cast('dict[str, Any]', source.library_info)
|
||||
|
||||
def __getitem__(self, key: str) -> Pattern:
|
||||
return self._materialize_pattern(key, persist=True)
|
||||
|
||||
def __iter__(self) -> Iterator[str]:
|
||||
return iter(self._source)
|
||||
|
||||
def __len__(self) -> int:
|
||||
return len(self._source)
|
||||
|
||||
def __contains__(self, key: object) -> bool:
|
||||
return key in self._source
|
||||
|
||||
def _materialize_pattern(self, name: str, *, persist: bool) -> Pattern:
|
||||
from .utils.ports2data import data_to_ports # noqa: PLC0415
|
||||
|
||||
if name in self._cache:
|
||||
return self._cache[name]
|
||||
|
||||
if name in self._lookups_in_progress:
|
||||
chain = ' -> '.join(self._lookups_in_progress + [name])
|
||||
raise LibraryError(
|
||||
f'Detected circular reference or recursive lookup of "{name}".\n'
|
||||
f'Lookup chain: {chain}\n'
|
||||
'This may be caused by an invalid (cyclical) reference, or buggy code.'
|
||||
)
|
||||
|
||||
self._lookups_in_progress.append(name)
|
||||
try:
|
||||
pat = _materialize_detached_pattern(self._source, name)
|
||||
pat = data_to_ports(
|
||||
layers=self._layers,
|
||||
library=self,
|
||||
pattern=pat,
|
||||
name=name,
|
||||
max_depth=self._max_depth,
|
||||
skip_subcells=self._skip_subcells,
|
||||
)
|
||||
finally:
|
||||
self._lookups_in_progress.pop()
|
||||
|
||||
if persist:
|
||||
self._cache[name] = pat
|
||||
return pat
|
||||
|
||||
def materialize_many(
|
||||
self,
|
||||
names: Sequence[str],
|
||||
*,
|
||||
persist: bool = True,
|
||||
) -> LibraryView:
|
||||
mats = {
|
||||
name: self._materialize_pattern(name, persist=persist)
|
||||
for name in dict.fromkeys(names)
|
||||
}
|
||||
return LibraryView(mats)
|
||||
|
||||
def source_order(self) -> tuple[str, ...]:
|
||||
return self._source.source_order()
|
||||
|
||||
def child_graph(
|
||||
self,
|
||||
dangling: dangling_mode_t = 'error',
|
||||
) -> dict[str, set[str]]:
|
||||
return self._source.child_graph(dangling=dangling)
|
||||
|
||||
def parent_graph(
|
||||
self,
|
||||
dangling: dangling_mode_t = 'error',
|
||||
) -> dict[str, set[str]]:
|
||||
return self._source.parent_graph(dangling=dangling)
|
||||
|
||||
def subtree(
|
||||
self,
|
||||
tops: str | Sequence[str],
|
||||
) -> ILibraryView:
|
||||
if isinstance(tops, str):
|
||||
tops = (tops,)
|
||||
keep = cast('set[str]', self._source.referenced_patterns(tops) - {None})
|
||||
keep |= set(tops)
|
||||
return self.materialize_many(tuple(keep), persist=True)
|
||||
|
||||
def tops(self) -> list[str]:
|
||||
return self._source.tops()
|
||||
|
||||
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]]]:
|
||||
finder = getattr(self._source, 'find_refs_local', None)
|
||||
if callable(finder):
|
||||
return cast('dict[str, list[NDArray[numpy.float64]]]', finder(name, parent_graph=parent_graph, dangling=dangling))
|
||||
return super().find_refs_local(name, parent_graph=parent_graph, dangling=dangling)
|
||||
|
||||
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]]:
|
||||
finder = getattr(self._source, 'find_refs_global', None)
|
||||
if callable(finder):
|
||||
return cast(
|
||||
'dict[tuple[str, ...], NDArray[numpy.float64]]',
|
||||
finder(name, order=order, parent_graph=parent_graph, dangling=dangling),
|
||||
)
|
||||
return super().find_refs_global(name, order=order, parent_graph=parent_graph, dangling=dangling)
|
||||
|
||||
def raw_struct_bytes(self, name: str) -> bytes:
|
||||
reader = getattr(self._source, 'raw_struct_bytes', None)
|
||||
if not callable(reader):
|
||||
raise TypeError('raw_struct_bytes')
|
||||
return cast('bytes', reader(name))
|
||||
|
||||
def can_copy_raw_struct(self, name: str) -> bool:
|
||||
can_copy = getattr(self._source, 'can_copy_raw_struct', None)
|
||||
if not callable(can_copy):
|
||||
return False
|
||||
return bool(can_copy(name))
|
||||
|
||||
def close(self) -> None:
|
||||
closer = getattr(self._source, 'close', None)
|
||||
if callable(closer):
|
||||
closer()
|
||||
|
||||
def __enter__(self) -> 'PortsLibraryView':
|
||||
return self
|
||||
|
||||
def __exit__(self, *_args: object) -> None:
|
||||
self.close()
|
||||
|
||||
|
||||
class OverlayLibrary(ILibrary):
|
||||
"""
|
||||
Mutable overlay over one or more source libraries.
|
||||
|
||||
Source-backed cells remain lazy until accessed through `__getitem__`, at
|
||||
which point that visible cell is promoted into an overlay-owned materialized
|
||||
`Pattern`.
|
||||
"""
|
||||
|
||||
def __init__(self) -> None:
|
||||
self._layers: list[_SourceLayer] = []
|
||||
self._entries: dict[str, Pattern | _SourceEntry] = {}
|
||||
self._order: list[str] = []
|
||||
self._target_remap: dict[str, str] = {}
|
||||
|
||||
def __iter__(self) -> Iterator[str]:
|
||||
return (name for name in self._order if name in self._entries)
|
||||
|
||||
def __len__(self) -> int:
|
||||
return len(self._entries)
|
||||
|
||||
def __contains__(self, key: object) -> bool:
|
||||
return key in self._entries
|
||||
|
||||
def __getitem__(self, key: str) -> Pattern:
|
||||
return self._materialize_pattern(key, persist=True)
|
||||
|
||||
def __setitem__(
|
||||
self,
|
||||
key: str,
|
||||
value: Pattern | Callable[[], Pattern],
|
||||
) -> None:
|
||||
if key in self._entries:
|
||||
raise LibraryError(f'"{key}" already exists in the library. Overwriting is not allowed!')
|
||||
pattern = value() if callable(value) else value
|
||||
self._entries[key] = pattern
|
||||
if key not in self._order:
|
||||
self._order.append(key)
|
||||
|
||||
def __delitem__(self, key: str) -> None:
|
||||
if key not in self._entries:
|
||||
raise KeyError(key)
|
||||
del self._entries[key]
|
||||
|
||||
def _merge(self, key_self: str, other: Mapping[str, Pattern], key_other: str) -> None:
|
||||
self[key_self] = copy.deepcopy(other[key_other])
|
||||
|
||||
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]:
|
||||
"""
|
||||
Add a source-backed library layer.
|
||||
|
||||
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`.
|
||||
"""
|
||||
view = source if isinstance(source, ILibraryView) else LibraryView(source)
|
||||
source_order = list(view.source_order())
|
||||
child_graph = view.child_graph(dangling='include')
|
||||
|
||||
source_to_visible = _plan_source_names(
|
||||
self,
|
||||
source_order,
|
||||
self._entries,
|
||||
rename_theirs = rename_theirs,
|
||||
rename_when = rename_when,
|
||||
)
|
||||
visible_to_source = {visible: source_name for source_name, visible in source_to_visible.items()}
|
||||
|
||||
layer = _SourceLayer(
|
||||
library=view,
|
||||
source_to_visible=source_to_visible,
|
||||
visible_to_source=visible_to_source,
|
||||
child_graph=child_graph,
|
||||
order=[source_to_visible[name] for name in source_order],
|
||||
)
|
||||
layer_index = len(self._layers)
|
||||
self._layers.append(layer)
|
||||
|
||||
for source_name, visible_name in source_to_visible.items():
|
||||
self._entries[visible_name] = _SourceEntry(layer_index=layer_index, source_name=source_name)
|
||||
if visible_name not in self._order:
|
||||
self._order.append(visible_name)
|
||||
|
||||
return _source_rename_map(source_to_visible)
|
||||
|
||||
def rename(
|
||||
self,
|
||||
old_name: str,
|
||||
new_name: str,
|
||||
move_references: bool = False,
|
||||
) -> 'OverlayLibrary':
|
||||
if old_name not in self._entries:
|
||||
raise LibraryError(f'"{old_name}" does not exist in the library.')
|
||||
if old_name == new_name:
|
||||
return self
|
||||
if new_name in self._entries:
|
||||
raise LibraryError(f'"{new_name}" already exists in the library.')
|
||||
|
||||
entry = self._entries.pop(old_name)
|
||||
self._entries[new_name] = entry
|
||||
if isinstance(entry, _SourceEntry):
|
||||
layer = self._layers[entry.layer_index]
|
||||
layer.source_to_visible[entry.source_name] = new_name
|
||||
del layer.visible_to_source[old_name]
|
||||
layer.visible_to_source[new_name] = entry.source_name
|
||||
|
||||
idx = self._order.index(old_name)
|
||||
self._order[idx] = new_name
|
||||
|
||||
if move_references:
|
||||
self.move_references(old_name, new_name)
|
||||
return self
|
||||
|
||||
def _resolve_target(self, target: str) -> str:
|
||||
seen: set[str] = set()
|
||||
current = target
|
||||
while current in self._target_remap:
|
||||
if current in seen:
|
||||
raise LibraryError(f'Cycle encountered while resolving target remap for {target!r}')
|
||||
seen.add(current)
|
||||
current = self._target_remap[current]
|
||||
return current
|
||||
|
||||
def _set_target_remap(self, old_target: str, new_target: str) -> None:
|
||||
resolved_new = self._resolve_target(new_target)
|
||||
if resolved_new == old_target:
|
||||
raise LibraryError(f'Ref target remap would create a cycle: {old_target!r} -> {new_target!r}')
|
||||
self._target_remap[old_target] = resolved_new
|
||||
for key in list(self._target_remap):
|
||||
self._target_remap[key] = self._resolve_target(self._target_remap[key])
|
||||
|
||||
def move_references(self, old_target: str, new_target: str) -> 'OverlayLibrary':
|
||||
if old_target == new_target:
|
||||
return self
|
||||
self._set_target_remap(old_target, new_target)
|
||||
for entry in list(self._entries.values()):
|
||||
if isinstance(entry, Pattern) and old_target in entry.refs:
|
||||
entry.refs[new_target].extend(entry.refs[old_target])
|
||||
del entry.refs[old_target]
|
||||
return self
|
||||
|
||||
def _effective_target(self, layer: _SourceLayer, target: str) -> str:
|
||||
visible = layer.source_to_visible.get(target, target)
|
||||
return self._resolve_target(visible)
|
||||
|
||||
def _materialize_pattern(self, name: str, *, persist: bool) -> Pattern:
|
||||
if name not in self._entries:
|
||||
raise KeyError(name)
|
||||
entry = self._entries[name]
|
||||
if isinstance(entry, Pattern):
|
||||
return entry
|
||||
|
||||
layer = self._layers[entry.layer_index]
|
||||
source_pat = layer.library[entry.source_name].deepcopy()
|
||||
|
||||
def remap(target: str | None) -> str | None:
|
||||
return None if target is None else self._effective_target(layer, target)
|
||||
|
||||
if source_pat.refs:
|
||||
source_pat.refs = map_targets(source_pat.refs, remap)
|
||||
pat = source_pat
|
||||
if persist:
|
||||
self._entries[name] = pat
|
||||
return pat
|
||||
|
||||
def child_graph(
|
||||
self,
|
||||
dangling: dangling_mode_t = 'error',
|
||||
) -> dict[str, set[str]]:
|
||||
graph: dict[str, set[str]] = {}
|
||||
for name in self._order:
|
||||
if name not in self._entries:
|
||||
continue
|
||||
entry = self._entries[name]
|
||||
if isinstance(entry, Pattern):
|
||||
graph[name] = {child for child, refs in entry.refs.items() if child is not None and refs}
|
||||
continue
|
||||
layer = self._layers[entry.layer_index]
|
||||
children = {self._effective_target(layer, child) for child in layer.child_graph.get(entry.source_name, set())}
|
||||
graph[name] = children
|
||||
|
||||
existing = set(graph)
|
||||
dangling_refs = set().union(*(children - existing for children in graph.values()))
|
||||
if dangling == 'error':
|
||||
if dangling_refs:
|
||||
raise self._dangling_refs_error(cast('set[str]', dangling_refs), 'building child graph')
|
||||
return graph
|
||||
if dangling == 'ignore':
|
||||
return {name: {child for child in children if child in existing} for name, children in graph.items()}
|
||||
|
||||
for child in dangling_refs:
|
||||
graph.setdefault(cast('str', child), set())
|
||||
return graph
|
||||
|
||||
def parent_graph(
|
||||
self,
|
||||
dangling: dangling_mode_t = 'error',
|
||||
) -> dict[str, set[str]]:
|
||||
child_graph = self.child_graph(dangling='include' if dangling == 'include' else 'ignore')
|
||||
existing = set(self.keys())
|
||||
igraph: dict[str, set[str]] = {name: set() for name in child_graph}
|
||||
for parent, children in child_graph.items():
|
||||
for child in children:
|
||||
if child in existing or dangling == 'include':
|
||||
igraph.setdefault(child, set()).add(parent)
|
||||
if dangling == 'error':
|
||||
raw = self.child_graph(dangling='include')
|
||||
dangling_refs = set().union(*(children - existing for children in raw.values()))
|
||||
if dangling_refs:
|
||||
raise self._dangling_refs_error(cast('set[str]', dangling_refs), 'building parent graph')
|
||||
return igraph
|
||||
|
||||
def subtree(
|
||||
self,
|
||||
tops: str | Sequence[str],
|
||||
) -> ILibraryView:
|
||||
if isinstance(tops, str):
|
||||
tops = (tops,)
|
||||
keep = cast('set[str]', self.referenced_patterns(tops) - {None})
|
||||
keep |= set(tops)
|
||||
return LibraryView({name: self[name] for name in keep})
|
||||
|
||||
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]]]:
|
||||
instances: dict[str, list[NDArray[numpy.float64]]] = 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()):
|
||||
pat = self._materialize_pattern(parent, persist=False)
|
||||
for ref in pat.refs.get(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]]:
|
||||
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():
|
||||
outer_tf = numpy.concatenate(outer)
|
||||
for path, inner in inners:
|
||||
combined = apply_transforms(outer_tf, inner)
|
||||
transforms[parent].append(((next_name,) + path, combined))
|
||||
|
||||
result = {}
|
||||
for parent, targets in transforms.items():
|
||||
for path, instances in targets:
|
||||
result[(parent,) + path] = instances
|
||||
return result
|
||||
|
||||
def source_order(self) -> tuple[str, ...]:
|
||||
return tuple(name for name in self._order if name in self._entries)
|
||||
|
||||
|
||||
@dataclass
|
||||
class _BuildRecipe:
|
||||
""" Captured deferred call to a pattern factory. """
|
||||
|
|
@ -1903,8 +2393,6 @@ class _BuildSessionLibrary(ILibrary):
|
|||
"""
|
||||
|
||||
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()
|
||||
|
|
@ -2107,13 +2595,13 @@ class _BuildSessionLibrary(ILibrary):
|
|||
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')
|
||||
raise BuildError(f'Recipe for "{name}" returned {type(pattern).__name__}, expected Pattern') # noqa: TRY301
|
||||
else:
|
||||
pattern = declaration.deepcopy()
|
||||
|
||||
if name in self._overlay:
|
||||
if self._overlay[name] is not pattern:
|
||||
raise BuildError(
|
||||
raise BuildError( # noqa: TRY301
|
||||
f'Recipe for "{name}" wrote a different pattern into the session under its own name.'
|
||||
)
|
||||
else:
|
||||
|
|
|
|||
|
|
@ -6,7 +6,7 @@ from numpy.testing import assert_allclose
|
|||
|
||||
from ..file import gdsii, gdsii_lazy
|
||||
from ..pattern import Pattern
|
||||
from ..library import Library
|
||||
from ..library import Library, OverlayLibrary
|
||||
|
||||
|
||||
def _make_lazy_port_library() -> Library:
|
||||
|
|
@ -73,7 +73,7 @@ def test_gdsii_lazy_overlay_add_source_stays_lazy_for_processed_view(tmp_path: P
|
|||
raw, _ = gdsii_lazy.readfile(gds_file)
|
||||
processed = raw.with_ports_from_data(layers=[(10, 0)], max_depth=2)
|
||||
|
||||
overlay = gdsii_lazy.OverlayLibrary()
|
||||
overlay = OverlayLibrary()
|
||||
overlay.add_source(processed)
|
||||
|
||||
assert not raw._cache
|
||||
|
|
@ -85,7 +85,7 @@ def test_gdsii_lazy_overlay_add_source_stays_lazy_for_processed_view(tmp_path: P
|
|||
|
||||
def test_gdsii_lazy_overlay_add_source_can_rename_every_source_cell() -> None:
|
||||
src = _make_lazy_port_library()
|
||||
overlay = gdsii_lazy.OverlayLibrary()
|
||||
overlay = OverlayLibrary()
|
||||
|
||||
rename_map = overlay.add_source(
|
||||
src,
|
||||
|
|
@ -106,10 +106,10 @@ def test_gdsii_lazy_overlay_add_source_rename_when_validation() -> None:
|
|||
src = _make_lazy_port_library()
|
||||
|
||||
with pytest.raises(TypeError, match='rename_theirs'):
|
||||
gdsii_lazy.OverlayLibrary().add_source(src, rename_when='always')
|
||||
OverlayLibrary().add_source(src, rename_when='always')
|
||||
|
||||
with pytest.raises(ValueError, match='rename mode'):
|
||||
gdsii_lazy.OverlayLibrary().add_source(src, rename_when='sometimes') # type: ignore[arg-type]
|
||||
OverlayLibrary().add_source(src, rename_when='sometimes') # type: ignore[arg-type]
|
||||
|
||||
|
||||
def test_gdsii_lazy_processed_write_roundtrips_without_explicit_units(tmp_path: Path) -> None:
|
||||
|
|
|
|||
|
|
@ -10,7 +10,7 @@ import pytest
|
|||
pytest.importorskip('pyarrow')
|
||||
|
||||
from .. import PatternError
|
||||
from ..library import Library
|
||||
from ..library import Library, OverlayLibrary
|
||||
from ..pattern import Pattern
|
||||
from ..repetition import Grid
|
||||
from ..file import gdsii, gdsii_lazy_arrow
|
||||
|
|
@ -155,7 +155,7 @@ def test_gdsii_lazy_arrow_local_and_global_refs(tmp_path: Path) -> None:
|
|||
assert sum(arr.shape[0] for arr in local['mid']) == 5
|
||||
|
||||
global_refs = lib.find_refs_global('leaf')
|
||||
assert {path for path in global_refs} == {('top', 'mid', 'leaf')}
|
||||
assert set(global_refs) == {('top', 'mid', 'leaf')}
|
||||
assert global_refs[('top', 'mid', 'leaf')].shape[0] == 5
|
||||
|
||||
|
||||
|
|
@ -265,7 +265,7 @@ def test_gdsii_lazy_overlay_merge_and_write(tmp_path: Path) -> None:
|
|||
lib_a, _ = gdsii_lazy_arrow.readfile(gds_a)
|
||||
lib_b, _ = gdsii_lazy_arrow.readfile(gds_b)
|
||||
|
||||
overlay = gdsii_lazy_arrow.OverlayLibrary()
|
||||
overlay = OverlayLibrary()
|
||||
overlay.add_source(lib_a)
|
||||
rename_map = overlay.add_source(lib_b, rename_theirs=lambda lib, name: lib.get_name(name))
|
||||
renamed_leaf = rename_map['leaf']
|
||||
|
|
@ -293,7 +293,7 @@ def test_gdsii_writer_accepts_overlay_library(tmp_path: Path) -> None:
|
|||
|
||||
lib, info = gdsii_lazy_arrow.readfile(gds_file)
|
||||
|
||||
overlay = gdsii_lazy_arrow.OverlayLibrary()
|
||||
overlay = OverlayLibrary()
|
||||
overlay.add_source(lib)
|
||||
overlay.rename('leaf', 'leaf_copy', move_references=True)
|
||||
|
||||
|
|
|
|||
Loading…
Add table
Add a link
Reference in a new issue