[wip] introduce BuildLibrary and make overlay first-class
This commit is contained in:
parent
e8e19a5a9b
commit
dc828cf5fc
7 changed files with 1335 additions and 98 deletions
|
|
@ -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, cast, TypeAlias, Protocol, Literal
|
||||
from typing import Self, TYPE_CHECKING, Any, cast, TypeAlias, Protocol, Literal
|
||||
from collections.abc import Iterator, Mapping, MutableMapping, Sequence, Callable
|
||||
import logging
|
||||
import re
|
||||
|
|
@ -22,12 +22,14 @@ import copy
|
|||
from pprint import pformat
|
||||
from collections import defaultdict
|
||||
from abc import ABCMeta, abstractmethod
|
||||
from contextvars import ContextVar
|
||||
from dataclasses import dataclass, replace
|
||||
from graphlib import TopologicalSorter, CycleError
|
||||
|
||||
import numpy
|
||||
from numpy.typing import ArrayLike, NDArray
|
||||
|
||||
from .error import LibraryError, PatternError
|
||||
from .error import BuildError, LibraryError, PatternError
|
||||
from .utils import layer_t, apply_transforms
|
||||
from .shapes import Shape, Polygon
|
||||
from .label import Label
|
||||
|
|
@ -40,6 +42,11 @@ if TYPE_CHECKING:
|
|||
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
_ACTIVE_BUILD_SESSIONS: ContextVar[dict[int, '_BuildSessionLibrary'] | None] = ContextVar(
|
||||
'masque_active_build_sessions',
|
||||
default=None,
|
||||
)
|
||||
|
||||
|
||||
class visitor_function_t(Protocol):
|
||||
""" Signature for `Library.dfs()` visitor functions. """
|
||||
|
|
@ -62,6 +69,69 @@ Tree: TypeAlias = MutableMapping[str, 'Pattern']
|
|||
dangling_mode_t: TypeAlias = Literal['error', 'ignore', 'include']
|
||||
""" How helpers should handle refs whose targets are not present in the library. """
|
||||
|
||||
emitted_via_t: TypeAlias = Literal['declaration', 'helper_write', 'tree_merge', 'source_import']
|
||||
""" Build-provenance origin tags for emitted cells. """
|
||||
|
||||
|
||||
@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:
|
||||
final_name: Name exposed by the completed library.
|
||||
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`.
|
||||
emitted_via: High-level path by which the cell entered the output.
|
||||
build_chain: Declared-cell dependency chain that was active when the
|
||||
cell was emitted.
|
||||
renamed_from: Original requested name when the final name differs.
|
||||
source_name: Original on-source name for imported cells.
|
||||
source_metadata: Optional source-library metadata copied through from
|
||||
lazy GDS readers.
|
||||
"""
|
||||
final_name: str
|
||||
requested_name: str
|
||||
kind: Literal['declared', 'helper', 'source']
|
||||
owner_declared_name: str | None
|
||||
emitted_via: emitted_via_t
|
||||
build_chain: tuple[str, ...]
|
||||
renamed_from: str | None = None
|
||||
source_name: str | None = None
|
||||
source_metadata: dict[str, Any] | None = None
|
||||
|
||||
|
||||
@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.
|
||||
owned_cells: Mapping from declared cell name to all final output cell
|
||||
names it owns, including helper cells emitted while that declared
|
||||
cell was building.
|
||||
dependency_graph: Declared-cell dependency graph discovered through
|
||||
library-mediated reads and explicit recipe hints.
|
||||
"""
|
||||
requested_roots: tuple[str, ...]
|
||||
provenance: Mapping[str, CellProvenance]
|
||||
owned_cells: Mapping[str, tuple[str, ...]]
|
||||
dependency_graph: Mapping[str, frozenset[str]]
|
||||
|
||||
|
||||
SINGLE_USE_PREFIX = '_'
|
||||
"""
|
||||
|
|
@ -1397,6 +1467,819 @@ class Library(ILibrary):
|
|||
return tree, pat
|
||||
|
||||
|
||||
class BuiltLibrary(Library):
|
||||
"""
|
||||
Eager library returned by `BuildLibrary.build(output='library')`.
|
||||
|
||||
This is a normal materialized `Library` with one additional attribute,
|
||||
`build_report`, which records how the library was assembled from
|
||||
declarations, helper emissions, and imported source-backed cells.
|
||||
"""
|
||||
|
||||
def __init__(
|
||||
self,
|
||||
mapping: MutableMapping[str, 'Pattern'] | None = None,
|
||||
*,
|
||||
build_report: BuildReport | None = None,
|
||||
) -> None:
|
||||
super().__init__(mapping=mapping)
|
||||
self.build_report = build_report
|
||||
|
||||
|
||||
class _CellFactory:
|
||||
"""
|
||||
Adapter that turns a plain pattern factory into a deferred recipe factory.
|
||||
|
||||
Calling the wrapper captures arguments and returns a `_BuildRecipe`
|
||||
instead of executing the function immediately.
|
||||
"""
|
||||
|
||||
def __init__(self, func: Callable[..., 'Pattern']) -> None:
|
||||
self.func = func
|
||||
self.__name__ = getattr(func, '__name__', type(self).__name__)
|
||||
self.__doc__ = getattr(func, '__doc__')
|
||||
|
||||
def __call__(self, *args: Any, **kwargs: Any) -> '_BuildRecipe':
|
||||
return _BuildRecipe(func=self.func, args=args, kwargs=kwargs)
|
||||
|
||||
|
||||
@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
|
||||
|
||||
|
||||
@dataclass(frozen=True)
|
||||
class _PatternDeclaration:
|
||||
""" Declared cell backed by an already-built `Pattern`. """
|
||||
pattern: 'Pattern'
|
||||
|
||||
|
||||
@dataclass(frozen=True)
|
||||
class _RecipeDeclaration:
|
||||
""" Declared cell backed by a deferred recipe. """
|
||||
recipe: _BuildRecipe
|
||||
|
||||
|
||||
@dataclass(frozen=True)
|
||||
class _SourceDeclaration:
|
||||
"""
|
||||
Imported source-backed names registered with a `BuildLibrary`.
|
||||
|
||||
The declaration stores visible-name remapping plus pre-scanned graph
|
||||
metadata. Underlying source cells stay lazy until a build session
|
||||
materializes or copies them through.
|
||||
"""
|
||||
library: ILibraryView
|
||||
source_to_visible: Mapping[str, str]
|
||||
visible_to_source: Mapping[str, str]
|
||||
child_graph: Mapping[str, set[str]]
|
||||
order: tuple[str, ...]
|
||||
|
||||
|
||||
def cell(func: Callable[..., 'Pattern']) -> _CellFactory:
|
||||
"""
|
||||
Wrap a plain pattern factory so calls return deferred build recipes.
|
||||
|
||||
Use as either `cell(fn)(...)` or `@cell`.
|
||||
"""
|
||||
return _CellFactory(func)
|
||||
|
||||
|
||||
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 a
|
||||
normal library-like object carrying a `build_report`.
|
||||
"""
|
||||
|
||||
def __init__(self, *, check_on_register: bool = False) -> None:
|
||||
self.check_on_register = check_on_register
|
||||
self.cells = BuildCellsView(self)
|
||||
self.last_build_report: BuildReport | None = None
|
||||
self._frozen = False
|
||||
self._declarations: dict[str, _PatternDeclaration | _RecipeDeclaration] = {}
|
||||
self._sources: list[_SourceDeclaration] = []
|
||||
self._names: set[str] = set()
|
||||
self._order: list[str] = []
|
||||
|
||||
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._order)
|
||||
|
||||
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 | Callable[[], Pattern]',
|
||||
) -> 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!')
|
||||
|
||||
declaration: _PatternDeclaration | _RecipeDeclaration
|
||||
if isinstance(value, _BuildRecipe):
|
||||
declaration = _RecipeDeclaration(value)
|
||||
else:
|
||||
if callable(value):
|
||||
raise TypeError('BuildLibrary recipes must be wrapped with cell(fn)(...) or @cell.')
|
||||
declaration = _PatternDeclaration(value)
|
||||
|
||||
self._declarations[key] = declaration
|
||||
self._names.add(key)
|
||||
self._order.append(key)
|
||||
|
||||
if self.check_on_register:
|
||||
try:
|
||||
self.validate(names=(key,))
|
||||
except Exception:
|
||||
del self._declarations[key]
|
||||
self._names.remove(key)
|
||||
self._order.remove(key)
|
||||
raise
|
||||
|
||||
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]
|
||||
self._names.remove(key)
|
||||
self._order.remove(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]:
|
||||
session = self._active_session()
|
||||
if session is not None:
|
||||
return session.add(other, rename_theirs=rename_theirs, mutate_other=mutate_other)
|
||||
return super().add(other, rename_theirs=rename_theirs, mutate_other=mutate_other)
|
||||
|
||||
def rename(
|
||||
self,
|
||||
old_name: str,
|
||||
new_name: str,
|
||||
move_references: bool = False,
|
||||
) -> Self:
|
||||
"""
|
||||
Rename an imported source-backed visible name during authoring.
|
||||
|
||||
Only imported source-backed cells may be renamed on the builder itself.
|
||||
Declared/generated cells must be registered under their intended final
|
||||
names. `move_references=True` is intentionally unsupported here because
|
||||
deferred recipes and declaration internals cannot be rewritten safely.
|
||||
"""
|
||||
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.')
|
||||
if new_name in self._names:
|
||||
raise LibraryError(f'"{new_name}" already exists in the builder.')
|
||||
if move_references:
|
||||
raise BuildError(
|
||||
'BuildLibrary.rename(..., move_references=True) is not supported for imported source cells. '
|
||||
'Builder-level renames only change the visible imported name.'
|
||||
)
|
||||
|
||||
source_index = next(
|
||||
(idx for idx, spec in enumerate(self._sources) if old_name in spec.visible_to_source),
|
||||
None,
|
||||
)
|
||||
if source_index is None:
|
||||
raise BuildError(
|
||||
f'Cannot rename "{old_name}" during authoring because only imported source-backed '
|
||||
'cells may be renamed on a BuildLibrary.'
|
||||
)
|
||||
|
||||
spec = self._sources[source_index]
|
||||
source_name = spec.visible_to_source[old_name]
|
||||
source_to_visible = dict(spec.source_to_visible)
|
||||
visible_to_source = dict(spec.visible_to_source)
|
||||
order = list(spec.order)
|
||||
|
||||
source_to_visible[source_name] = new_name
|
||||
del visible_to_source[old_name]
|
||||
visible_to_source[new_name] = source_name
|
||||
order[order.index(old_name)] = new_name
|
||||
|
||||
self._sources[source_index] = replace(
|
||||
spec,
|
||||
source_to_visible=source_to_visible,
|
||||
visible_to_source=visible_to_source,
|
||||
order=tuple(order),
|
||||
)
|
||||
self._names.remove(old_name)
|
||||
self._names.add(new_name)
|
||||
self._order[self._order.index(old_name)] = new_name
|
||||
return self
|
||||
|
||||
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,
|
||||
) -> dict[str, str]:
|
||||
"""
|
||||
Register an imported source-backed library with the builder.
|
||||
|
||||
The source is not materialized immediately. Instead, its names and
|
||||
child graph are scanned once and stored as an import declaration. The
|
||||
source may be renamed on entry to avoid collisions with existing
|
||||
declarations or other imported sources.
|
||||
|
||||
Returns:
|
||||
Mapping of `{source_name: visible_name}` for imported names that
|
||||
were renamed while being added.
|
||||
"""
|
||||
self._assert_editable()
|
||||
|
||||
view = source if isinstance(source, ILibraryView) else LibraryView(source)
|
||||
source_order = tuple(view.source_order())
|
||||
child_graph = view.child_graph(dangling='include')
|
||||
|
||||
source_to_visible: dict[str, str] = {}
|
||||
visible_to_source: dict[str, str] = {}
|
||||
rename_map: dict[str, str] = {}
|
||||
new_names: list[str] = []
|
||||
|
||||
for name in source_order:
|
||||
visible = name
|
||||
if visible in self._names or visible in visible_to_source:
|
||||
if rename_theirs is None:
|
||||
raise LibraryError(f'Conflicting name while adding source: {name!r}')
|
||||
visible = rename_theirs(self, name)
|
||||
if visible in self._names or visible in visible_to_source:
|
||||
raise LibraryError(f'Unresolved duplicate key encountered while adding source: {name!r} -> {visible!r}')
|
||||
rename_map[name] = visible
|
||||
source_to_visible[name] = visible
|
||||
visible_to_source[visible] = name
|
||||
new_names.append(visible)
|
||||
|
||||
self._sources.append(_SourceDeclaration(
|
||||
library=view,
|
||||
source_to_visible=dict(source_to_visible),
|
||||
visible_to_source=dict(visible_to_source),
|
||||
child_graph={name: set(children) for name, children in child_graph.items()},
|
||||
order=tuple(source_to_visible[name] for name in source_order),
|
||||
))
|
||||
for visible in new_names:
|
||||
self._names.add(visible)
|
||||
self._order.append(visible)
|
||||
return rename_map
|
||||
|
||||
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.
|
||||
"""
|
||||
report, _output = self._run_build(names=names, output='overlay', allow_dangling=allow_dangling, persist_output=False)
|
||||
self.last_build_report = report
|
||||
return report
|
||||
|
||||
def build(
|
||||
self,
|
||||
*,
|
||||
output: Literal['overlay', 'library'] = 'overlay',
|
||||
allow_dangling: bool = False,
|
||||
) -> 'BuiltLibrary | ILibrary':
|
||||
"""
|
||||
Materialize declarations and return a usable output library.
|
||||
|
||||
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.
|
||||
"""
|
||||
self._assert_editable()
|
||||
report, built_output = self._run_build(names=None, output=output, allow_dangling=allow_dangling, persist_output=True)
|
||||
self._frozen = True
|
||||
self.last_build_report = report
|
||||
return built_output
|
||||
|
||||
def _run_build(
|
||||
self,
|
||||
*,
|
||||
names: Sequence[str] | None,
|
||||
output: Literal['overlay', 'library'],
|
||||
allow_dangling: bool,
|
||||
persist_output: bool,
|
||||
) -> tuple[BuildReport, BuiltLibrary | ILibrary | None]:
|
||||
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')
|
||||
if output == 'library':
|
||||
built_output = session.to_library() if persist_output else None
|
||||
elif persist_output:
|
||||
built_output = session.to_overlay()
|
||||
else:
|
||||
built_output = None
|
||||
finally:
|
||||
_ACTIVE_BUILD_SESSIONS.reset(token)
|
||||
|
||||
report = session.build_report(roots)
|
||||
if built_output is not None:
|
||||
built_output.build_report = report
|
||||
return report, built_output
|
||||
|
||||
|
||||
class _BuildSessionLibrary(ILibrary):
|
||||
"""
|
||||
Internal overlay-backed library used while a `BuildLibrary` is executing.
|
||||
|
||||
This object provides the mutable-library surface that recipes expect while
|
||||
also tracking declared-cell dependencies, helper-cell provenance, and
|
||||
imported source cells. It exists only for the duration of a validation or
|
||||
build run.
|
||||
"""
|
||||
|
||||
def __init__(self, builder: BuildLibrary) -> None:
|
||||
from .file.gdsii_lazy_core import BuiltOverlayLibrary, _SourceEntry, _SourceLayer # noqa: PLC0415
|
||||
|
||||
self._builder = builder
|
||||
self._overlay = BuiltOverlayLibrary()
|
||||
self._source_entry_type = _SourceEntry
|
||||
self._source_layer_type = _SourceLayer
|
||||
self._states: dict[str, Literal['unbuilt', 'building', 'built']] = {
|
||||
name: 'unbuilt' for name in builder._declarations
|
||||
}
|
||||
self._declared_stack: list[str] = []
|
||||
self._emission_stack: list[str] = []
|
||||
self._emission_via_stack: list[emitted_via_t] = []
|
||||
self._names = set(builder._names)
|
||||
self._order = list(builder._order)
|
||||
self._provenance: dict[str, CellProvenance] = {}
|
||||
self._owned_cells: defaultdict[str, list[str]] = defaultdict(list)
|
||||
self._dependency_graph: defaultdict[str, set[str]] = defaultdict(set)
|
||||
self._install_sources()
|
||||
|
||||
def _install_sources(self) -> None:
|
||||
for spec in self._builder._sources:
|
||||
layer = self._source_layer_type(
|
||||
library=spec.library,
|
||||
source_to_visible=dict(spec.source_to_visible),
|
||||
visible_to_source=dict(spec.visible_to_source),
|
||||
child_graph={name: set(children) for name, children in spec.child_graph.items()},
|
||||
order=list(spec.order),
|
||||
)
|
||||
layer_index = len(self._overlay._layers)
|
||||
self._overlay._layers.append(layer)
|
||||
source_info = getattr(spec.library, 'library_info', None)
|
||||
source_meta = dict(source_info) if isinstance(source_info, dict) else None
|
||||
|
||||
for source_name, visible_name in spec.source_to_visible.items():
|
||||
self._overlay._entries[visible_name] = self._source_entry_type(
|
||||
layer_index=layer_index,
|
||||
source_name=source_name,
|
||||
)
|
||||
if visible_name not in self._overlay._order:
|
||||
self._overlay._order.append(visible_name)
|
||||
self._provenance[visible_name] = CellProvenance(
|
||||
final_name=visible_name,
|
||||
requested_name=source_name,
|
||||
kind='source',
|
||||
owner_declared_name=None,
|
||||
emitted_via='source_import',
|
||||
build_chain=(),
|
||||
renamed_from=source_name if visible_name != source_name else None,
|
||||
source_name=source_name,
|
||||
source_metadata=source_meta,
|
||||
)
|
||||
|
||||
def __iter__(self) -> Iterator[str]:
|
||||
return (name for name in self._order if name in self._names)
|
||||
|
||||
def __len__(self) -> int:
|
||||
return len(self._names)
|
||||
|
||||
def __contains__(self, key: object) -> bool:
|
||||
return key in self._names or key in self._overlay
|
||||
|
||||
def _touch_name(self, key: str) -> None:
|
||||
if key not in self._names:
|
||||
self._names.add(key)
|
||||
self._order.append(key)
|
||||
|
||||
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 _remove_owned_cell(self, owner: str | None, name: str) -> None:
|
||||
if owner is None or owner not in self._owned_cells:
|
||||
return
|
||||
cells = self._owned_cells[owner]
|
||||
self._owned_cells[owner] = [cell for cell in cells if cell != name]
|
||||
if not self._owned_cells[owner]:
|
||||
del self._owned_cells[owner]
|
||||
|
||||
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.discard(old_name)
|
||||
self._names.add(new_name)
|
||||
if old_name in self._order:
|
||||
idx = self._order.index(old_name)
|
||||
self._order[idx] = new_name
|
||||
|
||||
provenance = self._provenance.pop(old_name)
|
||||
requested_name = provenance.requested_name
|
||||
self._provenance[new_name] = replace(
|
||||
provenance,
|
||||
final_name=new_name,
|
||||
renamed_from=requested_name if new_name != requested_name else None,
|
||||
)
|
||||
|
||||
owner = provenance.owner_declared_name
|
||||
if owner is not None and owner in self._owned_cells:
|
||||
self._owned_cells[owner] = [
|
||||
new_name if cell_name == old_name else cell_name
|
||||
for cell_name in self._owned_cells[owner]
|
||||
]
|
||||
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._touch_name(key)
|
||||
|
||||
kind: Literal['declared', 'helper']
|
||||
via = self._emission_via_stack[-1] if self._emission_via_stack else 'helper_write'
|
||||
if current is not None and key == current:
|
||||
kind = 'declared'
|
||||
via = 'declaration'
|
||||
else:
|
||||
kind = 'helper'
|
||||
if not self._emission_via_stack:
|
||||
via = 'helper_write'
|
||||
|
||||
self._emission_stack.append(key)
|
||||
try:
|
||||
self._record_provenance(
|
||||
final_name=key,
|
||||
requested_name=key,
|
||||
kind=kind,
|
||||
owner_declared_name=current if kind == 'helper' else key,
|
||||
emitted_via=via,
|
||||
build_chain=tuple(self._declared_stack),
|
||||
renamed_from=None,
|
||||
)
|
||||
finally:
|
||||
self._emission_stack.pop()
|
||||
|
||||
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')
|
||||
provenance = self._provenance.get(key)
|
||||
if key in self._overlay:
|
||||
del self._overlay[key]
|
||||
self._names.discard(key)
|
||||
if key in self._order:
|
||||
self._order.remove(key)
|
||||
self._provenance.pop(key, None)
|
||||
if provenance is not None:
|
||||
self._remove_owned_cell(provenance.owner_declared_name, 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(
|
||||
self,
|
||||
other: Mapping[str, 'Pattern'],
|
||||
rename_theirs: Callable[['ILibraryView', str], str] = _rename_patterns,
|
||||
mutate_other: bool = False,
|
||||
) -> dict[str, str]:
|
||||
self._emission_via_stack.append('tree_merge')
|
||||
try:
|
||||
rename_map = super().add(other, rename_theirs=rename_theirs, mutate_other=mutate_other)
|
||||
finally:
|
||||
self._emission_via_stack.pop()
|
||||
|
||||
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,
|
||||
renamed_from=old_name,
|
||||
owner_declared_name=current if current is not None else self._provenance[new_name].owner_declared_name,
|
||||
)
|
||||
return rename_map
|
||||
|
||||
def _record_provenance(
|
||||
self,
|
||||
*,
|
||||
final_name: str,
|
||||
requested_name: str,
|
||||
kind: Literal['declared', 'helper'],
|
||||
owner_declared_name: str | None,
|
||||
emitted_via: emitted_via_t,
|
||||
build_chain: tuple[str, ...],
|
||||
renamed_from: str | None,
|
||||
) -> None:
|
||||
self._provenance[final_name] = CellProvenance(
|
||||
final_name=final_name,
|
||||
requested_name=requested_name,
|
||||
kind=kind,
|
||||
owner_declared_name=owner_declared_name,
|
||||
emitted_via=emitted_via,
|
||||
build_chain=build_chain,
|
||||
renamed_from=renamed_from,
|
||||
)
|
||||
if owner_declared_name is not None and final_name not in self._owned_cells[owner_declared_name]:
|
||||
self._owned_cells[owner_declared_name].append(final_name)
|
||||
|
||||
def _wrap_error(self, name: str, exc: Exception) -> BuildError:
|
||||
helper = self._emission_stack[-1] if self._emission_stack else None
|
||||
chain = tuple(self._declared_stack)
|
||||
msg = [f'Failed while building declared cell "{name}"']
|
||||
if helper is not None and helper != name:
|
||||
msg.append(f'while materializing helper/output "{helper}"')
|
||||
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
|
||||
|
||||
state = self._states[name]
|
||||
if state == 'built':
|
||||
return
|
||||
if state == 'building':
|
||||
chain = ' -> '.join(self._declared_stack + [name])
|
||||
raise BuildError(f'Cycle detected while building declared cells: {chain}')
|
||||
|
||||
declaration = self._builder._declarations[name]
|
||||
self._states[name] = 'building'
|
||||
self._declared_stack.append(name)
|
||||
try:
|
||||
if isinstance(declaration, _PatternDeclaration):
|
||||
pattern = declaration.pattern.deepcopy()
|
||||
else:
|
||||
for dep in declaration.recipe.explicit_dependencies:
|
||||
self._ensure_named(dep)
|
||||
pattern = declaration.recipe.func(*declaration.recipe.args, **declaration.recipe.kwargs)
|
||||
if not isinstance(pattern, Pattern):
|
||||
raise BuildError(f'Recipe for "{name}" returned {type(pattern).__name__}, expected Pattern')
|
||||
|
||||
if name in self._overlay:
|
||||
if self._overlay[name] is not pattern:
|
||||
raise BuildError(
|
||||
f'Recipe for "{name}" wrote a different pattern into the session under its own name.'
|
||||
)
|
||||
else:
|
||||
self[name] = pattern
|
||||
self._states[name] = 'built'
|
||||
except Exception as exc:
|
||||
self._states[name] = 'unbuilt'
|
||||
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
|
||||
}
|
||||
owned_cells = {
|
||||
name: tuple(cells)
|
||||
for name, cells in self._owned_cells.items()
|
||||
}
|
||||
return BuildReport(
|
||||
requested_roots=tuple(dict.fromkeys(requested_roots)),
|
||||
provenance=dict(self._provenance),
|
||||
owned_cells=owned_cells,
|
||||
dependency_graph=dependency_graph,
|
||||
)
|
||||
|
||||
def to_overlay(self) -> ILibrary:
|
||||
return self._overlay
|
||||
|
||||
def to_library(self) -> BuiltLibrary:
|
||||
mapping = {name: self._overlay[name] for name in self._overlay.source_order()}
|
||||
return BuiltLibrary(mapping)
|
||||
|
||||
|
||||
class LazyLibrary(ILibrary):
|
||||
"""
|
||||
This class is usually used to create a library of Patterns by mapping names to
|
||||
|
|
|
|||
Loading…
Add table
Add a link
Reference in a new issue