From 4c64352152e2a38b56a564882a0567d996b3b6b0 Mon Sep 17 00:00:00 2001 From: Jan Petykiewicz Date: Thu, 9 Jul 2026 13:05:01 -0700 Subject: [PATCH] [OverlayLibrary / PortsLibraryView] move into masque.library --- masque/__init__.py | 2 + masque/file/gdsii_lazy.py | 27 +- masque/file/gdsii_lazy_arrow.py | 25 +- masque/file/gdsii_lazy_core.py | 550 +-------------------------- masque/library.py | 510 ++++++++++++++++++++++++- masque/test/test_gdsii_lazy.py | 10 +- masque/test/test_gdsii_lazy_arrow.py | 8 +- 7 files changed, 558 insertions(+), 574 deletions(-) diff --git a/masque/__init__.py b/masque/__init__.py index f64b180..b5dddde 100644 --- a/masque/__init__.py +++ b/masque/__init__.py @@ -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, diff --git a/masque/file/gdsii_lazy.py b/masque/file/gdsii_lazy.py index b587c89..a0b6e6f 100644 --- a/masque/file/gdsii_lazy.py +++ b/masque/file/gdsii_lazy.py @@ -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) diff --git a/masque/file/gdsii_lazy_arrow.py b/masque/file/gdsii_lazy_arrow.py index f1dcc32..a53c1ff 100644 --- a/masque/file/gdsii_lazy_arrow.py +++ b/masque/file/gdsii_lazy_arrow.py @@ -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) diff --git a/masque/file/gdsii_lazy_core.py b/masque/file/gdsii_lazy_core.py index 88c4eed..d10a1a9 100644 --- a/masque/file/gdsii_lazy_core.py +++ b/masque/file/gdsii_lazy_core.py @@ -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)) diff --git a/masque/library.py b/masque/library.py index bcc23ce..0a8ae73 100644 --- a/masque/library.py +++ b/masque/library.py @@ -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: diff --git a/masque/test/test_gdsii_lazy.py b/masque/test/test_gdsii_lazy.py index 0134d6e..3abaab8 100644 --- a/masque/test/test_gdsii_lazy.py +++ b/masque/test/test_gdsii_lazy.py @@ -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: diff --git a/masque/test/test_gdsii_lazy_arrow.py b/masque/test/test_gdsii_lazy_arrow.py index eb62721..0984b20 100644 --- a/masque/test/test_gdsii_lazy_arrow.py +++ b/masque/test/test_gdsii_lazy_arrow.py @@ -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)