masque/masque/test/test_gdsii_lazy.py

580 lines
19 KiB
Python

from pathlib import Path
import gzip
import io
import numpy
import pytest
from numpy.testing import assert_allclose
from ..file import gdsii
from ..file.gdsii import lazy as gdsii_lazy
from ..file.gdsii import writer as gdsii_writer
from ..file.utils import preflight_source_aware
from ..error import LibraryError
from ..pattern import Pattern
from ..ports import Port
from ..library import (
IBorrowing, IMaterializable, LayerMappedView, LazyLibrary, Library, LibraryView,
OverlayLibrary, PortLoadView,
)
def _make_lazy_port_library() -> Library:
lib = Library()
leaf = Pattern()
leaf.label(layer=(10, 0), string='A:type1 0', offset=(5, 0))
lib['leaf'] = leaf
child = Pattern()
child.ref('leaf', offset=(10, 20), rotation=numpy.pi / 2)
lib['child'] = child
top = Pattern()
top.ref('child', offset=(100, 200))
lib['top'] = top
return lib
def test_gdsii_write_requires_units_without_gds_metadata() -> None:
lib = Library({'top': Pattern()})
lib.library_info = None # type: ignore[attr-defined]
with pytest.raises(LibraryError, match='meters_per_unit is required'):
gdsii.write(lib, io.BytesIO())
def test_gdsii_write_plain_library_defaults() -> None:
lib = _make_lazy_port_library()
stream = io.BytesIO()
gdsii.write(lib, stream, 1e-9)
stream.seek(0)
_roundtrip, info = gdsii.read(stream)
assert info == {
'name': 'masque-klamath',
'meters_per_unit': 1e-9,
'logical_units_per_unit': 1,
}
def test_gdsii_lazy_write_materializes_transiently() -> None:
lib = LazyLibrary()
lib['top'] = Pattern()
stream = io.BytesIO()
gdsii.write(
lib,
stream,
meters_per_unit=1e-9,
logical_units_per_unit=1,
library_name='transient',
)
assert not lib.cache
stream.seek(0)
roundtrip, info = gdsii.read(stream)
assert set(roundtrip) == {'top'}
assert info['name'] == 'transient'
def test_gdsii_raw_copy_provenance_cycle_falls_back() -> None:
class SelfBorrowingView(LibraryView, IBorrowing):
def borrowed_sources(self) -> tuple['SelfBorrowingView', ...]:
return (self,)
def source_cell(self, name: str) -> tuple['SelfBorrowingView', str] | None:
return self, name
view = SelfBorrowingView({'top': Pattern()})
assert gdsii_writer._resolve_raw_struct(view, 'top') is None
def test_gdsii_lazy_source_exposes_order_and_graph_without_materializing(tmp_path: Path) -> None:
gds_file = tmp_path / 'lazy_source.gds'
src = _make_lazy_port_library()
gdsii.writefile(src, gds_file, meters_per_unit=1e-9, library_name='classic-lazy')
lib, info = gdsii_lazy.readfile(gds_file)
assert info['name'] == 'classic-lazy'
assert lib.source_order() == ('leaf', 'child', 'top')
assert lib.child_graph(dangling='ignore') == {
'leaf': set(),
'child': {'leaf'},
'top': {'child'},
}
assert lib.parent_graph() == {
'leaf': {'child'},
'child': {'top'},
'top': set(),
}
assert lib.tops() == ['top']
global_refs = lib.find_refs_global('leaf')
assert_allclose(global_refs[('top', 'child', 'leaf')], [[110, 220, numpy.pi / 2, 0, 1]])
assert not lib._cache
with pytest.raises(ValueError, match='dangling-reference mode'):
lib.child_graph(dangling='typo')
with pytest.raises(ValueError, match='dangling-reference mode'):
lib.find_refs_local('leaf', dangling='typo')
child = lib['child']
assert list(child.refs.keys()) == ['leaf']
assert set(lib._cache) == {'child'}
def test_gdsii_lazy_graph_hooks_observe_cached_edits(tmp_path: Path) -> None:
gds_file = tmp_path / 'lazy_cached_graph.gds'
gdsii.writefile(_make_lazy_port_library(), gds_file, meters_per_unit=1e-9)
lib, _ = gdsii_lazy.readfile(gds_file)
del lib['child'].refs['leaf']
assert lib.child_graph(dangling='ignore')['child'] == set()
assert lib.find_refs_local('leaf') == {}
def test_gdsii_lazy_subtree_stays_borrowed_and_preserves_write_metadata(tmp_path: Path) -> None:
gds_file = tmp_path / 'lazy_subtree_source.gds'
src = _make_lazy_port_library()
src['unused'] = Pattern()
gdsii.writefile(src, gds_file, meters_per_unit=1e-9, library_name='classic-subtree')
raw, _ = gdsii_lazy.readfile(gds_file)
subtree = raw.subtree('top')
assert isinstance(raw, IMaterializable)
assert not isinstance(raw, IBorrowing)
assert isinstance(subtree, IMaterializable)
assert isinstance(subtree, IBorrowing)
assert subtree.source_order() == ('leaf', 'child', 'top')
assert not hasattr(subtree, 'library_info')
assert not raw._cache
out_file = tmp_path / 'lazy_subtree_out.gds'
gdsii.writefile(subtree, out_file)
assert not raw._cache
roundtrip, info = gdsii.readfile(out_file)
assert info['name'] == 'classic-subtree'
assert set(roundtrip) == {'leaf', 'child', 'top'}
def test_gdsii_lazy_borrowed_sources_require_matching_units(tmp_path: Path) -> None:
gds_a = tmp_path / 'units_a.gds'
gds_b = tmp_path / 'units_b.gds'
source = Library({'top': Pattern()})
gdsii.writefile(source, gds_a, meters_per_unit=1e-9, library_name='units-a')
gdsii.writefile(source, gds_b, meters_per_unit=2e-9, library_name='units-b')
lazy_a, _ = gdsii_lazy.readfile(gds_a)
lazy_b, _ = gdsii_lazy.readfile(gds_b)
overlay = OverlayLibrary()
overlay.add_source(lazy_a)
overlay.add_source(lazy_b, rename_theirs=lambda lib, name: lib.get_name(name))
with pytest.raises(LibraryError, match='identical units'):
gdsii.write(overlay, io.BytesIO())
def test_gdsii_lazy_overlay_subtree_preserves_source_laziness(tmp_path: Path) -> None:
gds_file = tmp_path / 'lazy_overlay_subtree_source.gds'
src = _make_lazy_port_library()
src['unused'] = Pattern()
gdsii.writefile(src, gds_file, meters_per_unit=1e-9, library_name='overlay-subtree')
raw, _ = gdsii_lazy.readfile(gds_file)
overlay = OverlayLibrary()
overlay.add_source(raw)
subtree = overlay.subtree('top')
assert isinstance(subtree, OverlayLibrary)
assert subtree.borrowed_sources() == (raw,)
out_file = tmp_path / 'lazy_overlay_subtree_out.gds'
gdsii.writefile(subtree, out_file)
assert not raw._cache
roundtrip, info = gdsii.readfile(out_file)
assert info['name'] == 'overlay-subtree'
assert set(roundtrip) == {'leaf', 'child', 'top'}
def test_gdsii_lazy_port_load_view_keeps_raw_source_unmodified(tmp_path: Path) -> None:
gds_file = tmp_path / 'lazy_ports.gds'
src = _make_lazy_port_library()
gdsii.writefile(src, gds_file, meters_per_unit=1e-9, library_name='classic-ports')
raw, _ = gdsii_lazy.readfile(gds_file)
processed = PortLoadView(raw, layers=[(10, 0)], max_depth=2)
assert not hasattr(processed, 'library_info')
top = processed['top']
assert set(top.ports) == {'A'}
assert_allclose(top.ports['A'].offset, [110, 225], atol=1e-10)
assert not raw._cache
raw_top = raw['top']
assert not raw_top.ports
def test_gdsii_lazy_port_load_view_detaches_previously_cached_source(tmp_path: Path) -> None:
gds_file = tmp_path / 'lazy_cached_ports.gds'
src = _make_lazy_port_library()
gdsii.writefile(src, gds_file, meters_per_unit=1e-9, library_name='classic-cached-ports')
raw, _ = gdsii_lazy.readfile(gds_file)
raw_top = raw['top']
processed = PortLoadView(raw, ports={
'top': {
'P': Port((1, 2), rotation=0, ptype='wire'),
},
})
processed_top = processed['top']
assert processed_top is not raw_top
assert not raw_top.ports
assert set(processed_top.ports) == {'P'}
assert raw['top'] is raw_top
assert not hasattr(processed, 'close')
def test_gdsii_lazy_port_overrides_without_data_stay_lazy(tmp_path: Path) -> None:
gds_file = tmp_path / 'lazy_port_overrides.gds'
src = _make_lazy_port_library()
gdsii.writefile(src, gds_file, meters_per_unit=1e-9, library_name='classic-overrides')
raw, _ = gdsii_lazy.readfile(gds_file)
processed = PortLoadView(raw, ports={
'top': {
'P': Port((1, 2), rotation=0, ptype='wire'),
},
})
top = processed['top']
assert set(top.ports) == {'P'}
assert_allclose(top.ports['P'].offset, [1, 2], atol=1e-10)
assert top.ports['P'].rotation == 0
assert top.ports['P'].ptype == 'wire'
assert not raw._cache
raw_top = raw['top']
assert not raw_top.ports
def test_gdsii_lazy_port_overrides_apply_after_extraction(tmp_path: Path) -> None:
gds_file = tmp_path / 'lazy_ports_override_extracted.gds'
src = _make_lazy_port_library()
gdsii.writefile(src, gds_file, meters_per_unit=1e-9, library_name='classic-override-extracted')
raw, _ = gdsii_lazy.readfile(gds_file)
processed = PortLoadView(
raw,
layers=[(10, 0)],
max_depth=2,
ports={
'top': {
'A': Port((1, 2), rotation=numpy.pi, ptype='manual'),
'B': Port((3, 4), rotation=None, ptype=None),
},
},
)
top = processed['top']
assert set(top.ports) == {'A', 'B'}
assert_allclose(top.ports['A'].offset, [1, 2], atol=1e-10)
assert top.ports['A'].rotation == numpy.pi
assert top.ports['A'].ptype == 'manual'
assert_allclose(top.ports['B'].offset, [3, 4], atol=1e-10)
assert top.ports['B'].rotation is None
assert top.ports['B'].ptype is None
assert not raw._cache
def test_gdsii_lazy_port_overrides_replace_extracted_ports(tmp_path: Path) -> None:
gds_file = tmp_path / 'lazy_ports_replace.gds'
src = _make_lazy_port_library()
gdsii.writefile(src, gds_file, meters_per_unit=1e-9, library_name='classic-replace-ports')
raw, _ = gdsii_lazy.readfile(gds_file)
processed = PortLoadView(
raw,
layers=[(10, 0)],
max_depth=2,
ports={
'top': {
'B': Port((3, 4), rotation=None, ptype=None),
},
},
replace=True,
)
top = processed['top']
assert set(top.ports) == {'B'}
assert_allclose(top.ports['B'].offset, [3, 4], atol=1e-10)
assert not raw._cache
def test_gdsii_lazy_overlay_add_source_stays_lazy_for_processed_view(tmp_path: Path) -> None:
gds_file = tmp_path / 'lazy_overlay.gds'
src = _make_lazy_port_library()
gdsii.writefile(src, gds_file, meters_per_unit=1e-9, library_name='classic-overlay')
raw, _ = gdsii_lazy.readfile(gds_file)
processed = PortLoadView(raw, layers=[(10, 0)], max_depth=2)
overlay = OverlayLibrary()
overlay.add_source(processed)
assert not raw._cache
assert not processed._cache
abstract = overlay.abstract('top')
assert set(abstract.ports) == {'A'}
def test_gdsii_lazy_overlay_add_source_sees_port_overrides(tmp_path: Path) -> None:
gds_file = tmp_path / 'lazy_overlay_override.gds'
src = _make_lazy_port_library()
gdsii.writefile(src, gds_file, meters_per_unit=1e-9, library_name='classic-overlay-override')
raw, _ = gdsii_lazy.readfile(gds_file)
processed = PortLoadView(raw, ports={
'top': {
'P': Port((1, 2), rotation=0, ptype='wire'),
},
})
overlay = OverlayLibrary()
overlay.add_source(processed)
assert not raw._cache
assert not processed._cache
abstract = overlay.abstract('top')
assert set(abstract.ports) == {'P'}
assert_allclose(abstract.ports['P'].offset, [1, 2], atol=1e-10)
def test_gdsii_lazy_overlay_add_source_can_rename_every_source_cell() -> None:
src = _make_lazy_port_library()
overlay = OverlayLibrary()
rename_map = overlay.add_source(
src,
rename_theirs=lambda _lib, name: f'mapped_{name}',
rename_when='always',
)
assert rename_map == {
'leaf': 'mapped_leaf',
'child': 'mapped_child',
'top': 'mapped_top',
}
assert tuple(overlay.keys()) == ('mapped_leaf', 'mapped_child', 'mapped_top')
assert 'mapped_leaf' in overlay['mapped_child'].refs
def test_gdsii_lazy_overlay_add_source_rename_when_validation() -> None:
src = _make_lazy_port_library()
with pytest.raises(TypeError, match='rename_theirs'):
OverlayLibrary().add_source(src, rename_theirs=None, rename_when='always')
with pytest.raises(ValueError, match='rename mode'):
OverlayLibrary().add_source(src, rename_when='sometimes') # type: ignore[arg-type]
def test_gdsii_lazy_processed_write_roundtrips_without_explicit_units(tmp_path: Path) -> None:
gds_file = tmp_path / 'lazy_roundtrip.gds'
src = _make_lazy_port_library()
gdsii.writefile(src, gds_file, meters_per_unit=1e-9, library_name='classic-roundtrip')
raw, _ = gdsii_lazy.readfile(gds_file)
processed = PortLoadView(raw, layers=[(10, 0)], max_depth=2)
out_file = tmp_path / 'lazy_roundtrip_out.gds'
gdsii.writefile(processed, out_file)
assert out_file.read_bytes() == gds_file.read_bytes()
def test_gdsii_lazy_source_aware_write_copies_untouched_structures(
tmp_path: Path,
monkeypatch: pytest.MonkeyPatch,
) -> None:
gds_file = tmp_path / 'classic_copy_source.gds'
src = _make_lazy_port_library()
gdsii.writefile(src, gds_file, meters_per_unit=1e-9, library_name='classic-copy')
raw, _ = gdsii_lazy.readfile(gds_file)
ports = PortLoadView(raw, layers=[(10, 0)], max_depth=2)
mapped = LayerMappedView(ports, lambda layer: (20, 0) if layer == (10, 0) else layer, copy_through=True)
prepared = preflight_source_aware(mapped)
copied: list[str] = []
raw_reader = raw.raw_struct_bytes
def record_raw_read(name: str) -> bytes:
copied.append(name)
return raw_reader(name)
def fail_materialize(_name: str, *, persist: bool = True) -> Pattern: # noqa: ARG001
raise AssertionError('untouched cells must not be materialized')
monkeypatch.setattr(raw, 'raw_struct_bytes', record_raw_read)
monkeypatch.setattr(raw, 'materialize', fail_materialize)
out_file = tmp_path / 'classic_copy_out.gds'
gdsii.writefile(prepared, out_file)
assert copied == ['leaf', 'child', 'top']
assert not raw._cache
assert not ports._cache
assert not mapped._cache
assert out_file.read_bytes() == gds_file.read_bytes()
@pytest.mark.parametrize('use_mmap', [False, True])
def test_gdsii_lazy_raw_copy_stream_backends(tmp_path: Path, use_mmap: bool) -> None:
gds_file = tmp_path / 'classic_stream_source.gds'
src = _make_lazy_port_library()
gdsii.writefile(src, gds_file, meters_per_unit=1e-9, library_name='classic-stream')
raw, _ = gdsii_lazy.readfile(gds_file, use_mmap=use_mmap)
out_file = tmp_path / f'classic_stream_{use_mmap}.gds'
gdsii.writefile(raw, out_file)
assert not raw._cache
assert out_file.read_bytes() == gds_file.read_bytes()
def test_gdsii_lazy_raw_copy_gzipped_source(tmp_path: Path) -> None:
gds_file = tmp_path / 'classic_gzip_source.gds'
gz_file = tmp_path / 'classic_gzip_source.gds.gz'
src = _make_lazy_port_library()
gdsii.writefile(src, gds_file, meters_per_unit=1e-9, library_name='classic-gzip')
with gzip.open(gz_file, 'wb') as stream:
stream.write(gds_file.read_bytes())
raw, _ = gdsii_lazy.readfile(gz_file, use_mmap=False)
for name in raw.source_order():
raw.raw_struct_bytes(name)
assert raw._source.stream.tell() == raw._cells[name].struct_end
out_file = tmp_path / 'classic_gzip_out.gds'
gdsii.writefile(raw, out_file)
assert not raw._cache
assert out_file.read_bytes() == gds_file.read_bytes()
def test_gdsii_lazy_cached_source_cell_disables_only_its_raw_copy(
tmp_path: Path,
monkeypatch: pytest.MonkeyPatch,
) -> None:
gds_file = tmp_path / 'classic_cached_source.gds'
src = _make_lazy_port_library()
gdsii.writefile(src, gds_file, meters_per_unit=1e-9, library_name='classic-cached')
raw, _ = gdsii_lazy.readfile(gds_file)
assert raw.can_copy_raw_struct('leaf')
raw['leaf'].label((30, 0), string='cached', offset=(1, 2))
assert not raw.can_copy_raw_struct('leaf')
copied: list[str] = []
raw_reader = raw.raw_struct_bytes
def record_raw_read(name: str) -> bytes:
copied.append(name)
return raw_reader(name)
monkeypatch.setattr(raw, 'raw_struct_bytes', record_raw_read)
out_file = tmp_path / 'classic_cached_out.gds'
gdsii.writefile(raw, out_file)
assert copied == ['child', 'top']
roundtrip, _ = gdsii.readfile(out_file)
assert set(roundtrip['leaf'].labels) == {(10, 0), (30, 0)}
def test_gdsii_lazy_detached_processing_copies_only_cached_patterns(
tmp_path: Path,
monkeypatch: pytest.MonkeyPatch,
) -> None:
gds_file = tmp_path / 'classic_detached_source.gds'
gdsii.writefile(_make_lazy_port_library(), gds_file, meters_per_unit=1e-9)
raw, _ = gdsii_lazy.readfile(gds_file)
mapped = LayerMappedView(raw, lambda layer: layer)
original_deepcopy = Pattern.deepcopy
copied: list[Pattern] = []
def count_deepcopy(pattern: Pattern) -> Pattern:
copied.append(pattern)
return original_deepcopy(pattern)
monkeypatch.setattr(Pattern, 'deepcopy', count_deepcopy)
fresh = mapped.materialize_detached('top')
assert not copied
assert not raw._cache
assert not mapped._cache
cached = raw['top']
copied.clear()
detached = mapped.materialize_detached('top')
assert copied == [cached]
assert detached is not cached
assert detached is not fresh
def test_gdsii_lazy_materialized_cell_disables_only_its_raw_copy(
tmp_path: Path,
monkeypatch: pytest.MonkeyPatch,
) -> None:
gds_file = tmp_path / 'classic_partial_copy_source.gds'
src = _make_lazy_port_library()
gdsii.writefile(src, gds_file, meters_per_unit=1e-9, library_name='classic-partial-copy')
raw, _ = gdsii_lazy.readfile(gds_file)
mapped = LayerMappedView(raw, lambda layer: (20, 0) if layer == (10, 0) else layer, copy_through=True)
assert set(mapped['leaf'].labels) == {(20, 0)}
prepared = preflight_source_aware(mapped)
copied: list[str] = []
raw_reader = raw.raw_struct_bytes
def record_raw_read(name: str) -> bytes:
copied.append(name)
return raw_reader(name)
monkeypatch.setattr(raw, 'raw_struct_bytes', record_raw_read)
out_file = tmp_path / 'classic_partial_copy_out.gds'
gdsii.writefile(prepared, out_file)
assert copied == ['child', 'top']
assert not raw._cache
roundtrip, _ = gdsii.readfile(out_file)
assert set(roundtrip['leaf'].labels) == {(20, 0)}
def test_gdsii_lazy_layer_mapped_write_materializes_without_source_cache(tmp_path: Path) -> None:
gds_file = tmp_path / 'lazy_layer_source.gds'
src = _make_lazy_port_library()
gdsii.writefile(src, gds_file, meters_per_unit=1e-9, library_name='classic-layers')
raw, _ = gdsii_lazy.readfile(gds_file)
mapped = LayerMappedView(raw, lambda layer: (20, 0) if layer == (10, 0) else layer)
out_file = tmp_path / 'lazy_layer_mapped.gds'
gdsii.writefile(mapped, out_file)
assert not raw._cache
roundtrip, info = gdsii.readfile(out_file)
assert info['name'] == 'classic-layers'
assert set(roundtrip['leaf'].labels) == {(20, 0)}
assert (10, 0) not in roundtrip['leaf'].labels
def test_gdsii_removed_closure_based_lazy_loader() -> None:
assert not hasattr(gdsii, 'load_library')
assert not hasattr(gdsii, 'load_libraryfile')
assert not hasattr(gdsii_lazy, 'write')
assert not hasattr(gdsii_lazy, 'writefile')