[library] Split library.py into multiple files
This commit is contained in:
parent
4c64352152
commit
02a3708b30
9 changed files with 2952 additions and 2842 deletions
|
|
@ -18,7 +18,8 @@ import klamath
|
|||
from . import gdsii
|
||||
from .utils import tmpfile
|
||||
from ..error import LibraryError
|
||||
from ..library import ILibraryView, OverlayLibrary, _SourceEntry, _materialize_detached_pattern
|
||||
from ..library import ILibraryView, OverlayLibrary
|
||||
from ..library.overlay import _SourceEntry, _materialize_detached_pattern
|
||||
|
||||
if TYPE_CHECKING:
|
||||
from collections.abc import Iterator, Mapping
|
||||
|
|
|
|||
2841
masque/library.py
2841
masque/library.py
File diff suppressed because it is too large
Load diff
29
masque/library/__init__.py
Normal file
29
masque/library/__init__.py
Normal file
|
|
@ -0,0 +1,29 @@
|
|||
"""Library classes for managing name-to-pattern mappings."""
|
||||
from .utils import (
|
||||
SINGLE_USE_PREFIX as SINGLE_USE_PREFIX,
|
||||
Tree as Tree,
|
||||
TreeView as TreeView,
|
||||
b64suffix as b64suffix,
|
||||
dangling_mode_t as dangling_mode_t,
|
||||
visitor_function_t as visitor_function_t,
|
||||
)
|
||||
from .base import (
|
||||
AbstractView as AbstractView,
|
||||
ILibrary as ILibrary,
|
||||
ILibraryView as ILibraryView,
|
||||
)
|
||||
from .mapping import (
|
||||
Library as Library,
|
||||
LibraryView as LibraryView,
|
||||
)
|
||||
from .overlay import (
|
||||
OverlayLibrary as OverlayLibrary,
|
||||
PortsLibraryView as PortsLibraryView,
|
||||
)
|
||||
from .build import (
|
||||
BuildLibrary as BuildLibrary,
|
||||
BuildReport as BuildReport,
|
||||
CellProvenance as CellProvenance,
|
||||
cell as cell,
|
||||
)
|
||||
from .lazy import LazyLibrary as LazyLibrary
|
||||
1259
masque/library/base.py
Normal file
1259
masque/library/base.py
Normal file
File diff suppressed because it is too large
Load diff
742
masque/library/build.py
Normal file
742
masque/library/build.py
Normal file
|
|
@ -0,0 +1,742 @@
|
|||
"""Two-phase build library implementation."""
|
||||
from __future__ import annotations
|
||||
|
||||
from collections import defaultdict
|
||||
from contextvars import ContextVar
|
||||
from dataclasses import dataclass, replace
|
||||
from functools import wraps
|
||||
from pprint import pformat
|
||||
from typing import TYPE_CHECKING, Any, Literal, Self, cast
|
||||
import copy
|
||||
|
||||
from ..error import BuildError, LibraryError
|
||||
from .base import ILibrary, ILibraryView
|
||||
from .utils import TreeView, dangling_mode_t, _plan_source_names, _rename_patterns, _source_rename_map
|
||||
from .mapping import Library, LibraryView
|
||||
from .overlay import OverlayLibrary
|
||||
|
||||
if TYPE_CHECKING:
|
||||
from collections.abc import Callable, Iterator, Mapping, Sequence
|
||||
|
||||
from ..abstract import Abstract
|
||||
from ..pattern import Pattern
|
||||
|
||||
|
||||
_ACTIVE_BUILD_SESSIONS: ContextVar[dict[int, _BuildSessionLibrary] | None] = ContextVar(
|
||||
'masque_active_build_sessions',
|
||||
default=None,
|
||||
)
|
||||
|
||||
|
||||
@dataclass(frozen=True)
|
||||
class CellProvenance:
|
||||
"""
|
||||
Provenance record for one cell in a completed build output.
|
||||
|
||||
Each output name in a `BuildReport` maps to one `CellProvenance`. The
|
||||
record captures both where the cell came from and how its visible name was
|
||||
chosen.
|
||||
|
||||
Attributes:
|
||||
requested_name: First name requested for this cell during the build.
|
||||
kind: Whether the cell came from a declaration, helper emission, or an
|
||||
imported source library.
|
||||
owner_declared_name: Declared cell responsible for this output cell, if
|
||||
any. Imported source cells leave this as `None`.
|
||||
build_chain: Declared-cell dependency chain that was active when the
|
||||
cell was emitted.
|
||||
"""
|
||||
requested_name: str
|
||||
kind: Literal['declared', 'helper', 'source']
|
||||
owner_declared_name: str | None
|
||||
build_chain: tuple[str, ...]
|
||||
|
||||
|
||||
@dataclass(frozen=True)
|
||||
class BuildReport:
|
||||
"""
|
||||
Immutable summary of one `BuildLibrary.validate()` or `.build()` run.
|
||||
|
||||
The report is designed to answer two questions after a build completes:
|
||||
which declared cells depended on which other declared cells, and where each
|
||||
output cell came from.
|
||||
|
||||
Attributes:
|
||||
requested_roots: Roots explicitly requested for the run. A full
|
||||
`build()` uses all declared cells.
|
||||
provenance: Mapping from final output name to provenance metadata.
|
||||
dependency_graph: Declared-cell dependency graph discovered through
|
||||
library-mediated reads and explicit recipe hints.
|
||||
"""
|
||||
requested_roots: tuple[str, ...]
|
||||
provenance: Mapping[str, CellProvenance]
|
||||
dependency_graph: Mapping[str, frozenset[str]]
|
||||
|
||||
@dataclass
|
||||
class _BuildRecipe:
|
||||
""" Captured deferred call to a pattern factory. """
|
||||
func: Callable[..., Pattern]
|
||||
args: tuple[Any, ...]
|
||||
kwargs: dict[str, Any]
|
||||
explicit_dependencies: tuple[str, ...] = ()
|
||||
|
||||
def depends_on(self, *names: str) -> _BuildRecipe:
|
||||
self.explicit_dependencies += tuple(names)
|
||||
return self
|
||||
|
||||
|
||||
def cell(func: Callable[..., Pattern]) -> Callable[..., _BuildRecipe]:
|
||||
"""
|
||||
Wrap a plain pattern factory so calls return deferred build recipes.
|
||||
|
||||
Use as either `cell(fn)(...)` or `@cell`.
|
||||
"""
|
||||
@wraps(func)
|
||||
def wrapper(*args: Any, **kwargs: Any) -> _BuildRecipe:
|
||||
return _BuildRecipe(func=func, args=args, kwargs=kwargs)
|
||||
|
||||
return wrapper
|
||||
|
||||
|
||||
class BuildCellsView:
|
||||
"""
|
||||
Attribute-based declaration namespace for `BuildLibrary`.
|
||||
|
||||
This is the ergonomic authoring surface exposed as `builder.cells`. It is
|
||||
intentionally write-focused: attribute assignment and deletion register
|
||||
declarations, while attribute reads fail with guidance to build first and
|
||||
use the returned library.
|
||||
"""
|
||||
|
||||
def __init__(self, library: BuildLibrary) -> None:
|
||||
object.__setattr__(self, '_library', library)
|
||||
|
||||
def __getattr__(self, name: str) -> Pattern:
|
||||
raise BuildError(
|
||||
f'BuildLibrary.cells.{name} is write-only during authoring. '
|
||||
'Call build() and index the returned library instead.'
|
||||
)
|
||||
|
||||
def __setattr__(self, name: str, value: Pattern | _BuildRecipe) -> None:
|
||||
if name.startswith('_'):
|
||||
object.__setattr__(self, name, value)
|
||||
return
|
||||
self._library[name] = value
|
||||
|
||||
def __delattr__(self, name: str) -> None:
|
||||
if name.startswith('_'):
|
||||
raise AttributeError(name)
|
||||
del self._library[name]
|
||||
|
||||
|
||||
class BuildLibrary(ILibrary):
|
||||
"""
|
||||
Two-phase declaration surface for mixed imported/generated libraries.
|
||||
|
||||
A `BuildLibrary` collects three kinds of inputs:
|
||||
- direct declared `Pattern` objects
|
||||
- deferred recipes created with `cell(...)`
|
||||
- imported source-backed library views added with `add_source(...)`
|
||||
|
||||
The builder itself is not a normal readable library during authoring.
|
||||
Instead, `validate()` and `build()` create a temporary build-session library
|
||||
that recipes can read from and write helper cells into while dependencies
|
||||
are resolved. `build()` then freezes the builder on success and returns the
|
||||
built library plus a `BuildReport`.
|
||||
"""
|
||||
|
||||
def __init__(self) -> None:
|
||||
self.cells = BuildCellsView(self)
|
||||
self._frozen = False
|
||||
self._declarations: dict[str, Pattern | _BuildRecipe] = {}
|
||||
self._sources: list[tuple[ILibraryView, dict[str, str]]] = []
|
||||
self._names: dict[str, None] = {}
|
||||
|
||||
def _active_session(self) -> _BuildSessionLibrary | None:
|
||||
sessions = _ACTIVE_BUILD_SESSIONS.get()
|
||||
if sessions is None:
|
||||
return None
|
||||
return sessions.get(id(self))
|
||||
|
||||
def _require_active_session(self, operation: str) -> _BuildSessionLibrary:
|
||||
session = self._active_session()
|
||||
if session is None:
|
||||
raise BuildError(
|
||||
f'BuildLibrary.{operation}() is only available while validate() or build() is running. '
|
||||
'Use the built output library for reads.'
|
||||
)
|
||||
return session
|
||||
|
||||
def _assert_editable(self) -> None:
|
||||
if self._frozen:
|
||||
raise BuildError('This BuildLibrary has already been built successfully and is now frozen.')
|
||||
|
||||
def __iter__(self) -> Iterator[str]:
|
||||
session = self._active_session()
|
||||
if session is not None:
|
||||
return iter(session)
|
||||
return iter(self._names)
|
||||
|
||||
def __len__(self) -> int:
|
||||
session = self._active_session()
|
||||
if session is not None:
|
||||
return len(session)
|
||||
return len(self._names)
|
||||
|
||||
def __contains__(self, key: object) -> bool:
|
||||
session = self._active_session()
|
||||
if session is not None:
|
||||
return key in session
|
||||
return key in self._names
|
||||
|
||||
def __getitem__(self, key: str) -> Pattern:
|
||||
return self._require_active_session('__getitem__')[key]
|
||||
|
||||
def __setitem__(
|
||||
self,
|
||||
key: str,
|
||||
value: Pattern | _BuildRecipe,
|
||||
) -> None:
|
||||
session = self._active_session()
|
||||
if session is not None:
|
||||
session[key] = value
|
||||
return
|
||||
|
||||
self._assert_editable()
|
||||
if key in self._names:
|
||||
raise LibraryError(f'"{key}" already exists in the builder. Overwriting is not allowed!')
|
||||
|
||||
if isinstance(value, _BuildRecipe):
|
||||
declaration = value
|
||||
else:
|
||||
if callable(value):
|
||||
raise TypeError('BuildLibrary recipes must be wrapped with cell(fn)(...) or @cell.')
|
||||
declaration = value
|
||||
|
||||
self._declarations[key] = declaration
|
||||
self._names[key] = None
|
||||
|
||||
def __delitem__(self, key: str) -> None:
|
||||
session = self._active_session()
|
||||
if session is not None:
|
||||
del session[key]
|
||||
return
|
||||
|
||||
self._assert_editable()
|
||||
if key not in self._declarations:
|
||||
raise KeyError(key)
|
||||
del self._declarations[key]
|
||||
del self._names[key]
|
||||
|
||||
def _merge(self, key_self: str, other: Mapping[str, Pattern], key_other: str) -> None:
|
||||
session = self._active_session()
|
||||
if session is not None:
|
||||
session._merge(key_self, other, key_other)
|
||||
return
|
||||
self[key_self] = copy.deepcopy(other[key_other])
|
||||
|
||||
def add(
|
||||
self,
|
||||
other: Mapping[str, Pattern],
|
||||
rename_theirs: Callable[[ILibraryView, str], str] = _rename_patterns,
|
||||
mutate_other: bool = False,
|
||||
) -> dict[str, str]:
|
||||
from ..pattern import map_targets # noqa: PLC0415
|
||||
|
||||
session = self._active_session()
|
||||
if session is not None:
|
||||
return session.add(other, rename_theirs=rename_theirs, mutate_other=mutate_other)
|
||||
|
||||
self._assert_editable()
|
||||
|
||||
source_backed = isinstance(other, ILibraryView) and not isinstance(other, Library | LibraryView)
|
||||
if source_backed:
|
||||
if mutate_other:
|
||||
raise BuildError('BuildLibrary.add(..., mutate_other=True) is not supported for source-backed inputs.')
|
||||
return self.add_source(
|
||||
other,
|
||||
rename_theirs = rename_theirs,
|
||||
rename_when = 'conflict',
|
||||
)
|
||||
|
||||
source_order = tuple(other.keys())
|
||||
source_to_visible = _plan_source_names(
|
||||
self,
|
||||
source_order,
|
||||
self._names,
|
||||
rename_theirs = rename_theirs,
|
||||
rename_when = 'conflict',
|
||||
)
|
||||
rename_map = _source_rename_map(source_to_visible)
|
||||
|
||||
if mutate_other:
|
||||
temp = other
|
||||
else:
|
||||
temp = Library(copy.deepcopy(dict(other)))
|
||||
|
||||
for source_name in source_order:
|
||||
visible_name = source_to_visible[source_name]
|
||||
pattern = temp[source_name]
|
||||
if rename_map:
|
||||
pattern.refs = map_targets(
|
||||
pattern.refs,
|
||||
lambda target: cast('dict[str | None, str | None]', rename_map).get(target, target),
|
||||
)
|
||||
self[visible_name] = pattern
|
||||
|
||||
return rename_map
|
||||
|
||||
def __lshift__(self, other: TreeView) -> str:
|
||||
session = self._active_session()
|
||||
if session is not None:
|
||||
return session << other
|
||||
|
||||
self._assert_editable()
|
||||
if len(other) == 1:
|
||||
name = next(iter(other))
|
||||
elif isinstance(other, ILibraryView) and not isinstance(other, Library | LibraryView):
|
||||
source_order = other.source_order()
|
||||
child_graph = other.child_graph(dangling='include')
|
||||
referenced = set().union(*child_graph.values()) if child_graph else set()
|
||||
tops = [candidate for candidate in source_order if candidate not in referenced]
|
||||
if len(tops) != 1:
|
||||
raise LibraryError(f'Asked for the single topcell, but found the following: {pformat(tops)}')
|
||||
name = tops[0]
|
||||
else:
|
||||
return super().__lshift__(other)
|
||||
|
||||
rename_map = self.add(other)
|
||||
return rename_map.get(name, name)
|
||||
|
||||
def __le__(self, other: Mapping[str, Pattern]) -> Abstract:
|
||||
if self._active_session() is not None:
|
||||
return super().__le__(other)
|
||||
raise BuildError('BuildLibrary.__le__() is only available while validate() or build() is running.')
|
||||
|
||||
def rename(
|
||||
self,
|
||||
old_name: str,
|
||||
new_name: str,
|
||||
move_references: bool = False,
|
||||
) -> Self:
|
||||
"""
|
||||
Rename a helper cell during an active build session.
|
||||
|
||||
During authoring, declared cells must be registered under their
|
||||
intended final names and imported source cells must be renamed through
|
||||
`add_source(...)`.
|
||||
"""
|
||||
session = self._active_session()
|
||||
if session is not None:
|
||||
session.rename(old_name, new_name, move_references=move_references)
|
||||
return self
|
||||
|
||||
self._assert_editable()
|
||||
if old_name == new_name:
|
||||
return self
|
||||
if old_name in self._declarations:
|
||||
raise BuildError(
|
||||
f'Cannot rename declared build cell "{old_name}" during authoring. '
|
||||
'Register it under the intended final name instead.'
|
||||
)
|
||||
if old_name not in self._names:
|
||||
raise LibraryError(f'"{old_name}" does not exist in the builder.')
|
||||
raise BuildError(
|
||||
f'Cannot rename imported source cell "{old_name}" during authoring. '
|
||||
'Choose visible source names with add_source(..., rename_theirs=..., rename_when=...).'
|
||||
)
|
||||
|
||||
def abstract(self, name: str) -> Abstract:
|
||||
return self._require_active_session('abstract').abstract(name)
|
||||
|
||||
def resolve(
|
||||
self,
|
||||
other: Abstract | str | Pattern | TreeView,
|
||||
append: bool = False,
|
||||
) -> Abstract | Pattern:
|
||||
return self._require_active_session('resolve').resolve(other, append=append)
|
||||
|
||||
def add_source(
|
||||
self,
|
||||
source: Mapping[str, Pattern] | ILibraryView,
|
||||
*,
|
||||
rename_theirs: Callable[[ILibraryView, str], str] | None = None,
|
||||
rename_when: Literal['conflict', 'always'] = 'conflict',
|
||||
) -> dict[str, str]:
|
||||
"""
|
||||
Register an imported source-backed library with the builder.
|
||||
|
||||
The source is not materialized immediately. Its names are scanned once
|
||||
to reserve visible builder names, then the source is read again when a
|
||||
build session starts. The source's cell membership must not be
|
||||
structurally mutated between `add_source()` and `build()`/`validate()`.
|
||||
Source cells may be renamed on entry to avoid collisions with existing
|
||||
declarations or other imported sources.
|
||||
|
||||
Args:
|
||||
rename_theirs: Function used to choose visible names for imported
|
||||
source cells.
|
||||
rename_when: If `'conflict'`, only conflicting names are renamed.
|
||||
If `'always'`, every imported source name is passed through
|
||||
`rename_theirs`.
|
||||
|
||||
Returns:
|
||||
Mapping of `{source_name: visible_name}` for imported names that
|
||||
were renamed while being added.
|
||||
"""
|
||||
if self._active_session() is not None:
|
||||
raise BuildError('BuildLibrary.add_source() is only available while authoring, not during validate() or build().')
|
||||
self._assert_editable()
|
||||
|
||||
view = source if isinstance(source, ILibraryView) else LibraryView(source)
|
||||
source_order = tuple(view.source_order())
|
||||
source_to_visible = _plan_source_names(
|
||||
self,
|
||||
source_order,
|
||||
self._names,
|
||||
rename_theirs = rename_theirs,
|
||||
rename_when = rename_when,
|
||||
)
|
||||
|
||||
self._sources.append((view, dict(source_to_visible)))
|
||||
for source_name in source_order:
|
||||
visible = source_to_visible[source_name]
|
||||
self._names[visible] = None
|
||||
return _source_rename_map(source_to_visible)
|
||||
|
||||
def validate(
|
||||
self,
|
||||
names: Sequence[str] | None = None,
|
||||
*,
|
||||
allow_dangling: bool = False,
|
||||
) -> BuildReport:
|
||||
"""
|
||||
Run the full build logic and return a `BuildReport` without producing output.
|
||||
|
||||
This is a dry run over the same dependency resolution and recipe
|
||||
execution path used by `build()`. Any generated library is discarded
|
||||
after validation completes.
|
||||
"""
|
||||
_session, report = self._run_build(names=names, allow_dangling=allow_dangling)
|
||||
return report
|
||||
|
||||
def build(
|
||||
self,
|
||||
*,
|
||||
output: Literal['overlay', 'library'] = 'overlay',
|
||||
allow_dangling: bool = False,
|
||||
) -> tuple[ILibrary, BuildReport]:
|
||||
"""
|
||||
Materialize declarations and return a usable output library plus report.
|
||||
|
||||
Args:
|
||||
output: `'overlay'` preserves imported source-backed cells where
|
||||
possible, while `'library'` eagerly materializes the full
|
||||
result.
|
||||
allow_dangling: If `False`, fail the build when the completed
|
||||
library still contains dangling references.
|
||||
"""
|
||||
if output not in ('overlay', 'library'):
|
||||
raise ValueError(f'Unknown build output mode: {output!r}')
|
||||
self._assert_editable()
|
||||
session, report = self._run_build(names=None, allow_dangling=allow_dangling)
|
||||
if output == 'library':
|
||||
built_output = session.to_library()
|
||||
else:
|
||||
built_output = session.to_overlay()
|
||||
self._frozen = True
|
||||
return built_output, report
|
||||
|
||||
def _run_build(
|
||||
self,
|
||||
*,
|
||||
names: Sequence[str] | None,
|
||||
allow_dangling: bool,
|
||||
) -> tuple[_BuildSessionLibrary, BuildReport]:
|
||||
roots = tuple(dict.fromkeys(names if names is not None else self._declarations.keys()))
|
||||
unknown = [name for name in roots if name not in self._names]
|
||||
if unknown:
|
||||
raise BuildError(f'Unknown build roots requested: {unknown}')
|
||||
|
||||
session = _BuildSessionLibrary(self)
|
||||
sessions = dict(_ACTIVE_BUILD_SESSIONS.get() or {})
|
||||
sessions[id(self)] = session
|
||||
token = _ACTIVE_BUILD_SESSIONS.set(sessions)
|
||||
try:
|
||||
session.materialize_many(roots)
|
||||
if not allow_dangling:
|
||||
session.child_graph(dangling='error')
|
||||
finally:
|
||||
_ACTIVE_BUILD_SESSIONS.reset(token)
|
||||
|
||||
report = session.build_report(roots)
|
||||
return session, report
|
||||
|
||||
|
||||
class _BuildSessionLibrary(ILibrary):
|
||||
"""
|
||||
Internal overlay-backed library used while a `BuildLibrary` is executing.
|
||||
|
||||
This object provides the mutable-library surface that recipes expect while
|
||||
also tracking declared-cell dependencies, helper-cell provenance, and
|
||||
imported source cells. It exists only for the duration of a validation or
|
||||
build run.
|
||||
"""
|
||||
|
||||
def __init__(self, builder: BuildLibrary) -> None:
|
||||
self._builder = builder
|
||||
self._overlay = OverlayLibrary()
|
||||
self._built: set[str] = set()
|
||||
self._declared_stack: list[str] = []
|
||||
self._names = dict(builder._names)
|
||||
self._provenance: dict[str, CellProvenance] = {}
|
||||
self._dependency_graph: defaultdict[str, set[str]] = defaultdict(set)
|
||||
self._install_sources()
|
||||
|
||||
def _install_sources(self) -> None:
|
||||
for source_library, source_to_visible in self._builder._sources:
|
||||
source_order = source_library.source_order()
|
||||
expected_names = set(source_to_visible)
|
||||
actual_names = set(source_order)
|
||||
if actual_names != expected_names:
|
||||
added_names = sorted(actual_names - expected_names)
|
||||
removed_names = sorted(expected_names - actual_names)
|
||||
detail = []
|
||||
if added_names:
|
||||
detail.append(f'added={added_names}')
|
||||
if removed_names:
|
||||
detail.append(f'removed={removed_names}')
|
||||
raise BuildError(
|
||||
'Imported source library changed after add_source() was called '
|
||||
f'({", ".join(detail)}). '
|
||||
'Do not structurally mutate source libraries between add_source() and build()/validate().'
|
||||
)
|
||||
|
||||
def rename_source(_lib: ILibraryView, name: str, *, mapping: Mapping[str, str] = source_to_visible) -> str:
|
||||
return mapping[name]
|
||||
|
||||
self._overlay.add_source(
|
||||
source_library,
|
||||
rename_theirs = rename_source,
|
||||
rename_when = 'always',
|
||||
)
|
||||
|
||||
for source_name in source_order:
|
||||
visible_name = source_to_visible[source_name]
|
||||
self._provenance[visible_name] = CellProvenance(
|
||||
requested_name = source_name,
|
||||
kind = 'source',
|
||||
owner_declared_name = None,
|
||||
build_chain = (),
|
||||
)
|
||||
|
||||
def __iter__(self) -> Iterator[str]:
|
||||
return iter(self._names)
|
||||
|
||||
def __len__(self) -> int:
|
||||
return len(self._names)
|
||||
|
||||
def __contains__(self, key: object) -> bool:
|
||||
return key in self._names
|
||||
|
||||
def _current_declared(self) -> str | None:
|
||||
if not self._declared_stack:
|
||||
return None
|
||||
return self._declared_stack[-1]
|
||||
|
||||
def _record_dependency(self, target: str) -> None:
|
||||
current = self._current_declared()
|
||||
if current is None or current == target or target not in self._builder._declarations:
|
||||
return
|
||||
self._dependency_graph[current].add(target)
|
||||
|
||||
def _guard_mutable_output_name(self, key: str, *, operation: str) -> None:
|
||||
if key in self._builder._declarations:
|
||||
raise BuildError(f'Cannot {operation} declared build cell "{key}" during an active build session.')
|
||||
|
||||
provenance = self._provenance.get(key)
|
||||
if provenance is not None and provenance.kind == 'source':
|
||||
raise BuildError(f'Cannot {operation} imported source cell "{key}" during an active build session.')
|
||||
|
||||
def rename(
|
||||
self,
|
||||
old_name: str,
|
||||
new_name: str,
|
||||
move_references: bool = False,
|
||||
) -> Self:
|
||||
if old_name == new_name:
|
||||
return self
|
||||
if old_name not in self._overlay:
|
||||
if old_name in self._builder._declarations:
|
||||
self._guard_mutable_output_name(old_name, operation='rename')
|
||||
raise LibraryError(f'"{old_name}" does not exist in the library.')
|
||||
|
||||
self._guard_mutable_output_name(old_name, operation='rename')
|
||||
if new_name in self._names:
|
||||
raise LibraryError(f'"{new_name}" already exists in the library.')
|
||||
|
||||
self._overlay.rename(old_name, new_name, move_references=move_references)
|
||||
self._names = {
|
||||
new_name if name == old_name else name: None
|
||||
for name in self._names
|
||||
}
|
||||
|
||||
provenance = self._provenance.pop(old_name)
|
||||
self._provenance[new_name] = provenance
|
||||
return self
|
||||
|
||||
def __getitem__(self, key: str) -> Pattern:
|
||||
if key in self._builder._declarations:
|
||||
self._record_dependency(key)
|
||||
self._ensure_declared(key)
|
||||
return self._overlay[key]
|
||||
|
||||
def __setitem__(
|
||||
self,
|
||||
key: str,
|
||||
value: Pattern | Callable[[], Pattern],
|
||||
) -> None:
|
||||
if key in self._overlay:
|
||||
raise LibraryError(f'"{key}" already exists in the library. Overwriting is not allowed!')
|
||||
current = self._current_declared()
|
||||
if key in self._builder._declarations and key != current:
|
||||
raise LibraryError(f'"{key}" is reserved for a declared cell and cannot be used as a helper name.')
|
||||
|
||||
pattern = value() if callable(value) else value
|
||||
self._overlay[key] = pattern
|
||||
self._names.setdefault(key, None)
|
||||
|
||||
kind: Literal['declared', 'helper']
|
||||
if current is not None and key == current:
|
||||
kind = 'declared'
|
||||
else:
|
||||
kind = 'helper'
|
||||
|
||||
self._provenance[key] = CellProvenance(
|
||||
requested_name = key,
|
||||
kind = kind,
|
||||
owner_declared_name = current if kind == 'helper' else key,
|
||||
build_chain = tuple(self._declared_stack),
|
||||
)
|
||||
|
||||
def __delitem__(self, key: str) -> None:
|
||||
if key not in self._overlay:
|
||||
if key in self._builder._declarations:
|
||||
self._guard_mutable_output_name(key, operation='delete')
|
||||
raise KeyError(key)
|
||||
|
||||
self._guard_mutable_output_name(key, operation='delete')
|
||||
if key in self._overlay:
|
||||
del self._overlay[key]
|
||||
self._names.pop(key, None)
|
||||
self._provenance.pop(key, None)
|
||||
|
||||
def _merge(self, key_self: str, other: Mapping[str, Pattern], key_other: str) -> None:
|
||||
self[key_self] = copy.deepcopy(other[key_other])
|
||||
|
||||
def add(
|
||||
self,
|
||||
other: Mapping[str, Pattern],
|
||||
rename_theirs: Callable[[ILibraryView, str], str] = _rename_patterns,
|
||||
mutate_other: bool = False,
|
||||
) -> dict[str, str]:
|
||||
rename_map = super().add(other, rename_theirs=rename_theirs, mutate_other=mutate_other)
|
||||
|
||||
current = self._current_declared()
|
||||
for old_name, new_name in rename_map.items():
|
||||
if new_name in self._provenance:
|
||||
self._provenance[new_name] = replace(
|
||||
self._provenance[new_name],
|
||||
requested_name = old_name,
|
||||
owner_declared_name = current if current is not None else self._provenance[new_name].owner_declared_name,
|
||||
)
|
||||
return rename_map
|
||||
|
||||
def _wrap_error(self, name: str, exc: Exception) -> BuildError:
|
||||
chain = tuple(self._declared_stack)
|
||||
msg = [f'Failed while building declared cell "{name}"']
|
||||
if chain:
|
||||
msg.append(f'Dependency chain: {" -> ".join(chain)}')
|
||||
msg.append(f'Cause: {exc}')
|
||||
return BuildError('\n'.join(msg))
|
||||
|
||||
def _ensure_named(self, name: str) -> None:
|
||||
if name in self._builder._declarations:
|
||||
self._record_dependency(name)
|
||||
self._ensure_declared(name)
|
||||
return
|
||||
if name in self._overlay:
|
||||
return
|
||||
raise BuildError(f'Missing dependency "{name}"')
|
||||
|
||||
def _ensure_declared(self, name: str) -> None:
|
||||
from ..pattern import Pattern # noqa: PLC0415
|
||||
|
||||
if name in self._built:
|
||||
return
|
||||
if name in self._declared_stack:
|
||||
chain = ' -> '.join(self._declared_stack + [name])
|
||||
raise BuildError(f'Cycle detected while building declared cells: {chain}')
|
||||
|
||||
declaration = self._builder._declarations[name]
|
||||
self._declared_stack.append(name)
|
||||
try:
|
||||
if isinstance(declaration, _BuildRecipe):
|
||||
for dep in declaration.explicit_dependencies:
|
||||
self._ensure_named(dep)
|
||||
pattern = declaration.func(*declaration.args, **declaration.kwargs)
|
||||
if not isinstance(pattern, Pattern):
|
||||
raise BuildError(f'Recipe for "{name}" returned {type(pattern).__name__}, expected Pattern') # noqa: TRY301
|
||||
else:
|
||||
pattern = declaration.deepcopy()
|
||||
|
||||
if name in self._overlay:
|
||||
if self._overlay[name] is not pattern:
|
||||
raise BuildError( # noqa: TRY301
|
||||
f'Recipe for "{name}" wrote a different pattern into the session under its own name.'
|
||||
)
|
||||
else:
|
||||
self[name] = pattern
|
||||
self._built.add(name)
|
||||
except Exception as exc:
|
||||
raise self._wrap_error(name, exc) from exc
|
||||
finally:
|
||||
self._declared_stack.pop()
|
||||
|
||||
def materialize_many(self, names: Sequence[str]) -> None:
|
||||
for name in dict.fromkeys(names):
|
||||
self._ensure_named(name)
|
||||
|
||||
def source_order(self) -> tuple[str, ...]:
|
||||
return self._overlay.source_order()
|
||||
|
||||
def child_graph(
|
||||
self,
|
||||
dangling: dangling_mode_t = 'error',
|
||||
) -> dict[str, set[str]]:
|
||||
return self._overlay.child_graph(dangling=dangling)
|
||||
|
||||
def parent_graph(
|
||||
self,
|
||||
dangling: dangling_mode_t = 'error',
|
||||
) -> dict[str, set[str]]:
|
||||
return self._overlay.parent_graph(dangling=dangling)
|
||||
|
||||
def build_report(self, requested_roots: Sequence[str]) -> BuildReport:
|
||||
dependency_graph = {
|
||||
name: frozenset(self._dependency_graph.get(name, set()))
|
||||
for name in self._builder._declarations
|
||||
if name in self._dependency_graph or name in requested_roots
|
||||
}
|
||||
return BuildReport(
|
||||
requested_roots = tuple(dict.fromkeys(requested_roots)),
|
||||
provenance = dict(self._provenance),
|
||||
dependency_graph = dependency_graph,
|
||||
)
|
||||
|
||||
def to_overlay(self) -> ILibrary:
|
||||
return self._overlay
|
||||
|
||||
def to_library(self) -> Library:
|
||||
mapping = {name: self._overlay[name] for name in self._overlay.source_order()}
|
||||
return Library(mapping)
|
||||
169
masque/library/lazy.py
Normal file
169
masque/library/lazy.py
Normal file
|
|
@ -0,0 +1,169 @@
|
|||
"""Closure-backed lazy library implementation."""
|
||||
from __future__ import annotations
|
||||
|
||||
from pprint import pformat
|
||||
from typing import TYPE_CHECKING, Self, cast
|
||||
import logging
|
||||
|
||||
from ..error import LibraryError
|
||||
from .base import ILibrary
|
||||
|
||||
if TYPE_CHECKING:
|
||||
from collections.abc import Callable, Iterator, Mapping
|
||||
|
||||
from ..pattern import Pattern
|
||||
|
||||
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
|
||||
class LazyLibrary(ILibrary):
|
||||
"""
|
||||
This class is usually used to create a library of Patterns by mapping names to
|
||||
functions which generate or load the relevant `Pattern` object as-needed.
|
||||
|
||||
TODO: lots of stuff causes recursive loads (e.g. data_to_ports?). What should you avoid?
|
||||
"""
|
||||
mapping: dict[str, Callable[[], Pattern]]
|
||||
cache: dict[str, Pattern]
|
||||
_lookups_in_progress: list[str]
|
||||
|
||||
def __init__(self) -> None:
|
||||
self.mapping = {}
|
||||
self.cache = {}
|
||||
self._lookups_in_progress = []
|
||||
|
||||
def __setitem__(
|
||||
self,
|
||||
key: str,
|
||||
value: Pattern | Callable[[], Pattern],
|
||||
) -> None:
|
||||
if key in self.mapping:
|
||||
raise LibraryError(f'"{key}" already exists in the library. Overwriting is not allowed!')
|
||||
|
||||
if callable(value):
|
||||
value_func = value
|
||||
else:
|
||||
value_func = lambda: cast('Pattern', value) # noqa: E731
|
||||
|
||||
self.mapping[key] = value_func
|
||||
if key in self.cache:
|
||||
del self.cache[key]
|
||||
|
||||
def __delitem__(self, key: str) -> None:
|
||||
del self.mapping[key]
|
||||
if key in self.cache:
|
||||
del self.cache[key]
|
||||
|
||||
def __getitem__(self, key: str) -> Pattern:
|
||||
logger.debug(f'loading {key}')
|
||||
if key in self.cache:
|
||||
logger.debug(f'found {key} in cache')
|
||||
return self.cache[key]
|
||||
|
||||
if key in self._lookups_in_progress:
|
||||
chain = ' -> '.join(self._lookups_in_progress + [key])
|
||||
raise LibraryError(
|
||||
f'Detected circular reference or recursive lookup of "{key}".\n'
|
||||
f'Lookup chain: {chain}\n'
|
||||
'This may be caused by an invalid (cyclical) reference, or buggy code.\n'
|
||||
'If you are lazy-loading a file, try a non-lazy load and check for reference cycles.'
|
||||
)
|
||||
|
||||
self._lookups_in_progress.append(key)
|
||||
try:
|
||||
func = self.mapping[key]
|
||||
pat = func()
|
||||
finally:
|
||||
self._lookups_in_progress.pop()
|
||||
self.cache[key] = pat
|
||||
return pat
|
||||
|
||||
def __iter__(self) -> Iterator[str]:
|
||||
return iter(self.mapping)
|
||||
|
||||
def __len__(self) -> int:
|
||||
return len(self.mapping)
|
||||
|
||||
def __contains__(self, key: object) -> bool:
|
||||
return key in self.mapping
|
||||
|
||||
def _merge(self, key_self: str, other: Mapping[str, Pattern], key_other: str) -> None:
|
||||
if isinstance(other, LazyLibrary):
|
||||
self.mapping[key_self] = other.mapping[key_other]
|
||||
if key_other in other.cache:
|
||||
self.cache[key_self] = other.cache[key_other]
|
||||
else:
|
||||
self[key_self] = other[key_other]
|
||||
|
||||
def __repr__(self) -> str:
|
||||
return '<LazyLibrary with keys\n' + pformat(list(self.keys())) + '>'
|
||||
|
||||
def rename(
|
||||
self,
|
||||
old_name: str,
|
||||
new_name: str,
|
||||
move_references: bool = False,
|
||||
) -> Self:
|
||||
"""
|
||||
Rename a pattern.
|
||||
|
||||
Args:
|
||||
old_name: Current name for the pattern
|
||||
new_name: New name for the pattern
|
||||
move_references: Whether to scan all refs in the pattern and
|
||||
move them to point to `new_name` as necessary.
|
||||
Default `False`.
|
||||
|
||||
Returns:
|
||||
self
|
||||
"""
|
||||
if old_name not in self.mapping:
|
||||
raise LibraryError(f'"{old_name}" does not exist in the library.')
|
||||
if old_name == new_name:
|
||||
return self
|
||||
|
||||
self[new_name] = self.mapping[old_name] # copy over function
|
||||
if old_name in self.cache:
|
||||
self.cache[new_name] = self.cache[old_name]
|
||||
del self[old_name]
|
||||
|
||||
if move_references:
|
||||
self.move_references(old_name, new_name)
|
||||
|
||||
return self
|
||||
|
||||
def move_references(self, old_target: str, new_target: str) -> Self:
|
||||
"""
|
||||
Change all references pointing at `old_target` into references pointing at `new_target`.
|
||||
|
||||
Args:
|
||||
old_target: Current reference target
|
||||
new_target: New target for the reference
|
||||
|
||||
Returns:
|
||||
self
|
||||
"""
|
||||
if old_target == new_target:
|
||||
return self
|
||||
|
||||
self.precache()
|
||||
for pattern in self.cache.values():
|
||||
if old_target in pattern.refs:
|
||||
pattern.refs[new_target].extend(pattern.refs[old_target])
|
||||
del pattern.refs[old_target]
|
||||
return self
|
||||
|
||||
def precache(self) -> Self:
|
||||
"""
|
||||
Force all patterns into the cache
|
||||
|
||||
Returns:
|
||||
self
|
||||
"""
|
||||
for key in self.mapping:
|
||||
_ = self[key] # want to trigger our own __getitem__
|
||||
return self
|
||||
|
||||
def __deepcopy__(self, memo: dict | None = None) -> LazyLibrary:
|
||||
raise LibraryError('LazyLibrary cannot be deepcopied (deepcopy doesn\'t descend into closures)')
|
||||
112
masque/library/mapping.py
Normal file
112
masque/library/mapping.py
Normal file
|
|
@ -0,0 +1,112 @@
|
|||
"""Concrete mapping-backed library implementations."""
|
||||
from __future__ import annotations
|
||||
|
||||
from pprint import pformat
|
||||
from typing import TYPE_CHECKING, Self
|
||||
|
||||
from ..error import LibraryError
|
||||
from .base import ILibrary, ILibraryView
|
||||
|
||||
if TYPE_CHECKING:
|
||||
from collections.abc import Callable, Iterator, Mapping, MutableMapping
|
||||
|
||||
from ..pattern import Pattern
|
||||
|
||||
|
||||
class LibraryView(ILibraryView):
|
||||
"""
|
||||
Default implementation for a read-only library.
|
||||
|
||||
A library is a mapping from unique names (str) to collections of geometry (`Pattern`).
|
||||
This library is backed by an arbitrary python object which implements the `Mapping` interface.
|
||||
"""
|
||||
mapping: Mapping[str, Pattern]
|
||||
|
||||
def __init__(
|
||||
self,
|
||||
mapping: Mapping[str, Pattern],
|
||||
) -> None:
|
||||
self.mapping = mapping
|
||||
|
||||
def __getitem__(self, key: str) -> Pattern:
|
||||
return self.mapping[key]
|
||||
|
||||
def __iter__(self) -> Iterator[str]:
|
||||
return iter(self.mapping)
|
||||
|
||||
def __len__(self) -> int:
|
||||
return len(self.mapping)
|
||||
|
||||
def __contains__(self, key: object) -> bool:
|
||||
return key in self.mapping
|
||||
|
||||
def __repr__(self) -> str:
|
||||
return f'<LibraryView ({type(self.mapping)}) with keys\n' + pformat(list(self.keys())) + '>'
|
||||
|
||||
|
||||
class Library(ILibrary):
|
||||
"""
|
||||
Default implementation for a writeable library.
|
||||
|
||||
A library is a mapping from unique names (str) to collections of geometry (`Pattern`).
|
||||
This library is backed by an arbitrary python object which implements the `MutableMapping` interface.
|
||||
"""
|
||||
mapping: MutableMapping[str, Pattern]
|
||||
|
||||
def __init__(
|
||||
self,
|
||||
mapping: MutableMapping[str, Pattern] | None = None,
|
||||
) -> None:
|
||||
if mapping is None:
|
||||
self.mapping = {}
|
||||
else:
|
||||
self.mapping = mapping
|
||||
|
||||
def __getitem__(self, key: str) -> Pattern:
|
||||
return self.mapping[key]
|
||||
|
||||
def __iter__(self) -> Iterator[str]:
|
||||
return iter(self.mapping)
|
||||
|
||||
def __len__(self) -> int:
|
||||
return len(self.mapping)
|
||||
|
||||
def __contains__(self, key: object) -> bool:
|
||||
return key in self.mapping
|
||||
|
||||
def __setitem__(
|
||||
self,
|
||||
key: str,
|
||||
value: Pattern | Callable[[], Pattern],
|
||||
) -> None:
|
||||
if key in self.mapping:
|
||||
raise LibraryError(f'"{key}" already exists in the library. Overwriting is not allowed!')
|
||||
|
||||
value = value() if callable(value) else value
|
||||
self.mapping[key] = value
|
||||
|
||||
def __delitem__(self, key: str) -> None:
|
||||
del self.mapping[key]
|
||||
|
||||
def _merge(self, key_self: str, other: Mapping[str, Pattern], key_other: str) -> None:
|
||||
self[key_self] = other[key_other]
|
||||
|
||||
def __repr__(self) -> str:
|
||||
return f'<Library ({type(self.mapping)}) with keys\n' + pformat(list(self.keys())) + '>'
|
||||
|
||||
@classmethod
|
||||
def mktree(cls: type[Self], name: str) -> tuple[Self, Pattern]:
|
||||
"""
|
||||
Create a new Library and immediately add a pattern
|
||||
|
||||
Args:
|
||||
name: The name for the new pattern (usually the name of the topcell).
|
||||
|
||||
Returns:
|
||||
The newly created `Library` and the newly created `Pattern`
|
||||
"""
|
||||
from ..pattern import Pattern # noqa: PLC0415
|
||||
tree = cls()
|
||||
pat = Pattern()
|
||||
tree[name] = pat
|
||||
return tree, pat
|
||||
515
masque/library/overlay.py
Normal file
515
masque/library/overlay.py
Normal file
|
|
@ -0,0 +1,515 @@
|
|||
"""Overlay and ports-importing library views."""
|
||||
from __future__ import annotations
|
||||
|
||||
from collections import defaultdict
|
||||
from dataclasses import dataclass
|
||||
from typing import TYPE_CHECKING, Any, Literal, cast
|
||||
import copy
|
||||
|
||||
import numpy
|
||||
|
||||
from ..error import LibraryError
|
||||
from ..pattern import Pattern, map_targets
|
||||
from ..utils import apply_transforms, layer_t
|
||||
from .base import ILibrary, ILibraryView
|
||||
from .utils import dangling_mode_t, _plan_source_names, _source_rename_map
|
||||
from .mapping import LibraryView
|
||||
|
||||
if TYPE_CHECKING:
|
||||
from collections.abc import Callable, Iterator, Mapping, Sequence
|
||||
|
||||
from numpy.typing import NDArray
|
||||
|
||||
|
||||
@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)
|
||||
124
masque/library/utils.py
Normal file
124
masque/library/utils.py
Normal file
|
|
@ -0,0 +1,124 @@
|
|||
"""Shared types and helpers for library implementations."""
|
||||
from __future__ import annotations
|
||||
|
||||
from typing import TYPE_CHECKING, Literal, Protocol, TypeAlias
|
||||
from collections.abc import Callable, Container, Mapping, MutableMapping, Sequence
|
||||
|
||||
from ..error import LibraryError
|
||||
|
||||
if TYPE_CHECKING:
|
||||
import numpy
|
||||
from numpy.typing import NDArray
|
||||
|
||||
from ..pattern import Pattern
|
||||
from .base import ILibraryView
|
||||
|
||||
|
||||
class visitor_function_t(Protocol):
|
||||
""" Signature for `Library.dfs()` visitor functions. """
|
||||
def __call__(
|
||||
self,
|
||||
pattern: Pattern,
|
||||
hierarchy: tuple[str | None, ...],
|
||||
memo: dict,
|
||||
transform: NDArray[numpy.float64] | Literal[False],
|
||||
) -> Pattern:
|
||||
...
|
||||
|
||||
|
||||
TreeView: TypeAlias = Mapping[str, 'Pattern']
|
||||
""" A name-to-`Pattern` mapping which is expected to have only one top-level cell """
|
||||
|
||||
Tree: TypeAlias = MutableMapping[str, 'Pattern']
|
||||
""" A mutable name-to-`Pattern` mapping which is expected to have only one top-level cell """
|
||||
|
||||
dangling_mode_t: TypeAlias = Literal['error', 'ignore', 'include']
|
||||
""" How helpers should handle refs whose targets are not present in the library. """
|
||||
SINGLE_USE_PREFIX = '_'
|
||||
"""
|
||||
Names starting with this prefix are assumed to refer to single-use patterns,
|
||||
which may be renamed automatically by `ILibrary.add()` (via
|
||||
`rename_theirs=_rename_patterns()` )
|
||||
"""
|
||||
# TODO what are the consequences of making '_' special? maybe we can make this decision everywhere?
|
||||
|
||||
|
||||
def _rename_patterns(lib: ILibraryView, name: str) -> str:
|
||||
"""
|
||||
The default `rename_theirs` function for `ILibrary.add`.
|
||||
|
||||
Treats names starting with `SINGLE_USE_PREFIX` (default: one underscore) as
|
||||
"one-offs" for which name conflicts should be automatically resolved.
|
||||
Conflicts are resolved by calling `lib.get_name(SINGLE_USE_PREFIX + stem)`
|
||||
where `stem = name.removeprefix(SINGLE_USE_PREFIX).split('$')[0]`.
|
||||
Names lacking the prefix are directly returned (not renamed).
|
||||
|
||||
Args:
|
||||
lib: The library into which `name` is to be added (but is presumed to conflict)
|
||||
name: The original name, to be modified
|
||||
|
||||
Returns:
|
||||
The new name, not guaranteed to be conflict-free!
|
||||
"""
|
||||
if not name.startswith(SINGLE_USE_PREFIX):
|
||||
return name
|
||||
|
||||
stem = name.removeprefix(SINGLE_USE_PREFIX).split('$')[0]
|
||||
return lib.get_name(SINGLE_USE_PREFIX + stem)
|
||||
|
||||
|
||||
def _plan_source_names(
|
||||
target: ILibraryView,
|
||||
source_order: Sequence[str],
|
||||
existing_names: Container[str],
|
||||
*,
|
||||
rename_theirs: Callable[[ILibraryView, str], str] | None = None,
|
||||
rename_when: Literal['conflict', 'always'] = 'conflict',
|
||||
) -> dict[str, str]:
|
||||
if rename_when not in ('conflict', 'always'):
|
||||
raise ValueError(f'Unknown source rename mode: {rename_when!r}')
|
||||
if rename_when == 'always' and rename_theirs is None:
|
||||
raise TypeError('rename_theirs is required when rename_when="always"')
|
||||
|
||||
source_to_visible: dict[str, str] = {}
|
||||
visible_names: set[str] = set()
|
||||
|
||||
for name in source_order:
|
||||
visible = name
|
||||
if rename_when == 'always':
|
||||
assert rename_theirs is not None
|
||||
visible = rename_theirs(target, name)
|
||||
elif visible in existing_names or visible in visible_names:
|
||||
if rename_theirs is None:
|
||||
raise LibraryError(f'Conflicting name while adding source: {name!r}')
|
||||
visible = rename_theirs(target, name)
|
||||
if visible in existing_names or visible in visible_names:
|
||||
raise LibraryError(f'Unresolved duplicate key encountered while adding source: {name!r} -> {visible!r}')
|
||||
source_to_visible[name] = visible
|
||||
visible_names.add(visible)
|
||||
|
||||
return source_to_visible
|
||||
|
||||
|
||||
def _source_rename_map(source_to_visible: Mapping[str, str]) -> dict[str, str]:
|
||||
return {
|
||||
source_name: visible_name
|
||||
for source_name, visible_name in source_to_visible.items()
|
||||
if source_name != visible_name
|
||||
}
|
||||
|
||||
def b64suffix(ii: int) -> str:
|
||||
"""
|
||||
Turn an integer into a base64-equivalent suffix.
|
||||
|
||||
This could be done with base64.b64encode, but this way is faster for many small `ii`.
|
||||
"""
|
||||
def i2a(nn: int) -> str:
|
||||
return 'ABCDEFGHIJKLMNOPQRSTUVWXYZabcdefghijklmnopqrstuvwxyz0123456789$?'[nn]
|
||||
|
||||
parts = ['$', i2a(ii % 64)]
|
||||
ii >>= 6
|
||||
while ii:
|
||||
parts.append(i2a(ii % 64))
|
||||
ii >>= 6
|
||||
return ''.join(parts)
|
||||
Loading…
Add table
Add a link
Reference in a new issue