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

[file.utils] add preflight_source_aware

Browse source
This commit is contained in:
Jan Petykiewicz 2026-07-14 19:18:46 -07:00
commit 95e54acf96
5 changed files with 298 additions and 35 deletions
Download patch file Download diff file Expand all files Collapse all files

10
MIGRATION.md
View file

@ -617,6 +617,16 @@ serialization of every cell. Set `copy_through=True` only when source-aware
writers may copy untouched cells unchanged and intentionally skip their layer
mapping; persistently accessed cells are mapped and no longer copied through.
`preflight_source_aware(...)` preserves that per-cell provenance. It returns a
borrowing `OverlayLibrary`, skips source-backed cells completely, and applies
pattern sorting, named-layer validation, safe empty-cell pruning, and
repeated-shape wrapping only to cells without reusable source provenance.
Library name order is retained because sorting source-backed cells would
require loading them. Always use the returned overlay for writing, and keep the
original lazy source open until the write finishes. The comprehensive
`preflight(...)` operation continues to materialize every cell when sorting is
enabled.
Generic borrowing views no longer expose GDS-specific `raw_struct_bytes()`,
`can_copy_raw_struct()`, or forwarded `library_info` attributes. Keep the
owning GDS source or the metadata returned by `readfile()` when direct access is

228
masque/file/utils.py
View file

@ -14,12 +14,188 @@ from pprint import pformat
from itertools import chain
from .. import Pattern, PatternError, Library, LibraryError
from ..library import (
IBorrowing, ILibraryView, OverlayLibrary, SINGLE_USE_PREFIX,
dangling_mode_t,
)
from ..shapes import Polygon, Path
logger = logging.getLogger(__name__)
def _has_source_provenance(library: ILibraryView, name: str) -> bool:
"""Return whether `name` has an uninterrupted borrowing provenance chain."""
current = library
current_name = name
seen: set[tuple[int, str]] = set()
followed_source = False
while isinstance(current, IBorrowing):
key = (id(current), current_name)
if key in seen:
return False
seen.add(key)
source_cell = current.source_cell(current_name)
if source_cell is None:
return False
followed_source = True
current, current_name = source_cell
return followed_source
def _prune_checked_empty(
library: OverlayLibrary,
checked_names: set[str],
*,
dangling: dangling_mode_t,
) -> set[str]:
"""Prune checked empty cells without modifying source-backed parents."""
parent_graph = library.parent_graph(dangling=dangling)
source_backed = set(library) - checked_names
def safely_empty(name: str) -> bool:
return (
name in library
and name in checked_names
and not (parent_graph.get(name, set()) & source_backed)
and library[name].is_empty()
)
empty = {name for name in checked_names if safely_empty(name)}
pruned: set[str] = set()
while empty:
parents: set[str] = set()
for name in empty:
name_parents = parent_graph.get(name, set())
del library[name]
checked_names.discard(name)
for parent in name_parents & checked_names:
if parent in library and name in library[parent].refs:
del library[parent].refs[name]
parents |= name_parents
pruned |= empty
empty = {parent for parent in parents if safely_empty(parent)}
return pruned
def _wrap_checked_repeated_shapes(
library: OverlayLibrary,
checked_names: set[str],
) -> None:
"""Wrap repetitions in checked cells while leaving source-backed cells untouched."""
for pattern_name in tuple(checked_names):
if pattern_name not in library:
continue
pattern = library[pattern_name]
for layer in pattern.shapes:
new_shapes = []
for shape in pattern.shapes[layer]:
if shape.repetition is None:
new_shapes.append(shape)
continue
name = library.get_name(SINGLE_USE_PREFIX + 'rep')
library[name] = Pattern(shapes={layer: [shape]})
checked_names.add(name)
pattern.ref(name, repetition=shape.repetition)
shape.repetition = None
pattern.shapes[layer] = new_shapes
for layer in pattern.labels:
new_labels = []
for label in pattern.labels[layer]:
if label.repetition is None:
new_labels.append(label)
continue
name = library.get_name(SINGLE_USE_PREFIX + 'rep')
library[name] = Pattern(labels={layer: [label]})
checked_names.add(name)
pattern.ref(name, repetition=label.repetition)
label.repetition = None
pattern.labels[layer] = new_labels
def preflight_source_aware(
lib: ILibraryView,
sort: bool = True,
sort_elements: bool = False,
allow_dangling_refs: bool | None = None,
allow_named_layers: bool = True,
prune_empty_patterns: bool = False,
wrap_repeated_shapes: bool = False,
) -> OverlayLibrary:
"""
Preflight cells without reusable source provenance.
Returns a borrowing `OverlayLibrary`. Cells with uninterrupted
`IBorrowing.source_cell()` provenance remain source-backed and receive no
per-pattern checks. Other cells are detached into the overlay and checked.
Keep `lib` and its borrowed sources open for the result's lifetime.
Args:
sort: Whether to sort checked pattern contents. Library name order is
retained because sorting source-backed cells would require loading them.
sort_elements: Whether to sort elements within checked patterns.
allow_dangling_refs: If `None` (default), warns about any refs to patterns that are not
in the provided library. If `True`, no check is performed; if `False`, a `LibraryError`
is raised instead.
allow_named_layers: If `False`, raises a `PatternError` if any layer is referred to by
a string in a checked pattern instead of a number (or tuple).
prune_empty_patterns: Recursively delete checked empty patterns when
doing so does not require modifying a source-backed parent.
wrap_repeated_shapes: Turn repeated shapes in checked patterns into
repeated refs containing non-repeated shapes.
Returns:
A borrowing overlay containing checked patterns and source-backed cells.
"""
checked_names = {
name
for name in lib
if not _has_source_provenance(lib, name)
}
overlay = OverlayLibrary()
overlay.add_source(lib)
if sort:
for name in sorted(checked_names):
overlay[name].sort(sort_elements=sort_elements)
if not allow_dangling_refs:
refs = overlay.referenced_patterns()
dangling = refs - set(overlay.keys())
if dangling:
msg = 'Dangling refs found: ' + pformat(dangling)
if allow_dangling_refs is None:
logger.warning(msg)
else:
raise LibraryError(msg)
if not allow_named_layers:
checked_named_layers: Mapping[str, set] = defaultdict(set)
for name in checked_names:
pattern = overlay[name]
for layer in chain(pattern.shapes.keys(), pattern.labels.keys()):
if isinstance(layer, str):
checked_named_layers[name].add(layer)
checked_named_layers = dict(checked_named_layers)
if checked_named_layers:
raise PatternError('Non-numeric layers found:' + pformat(checked_named_layers))
if prune_empty_patterns:
prune_dangling: dangling_mode_t = 'error' if allow_dangling_refs is False else 'ignore'
pruned = _prune_checked_empty(overlay, checked_names, dangling=prune_dangling)
if pruned:
logger.info(f'Preflight pruned {len(pruned)} checked empty patterns')
logger.debug('Pruned: ' + pformat(pruned))
else:
logger.debug('Preflight found no safely prunable checked patterns')
if wrap_repeated_shapes:
_wrap_checked_repeated_shapes(overlay, checked_names)
return overlay
def preflight(
lib: Library,
sort: bool = True,
@ -30,39 +206,35 @@ def preflight(
wrap_repeated_shapes: bool = False,
) -> Library:
"""
Run a standard set of useful operations and checks, usually done immediately prior
to writing to a file (or immediately after reading).
Run a standard set of useful operations and checks on an entire library.
Note that this helper is not copy-isolating. When `sort=True`, it constructs a new
`Library` wrapper around the same `Pattern` objects after sorting them in place, so
later mutating preflight steps such as `prune_empty_patterns` and
`wrap_repeated_shapes` may still mutate caller-owned patterns. Callers that need
isolation should deep-copy the library before calling `preflight()`.
This helper is not copy-isolating. When `sort=True`, it constructs a new
`Library` wrapper around the same `Pattern` objects after sorting them in
place. Later mutating steps may still mutate caller-owned patterns. Deep-copy
the library first when isolation is required.
Args:
sort: Whether to sort the patterns based on their names, and optionaly sort the pattern contents.
Default True. Useful for reproducible builds.
sort_elements: Whether to sort the pattern contents. Requires sort=True to run.
allow_dangling_refs: If `None` (default), warns about any refs to patterns that are not
in the provided library. If `True`, no check is performed; if `False`, a `LibraryError`
is raised instead.
allow_named_layers: If `False`, raises a `PatternError` if any layer is referred to by
a string instead of a number (or tuple).
prune_empty_patterns: Runs `Library.prune_empty()`, recursively deleting any empty patterns.
wrap_repeated_shapes: Runs `Library.wrap_repeated_shapes()`, turning repeated shapes into
repeated refs containing non-repeated shapes.
sort: Whether to sort patterns by name and sort each pattern's contents.
sort_elements: Whether to sort elements within each pattern. Requires
`sort=True`.
allow_dangling_refs: If `None`, warn about missing targets. If `True`,
skip the check. If `False`, raise `LibraryError`.
allow_named_layers: If `False`, raise `PatternError` for string layers.
prune_empty_patterns: Recursively delete empty patterns.
wrap_repeated_shapes: Move shape and label repetitions onto wrapping refs.
Returns:
`lib` or an equivalent sorted library
`lib`, or an equivalent name-sorted `Library` when `sort=True`.
"""
mutable_lib = lib
if sort:
lib = Library(dict(sorted(
(nn, pp.sort(sort_elements=sort_elements)) for nn, pp in lib.items()
mutable_lib = Library(dict(sorted(
(nn, pp.sort(sort_elements=sort_elements)) for nn, pp in mutable_lib.items()
)))
if not allow_dangling_refs:
refs = lib.referenced_patterns()
dangling = refs - set(lib.keys())
refs = mutable_lib.referenced_patterns()
dangling = refs - set(mutable_lib.keys())
if dangling:
msg = 'Dangling refs found: ' + pformat(dangling)
if allow_dangling_refs is None:
@ -72,7 +244,7 @@ def preflight(
if not allow_named_layers:
named_layers: Mapping[str, set] = defaultdict(set)
for name, pat in lib.items():
for name, pat in mutable_lib.items():
for layer in chain(pat.shapes.keys(), pat.labels.keys()):
if isinstance(layer, str):
named_layers[name].add(layer)
@ -81,8 +253,8 @@ def preflight(
raise PatternError('Non-numeric layers found:' + pformat(named_layers))
if prune_empty_patterns:
prune_dangling = 'error' if allow_dangling_refs is False else 'ignore'
pruned = lib.prune_empty(dangling=prune_dangling)
prune_dangling: dangling_mode_t = 'error' if allow_dangling_refs is False else 'ignore'
pruned = mutable_lib.prune_empty(dangling=prune_dangling)
if pruned:
logger.info(f'Preflight pruned {len(pruned)} empty patterns')
logger.debug('Pruned: ' + pformat(pruned))
@ -90,9 +262,9 @@ def preflight(
logger.debug('Preflight found no empty patterns')
if wrap_repeated_shapes:
lib.wrap_repeated_shapes()
mutable_lib.wrap_repeated_shapes()
return lib
return mutable_lib
def mangle_name(name: str) -> str:

5
masque/library/overlay.py
View file

@ -214,9 +214,8 @@ class OverlayLibrary(ILibrary, IMaterializable, IBorrowing):
"""
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`.
Source-backed cells remain lazy until accessed through `__getitem__`, which
persistently materializes a detached, overlay-owned `Pattern`.
Source libraries must remain open and must not be mutated after they are
added. The overlay borrows each source and snapshots its names, hierarchy,

12
masque/test/test_gdsii_lazy_arrow.py
View file

@ -14,6 +14,7 @@ from ..library import IBorrowing, IMaterializable, LayerMappedView, Library, Ove
from ..pattern import Pattern
from ..repetition import Grid
from ..file import gdsii
from ..file.utils import preflight_source_aware
from ..file.gdsii import lazy_arrow as gdsii_lazy_arrow
from ..file.gdsii import lazy_write as gdsii_lazy_write
from ..file.gdsii.perf import write_fixture
@ -301,8 +302,10 @@ def test_gdsii_layer_mapped_view_controls_raw_copy_through(
assert set(roundtrip['leaf'].shapes) == {(20, 0)}
passthrough = LayerMappedView(raw, map_layer, copy_through=True)
preflighted = preflight_source_aware(passthrough)
assert isinstance(preflighted, OverlayLibrary)
copied_file = tmp_path / 'layer_mapped_copied.gds'
gdsii_lazy_arrow.writefile(passthrough, copied_file)
gdsii_lazy_arrow.writefile(preflighted, copied_file)
assert copied == ['leaf', 'mid', 'top']
assert copied_file.read_bytes() == gds_file.read_bytes()
@ -310,12 +313,13 @@ def test_gdsii_layer_mapped_view_controls_raw_copy_through(
copied.clear()
assert set(passthrough['leaf'].shapes) == {(20, 0)}
promoted_file = tmp_path / 'layer_mapped_promoted.gds'
gdsii_lazy_arrow.writefile(passthrough, promoted_file)
preflighted = preflight_source_aware(passthrough)
materialized_file = tmp_path / 'layer_mapped_materialized.gds'
gdsii_lazy_arrow.writefile(preflighted, materialized_file)
assert copied == ['mid', 'top']
assert not raw._cache
roundtrip, _ = gdsii.readfile(promoted_file)
roundtrip, _ = gdsii.readfile(materialized_file)
assert set(roundtrip['leaf'].shapes) == {(20, 0)}

82
masque/test/test_library.py
View file

@ -11,7 +11,7 @@ from ..error import LibraryError, PatternError
from ..ports import Port
from ..repetition import Grid
from ..shapes import Arc, Ellipse, Path, Text
from ..file.utils import preflight
from ..file.utils import preflight, preflight_source_aware
if TYPE_CHECKING:
from ..shapes import Polygon
@ -193,6 +193,84 @@ def test_preflight_prune_empty_preserves_dangling_policy(caplog: pytest.LogCaptu
preflight(make_lib(), allow_dangling_refs=False, prune_empty_patterns=True)
def test_preflight_source_aware_skips_source_backed_patterns() -> None:
source = Library()
source["copied"] = Pattern()
source["materialized"] = Pattern()
mapped_names: list[str] = []
def map_layer(layer: int | tuple[int, int] | str) -> int | tuple[int, int] | str:
mapped_names.append(str(layer))
return layer
view = LayerMappedView(source, map_layer, copy_through=True)
view["materialized"].rect("named", xmin=0, xmax=1, ymin=0, ymax=1)
result = preflight_source_aware(view, allow_named_layers=True)
assert isinstance(result, OverlayLibrary)
assert mapped_names == []
assert result.source_cell("copied") is not None
assert result.source_cell("materialized") is None
assert not source["materialized"].shapes
assert "named" in result["materialized"].shapes
def test_preflight_source_aware_checks_only_cells_without_provenance() -> None:
source = Library()
source["copied"] = Pattern().rect("copied_layer", xmin=0, xmax=1, ymin=0, ymax=1)
source["materialized"] = Pattern().rect("materialized_layer", xmin=0, xmax=1, ymin=0, ymax=1)
view = LayerMappedView(source, lambda layer: layer, copy_through=True)
preflight_source_aware(view, allow_named_layers=False)
view["materialized"]
with pytest.raises(PatternError, match="materialized_layer"):
preflight_source_aware(view, allow_named_layers=False)
def test_preflight_source_aware_prunes_only_safe_checked_patterns() -> None:
source = Library({"child": Pattern(), "parent": Pattern()})
source["parent"].ref("child")
view = LayerMappedView(source, lambda layer: layer, copy_through=True)
view["child"]
result = preflight_source_aware(view, prune_empty_patterns=True)
assert isinstance(result, OverlayLibrary)
assert "child" in result
assert result.source_cell("parent") is not None
view["parent"]
result = preflight_source_aware(view, prune_empty_patterns=True)
assert "child" not in result
assert "parent" not in result
assert "child" in source
assert "child" in source["parent"].refs
def test_preflight_source_aware_wraps_only_checked_patterns() -> None:
source = Library({"copied": Pattern(), "materialized": Pattern()})
for name in source:
source[name].rect((1, 0), xmin=0, xmax=1, ymin=0, ymax=1)
source[name].shapes[(1, 0)][0].repetition = Grid(a_vector=(10, 0), a_count=2)
view = LayerMappedView(source, lambda layer: layer, copy_through=True)
view["materialized"]
result = preflight_source_aware(view, wrap_repeated_shapes=True)
assert isinstance(result, OverlayLibrary)
assert result.source_cell("copied") is not None
assert source["copied"].shapes[(1, 0)][0].repetition is not None
assert result["materialized"].shapes[(1, 0)] == []
wrapper_names = result["materialized"].referenced_patterns()
assert len(wrapper_names) == 1
wrapper_name = next(iter(wrapper_names))
assert wrapper_name is not None
assert result[wrapper_name].shapes[(1, 0)][0].repetition is None
def test_library_flatten() -> None:
lib = Library()
child = Pattern()
@ -667,7 +745,7 @@ def test_lazy_library_subtree_preserves_type_and_shares_materialized_patterns()
assert "child" in lib
def test_overlay_subtree_preserves_type_sources_and_independent_promotion() -> None:
def test_overlay_subtree_preserves_type_sources_and_independent_materialization() -> None:
source = Library({"child": Pattern(), "top": Pattern(), "unused": Pattern()})
source["top"].ref("child")
overlay = OverlayLibrary()