jan/masque
1
1
Fork 0
Code Issues Pull requests Releases Wiki Activity

[BuildLibrary / OverlayLibrary] further simplifications

Browse source
This commit is contained in:
Jan Petykiewicz 2026-06-19 23:11:23 -07:00
commit 1723212424
3 changed files with 105 additions and 133 deletions
Download patch file Download diff file Expand all files Collapse all files

38
masque/file/gdsii_lazy_core.py
View file

@ -32,7 +32,7 @@ from numpy.typing import NDArray
from . import gdsii
from .utils import tmpfile
from ..error import LibraryError
from ..library import ILibrary, ILibraryView, LibraryView, dangling_mode_t
from ..library import ILibrary, ILibraryView, LibraryView, _plan_source_names, dangling_mode_t
from ..pattern import Pattern, map_targets
from ..utils import apply_transforms
from ..utils.ports2data import data_to_ports
@ -315,38 +315,18 @@ class OverlayLibrary(ILibrary):
If `'always'`, every imported source name is passed through
`rename_theirs`.
"""
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"')
view = _coerce_library_view(source)
source_order = list(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] = {}
for name in source_order:
visible = name
renamed = False
if rename_when == 'always':
visible = cast('Callable[[ILibraryView, str], str]', rename_theirs)(self, name)
renamed = True
elif visible in self._entries 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)
renamed = True
if visible in self._entries or visible in visible_to_source:
if not renamed:
raise LibraryError(f'Conflicting name while adding source: {name!r}')
raise LibraryError(f'Unresolved duplicate key encountered while adding source: {name!r} -> {visible!r}')
if visible != name:
rename_map[name] = visible
source_to_visible[name] = visible
visible_to_source[visible] = name
source_to_visible, rename_map = _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,

171
masque/library.py
View file

@ -15,10 +15,11 @@ Classes include:
library. Generated with `ILibraryView.abstract_view()`.
"""
from typing import Self, TYPE_CHECKING, Any, cast, TypeAlias, Protocol, Literal
from collections.abc import Iterator, Mapping, MutableMapping, Sequence, Callable
from collections.abc import Container, Iterator, Mapping, MutableMapping, Sequence, Callable
import logging
import re
import copy
from functools import wraps
from pprint import pformat
from collections import defaultdict
from abc import ABCMeta, abstractmethod
@ -87,15 +88,11 @@ class CellProvenance:
any. Imported source cells leave this as `None`.
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.
"""
requested_name: str
kind: Literal['declared', 'helper', 'source']
owner_declared_name: str | None
build_chain: tuple[str, ...]
renamed_from: str | None = None
source_name: str | None = None
@dataclass(frozen=True)
@ -152,6 +149,47 @@ def _rename_patterns(lib: 'ILibraryView', name: str) -> str:
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',
) -> tuple[dict[str, str], 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()
rename_map: dict[str, str] = {}
for name in source_order:
visible = name
renamed = False
if rename_when == 'always':
assert rename_theirs is not None
visible = rename_theirs(target, name)
renamed = True
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)
renamed = True
if visible in existing_names or visible in visible_names:
if not renamed:
raise LibraryError(f'Conflicting name while adding source: {name!r}')
raise LibraryError(f'Unresolved duplicate key encountered while adding source: {name!r} -> {visible!r}')
if visible != name:
rename_map[name] = visible
source_to_visible[name] = visible
visible_names.add(visible)
return source_to_visible, rename_map
class ILibraryView(Mapping[str, 'Pattern'], metaclass=ABCMeta):
"""
Interface for a read-only library.
@ -1453,23 +1491,6 @@ class Library(ILibrary):
return tree, pat
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. """
@ -1483,25 +1504,17 @@ class _BuildRecipe:
return self
@dataclass(frozen=True)
class _SourceDeclaration:
"""
Imported source-backed names registered with a `BuildLibrary`.
The declaration stores visible-name remapping. Underlying source cells stay
lazy until a build session materializes or copies them through.
"""
library: ILibraryView
source_to_visible: Mapping[str, str]
def cell(func: Callable[..., 'Pattern']) -> _CellFactory:
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`.
"""
return _CellFactory(func)
@wraps(func)
def wrapper(*args: Any, **kwargs: Any) -> _BuildRecipe:
return _BuildRecipe(func=func, args=args, kwargs=kwargs)
return wrapper
class BuildCellsView:
@ -1555,7 +1568,7 @@ class BuildLibrary(ILibrary):
self.cells = BuildCellsView(self)
self._frozen = False
self._declarations: dict[str, 'Pattern | _BuildRecipe'] = {}
self._sources: list[_SourceDeclaration] = []
self._sources: list[tuple[ILibraryView, dict[str, str]]] = []
self._names: set[str] = set()
self._order: list[str] = []
@ -1694,8 +1707,8 @@ class BuildLibrary(ILibrary):
source_index: int | None = None
source_name: str | None = None
for idx, spec in enumerate(self._sources):
for candidate_source, candidate_visible in spec.source_to_visible.items():
for idx, (_source, source_to_visible) in enumerate(self._sources):
for candidate_source, candidate_visible in source_to_visible.items():
if candidate_visible == old_name:
source_index = idx
source_name = candidate_source
@ -1708,16 +1721,13 @@ class BuildLibrary(ILibrary):
'cells may be renamed on a BuildLibrary.'
)
spec = self._sources[source_index]
source_to_visible = dict(spec.source_to_visible)
source_library, source_to_visible = self._sources[source_index]
source_to_visible = dict(source_to_visible)
assert source_name is not None
source_to_visible[source_name] = new_name
self._sources[source_index] = replace(
spec,
source_to_visible = source_to_visible,
)
self._sources[source_index] = (source_library, source_to_visible)
self._names.remove(old_name)
self._names.add(new_name)
self._order[self._order.index(old_name)] = new_name
@ -1761,48 +1771,23 @@ class BuildLibrary(ILibrary):
Mapping of `{source_name: visible_name}` for imported names that
were renamed while being added.
"""
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"')
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, rename_map = _plan_source_names(
self,
source_order,
self._names,
rename_theirs = rename_theirs,
rename_when = rename_when,
)
source_to_visible: dict[str, str] = {}
visible_names: set[str] = set()
rename_map: dict[str, str] = {}
new_names: list[str] = []
for name in source_order:
visible = name
renamed = False
if rename_when == 'always':
visible = cast('Callable[[ILibraryView, str], str]', rename_theirs)(self, name)
renamed = True
elif visible in self._names or visible in visible_names:
if rename_theirs is None:
raise LibraryError(f'Conflicting name while adding source: {name!r}')
visible = rename_theirs(self, name)
renamed = True
if visible in self._names or visible in visible_names:
if not renamed:
raise LibraryError(f'Conflicting name while adding source: {name!r}')
raise LibraryError(f'Unresolved duplicate key encountered while adding source: {name!r} -> {visible!r}')
if visible != name:
rename_map[name] = visible
source_to_visible[name] = visible
visible_names.add(visible)
new_names.append(visible)
self._sources.append(_SourceDeclaration(
library = view,
source_to_visible = dict(source_to_visible),
))
for visible in new_names:
self._sources.append((view, dict(source_to_visible)))
for source_name in source_order:
visible = source_to_visible[source_name]
self._names.add(visible)
self._order.append(visible)
return rename_map
@ -1902,9 +1887,9 @@ class _BuildSessionLibrary(ILibrary):
self._install_sources()
def _install_sources(self) -> None:
for spec in self._builder._sources:
source_order = spec.library.source_order()
expected_names = set(spec.source_to_visible)
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)
@ -1920,24 +1905,22 @@ class _BuildSessionLibrary(ILibrary):
'Do not structurally mutate source libraries between add_source() and build()/validate().'
)
def rename_source(_lib: ILibraryView, name: str, *, mapping: Mapping[str, str] = spec.source_to_visible) -> str:
def rename_source(_lib: ILibraryView, name: str, *, mapping: Mapping[str, str] = source_to_visible) -> str:
return mapping[name]
self._overlay.add_source(
spec.library,
source_library,
rename_theirs = rename_source,
rename_when = 'always',
)
for source_name in source_order:
visible_name = spec.source_to_visible[source_name]
visible_name = source_to_visible[source_name]
self._provenance[visible_name] = CellProvenance(
requested_name = source_name,
kind = 'source',
owner_declared_name = None,
build_chain = (),
renamed_from = source_name if visible_name != source_name else None,
source_name = source_name,
)
def __iter__(self) -> Iterator[str]:
@ -1947,7 +1930,7 @@ class _BuildSessionLibrary(ILibrary):
return len(self._names)
def __contains__(self, key: object) -> bool:
return key in self._names or key in self._overlay
return key in self._names
def _touch_name(self, key: str) -> None:
if key not in self._names:
@ -1998,11 +1981,7 @@ class _BuildSessionLibrary(ILibrary):
self._order[idx] = new_name
provenance = self._provenance.pop(old_name)
requested_name = provenance.requested_name
self._provenance[new_name] = replace(
provenance,
renamed_from = requested_name if new_name != requested_name else None,
)
self._provenance[new_name] = provenance
return self
def __getitem__(self, key: str) -> 'Pattern':
@ -2038,7 +2017,6 @@ class _BuildSessionLibrary(ILibrary):
kind = kind,
owner_declared_name = current if kind == 'helper' else key,
build_chain = tuple(self._declared_stack),
renamed_from = None,
)
def __delitem__(self, key: str) -> None:
@ -2072,7 +2050,6 @@ class _BuildSessionLibrary(ILibrary):
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
@ -2085,14 +2062,12 @@ class _BuildSessionLibrary(ILibrary):
kind: Literal['declared', 'helper'],
owner_declared_name: str | None,
build_chain: tuple[str, ...],
renamed_from: str | None,
) -> None:
self._provenance[name] = CellProvenance(
requested_name = requested_name,
kind = kind,
owner_declared_name = owner_declared_name,
build_chain = build_chain,
renamed_from = renamed_from,
)
def _wrap_error(self, name: str, exc: Exception) -> BuildError:

29
masque/test/test_build_library.py
View file

@ -50,13 +50,13 @@ def test_build_library_tracks_helper_provenance_and_tree_merge_renames() -> None
_built, report = builder.build()
helpers = [
prov for prov in report.provenance.values()
(name, prov) for name, prov in report.provenance.items()
if prov.owner_declared_name == "top" and prov.kind == "helper"
]
assert "top" in _owned_by(report, "top")
assert len(helpers) == 2
assert any(prov.renamed_from == "_helper" for prov in helpers)
assert any(name != prov.requested_name for name, prov in helpers)
def test_build_library_requires_build_session_for_reads_and_freezes_after_build() -> None:
@ -152,6 +152,25 @@ def test_build_library_allows_helper_writes_via_pather() -> None:
assert helper_prov.owner_declared_name == "top"
def test_build_library_contains_tracks_active_session_names() -> None:
builder = BuildLibrary()
builder["leaf"] = Pattern()
builder.add_source(Library({"src": Pattern()}))
def make_top(lib: BuildLibrary) -> Pattern:
assert "leaf" in lib
assert "src" in lib
assert "_helper" not in lib
lib["_helper"] = Pattern()
assert "_helper" in lib
return Pattern()
builder.cells.top = cell(make_top)(builder)
built, _report = builder.build()
assert "_helper" in built
def test_build_library_preserves_source_cells_and_records_source_provenance() -> None:
source = Library({"src": Pattern()})
builder = BuildLibrary()
@ -184,7 +203,7 @@ def test_build_library_add_source_can_rename_every_source_cell() -> None:
"parent": "mapped_parent",
}
assert "mapped_child" in built["mapped_parent"].refs
assert report.provenance["mapped_child"].source_name == "child"
assert report.provenance["mapped_child"].requested_name == "child"
def test_build_library_rejects_source_cells_added_after_add_source() -> None:
@ -236,7 +255,7 @@ def test_build_library_can_rename_imported_source_cells_during_authoring() -> No
assert "renamed_child" in built
assert "child" not in built
assert "renamed_child" in built["parent"].refs
assert report.provenance["renamed_child"].source_name == "child"
assert report.provenance["renamed_child"].requested_name == "child"
def test_build_library_rejects_move_references_for_source_rename() -> None:
@ -276,7 +295,6 @@ def test_build_library_helper_rename_updates_provenance_owner() -> None:
prov = report.provenance["final_helper"]
assert prov.kind == "helper"
assert prov.requested_name == "_helper"
assert prov.renamed_from == "_helper"
def test_build_library_helper_delete_removes_provenance_and_ownership() -> None:
@ -314,7 +332,6 @@ def test_build_library_helper_rename_after_auto_rename_preserves_requested_name(
assert "final_helper" in built
prov = report.provenance["final_helper"]
assert prov.requested_name == "_helper"
assert prov.renamed_from == "_helper"
def test_build_library_rejects_renaming_declared_or_source_cells_during_build() -> None: