masque/masque/test/test_gdsii_lazy_arrow.py

601 lines
21 KiB
Python

from pathlib import Path
import subprocess
import sys
import textwrap
import klamath
import numpy
import pytest
pytest.importorskip('pyarrow')
from .. import PatternError, LibraryError
from ..library import IBorrowing, IMaterializable, LayerMappedView, Library, OverlayLibrary, PortLoadView
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 arrow as gdsii_arrow
from ..file.gdsii import writer as gdsii_writer
from tools.generate_gds_perf import write_fixture
if not gdsii_arrow.is_available():
pytest.skip('klamath_rs_ext shared library is not available', allow_module_level=True)
@pytest.mark.parametrize('cached', [False, True])
def test_arrow_detached_mutation_is_isolated(tmp_path: Path, cached: bool) -> None:
filename = tmp_path / 'detached.gds'
gdsii.writefile(_make_small_library(), filename, meters_per_unit=1e-9)
with gdsii_lazy_arrow.ArrowLibrary.from_file(filename) as lib:
if cached:
lib['leaf']
detached = lib.materialize_detached('leaf')
detached.translate_elements((0.25, 0.5))
numpy.testing.assert_allclose(detached.get_bounds(), [[0.25, 0.5], [10.25, 5.5]])
numpy.testing.assert_allclose(lib.materialize_detached('leaf').get_bounds(), [[0, 0], [10, 5]])
assert lib.can_copy_raw_struct('leaf') == (not cached)
top = lib.materialize_detached('top').flatten(lib)
assert top.get_bounds() is not None
def test_arrow_rect_hierarchy_requires_explicit_polygonization(tmp_path: Path) -> None:
original = _make_small_library()
original['mid'].refs['leaf'][0].rotation = numpy.pi / 4
filename = tmp_path / 'rotated.gds'
gdsii.writefile(original, filename, meters_per_unit=1e-9)
with gdsii_lazy_arrow.ArrowLibrary.from_file(filename) as lib:
for action in ('bounds', 'flatten'):
top = lib.materialize_detached('top')
with pytest.raises(PatternError, match='Pattern.polygonize'):
top.get_bounds(lib) if action == 'bounds' else top.flatten(lib)
lib['leaf'].polygonize()
numpy.testing.assert_allclose(lib['top'].get_bounds(lib), original['top'].get_bounds(original))
assert lib.materialize_detached('top').flatten(lib).get_bounds() is not None
def test_arrow_dangling_ref_queries_are_cache_independent(tmp_path: Path) -> None:
original = Library({'parent': Pattern()})
original['parent'].ref('missing', offset=(3, 4), rotation=numpy.pi / 3, mirrored=True, scale=2)
original['parent'].ref('missing', offset=(10, 20), repetition=Grid(a_vector=(7, 0), a_count=3))
filename = tmp_path / 'dangling.gds'
gdsii.writefile(original, filename, meters_per_unit=1e-9)
with gdsii_lazy_arrow.ArrowLibrary.from_file(filename) as lib:
expected = _local_refs_key(original.find_refs_local('missing', dangling='include'))
assert _local_refs_key(lib.find_refs_local('missing', dangling='include')) == expected
assert lib.can_copy_raw_struct('parent')
assert not lib.find_refs_local('missing', dangling='ignore')
with pytest.raises(LibraryError, match='missing'):
lib.find_refs_local('missing', dangling='error')
assert not lib.find_refs_local('unknown', dangling='include')
lib['parent']
assert _local_refs_key(lib.find_refs_local('missing', dangling='include')) == expected
def test_gdsii_lazy_arrow_has_reader_only_surface() -> None:
assert not hasattr(gdsii_lazy_arrow, 'is_available')
assert not hasattr(gdsii_lazy_arrow, 'write')
assert not hasattr(gdsii_lazy_arrow, 'writefile')
def _make_small_library() -> Library:
lib = Library()
leaf = Pattern()
leaf.polygon((1, 0), vertices=[[0, 0], [10, 0], [10, 5], [0, 5]])
lib['leaf'] = leaf
mid = Pattern()
mid.ref('leaf', offset=(10, 20))
mid.ref('leaf', offset=(40, 0), repetition=Grid(a_vector=(12, 0), a_count=2, b_vector=(0, 9), b_count=2))
lib['mid'] = mid
top = Pattern()
top.ref('mid', offset=(100, 200))
lib['top'] = top
return lib
def _make_complex_ref_library() -> Library:
lib = Library()
leaf = Pattern()
leaf.polygon((1, 0), vertices=[[0, 0], [10, 0], [10, 10], [0, 10]])
lib['leaf'] = leaf
child = Pattern()
child.ref('leaf', offset=(100, 200), rotation=numpy.pi / 2, mirrored=True, scale=1.25)
lib['child'] = child
sibling = Pattern()
sibling.ref(
'leaf',
offset=(-50, 60),
repetition=Grid(a_vector=(20, 0), a_count=3, b_vector=(0, 30), b_count=2),
)
lib['sibling'] = sibling
fanout = Pattern()
fanout.ref('leaf', offset=(0, 0))
fanout.ref('child', offset=(10, 0), mirrored=True, rotation=numpy.pi / 6, scale=1.1)
fanout.ref('leaf', offset=(30, 0), repetition=Grid(a_vector=(5, 0), a_count=2, b_vector=(0, 7), b_count=3))
fanout.ref(
'child',
offset=(40, 0),
mirrored=True,
rotation=numpy.pi / 4,
scale=1.2,
repetition=Grid(a_vector=(9, 0), a_count=2, b_vector=(0, 11), b_count=2),
)
lib['fanout'] = fanout
top = Pattern()
top.ref('child', offset=(500, 600))
top.ref('sibling', offset=(-100, 50), rotation=numpy.pi)
top.ref('fanout', offset=(250, -75))
lib['top'] = top
return lib
def _write_invalid_path_type_fixture(path: Path) -> None:
with path.open('wb') as stream:
header = klamath.library.FileHeader(
name=b'test',
user_units_per_db_unit=1.0,
meters_per_db_unit=1e-9,
)
header.write(stream)
elem = klamath.elements.Path(
layer=(1, 0),
path_type=3,
width=10,
extension=(0, 0),
xy=numpy.array([[0, 0], [10, 0]], dtype=numpy.int32),
properties={},
)
klamath.library.write_struct(stream, name=b'top', elements=[elem])
klamath.records.ENDLIB.write(stream, None)
def _transform_rows_key(values: numpy.ndarray) -> tuple[tuple[object, ...], ...]:
arr = numpy.asarray(values, dtype=float)
arr = numpy.atleast_2d(arr)
rows = [
(
round(float(row[0]), 8),
round(float(row[1]), 8),
round(float(row[2]), 8),
bool(int(round(float(row[3])))),
round(float(row[4]), 8),
)
for row in arr
]
return tuple(sorted(rows))
def _local_refs_key(refs: dict[str, list[numpy.ndarray]]) -> dict[str, tuple[tuple[object, ...], ...]]:
return {
parent: _transform_rows_key(numpy.concatenate(transforms))
for parent, transforms in refs.items()
}
def _global_refs_key(refs: dict[tuple[str, ...], numpy.ndarray]) -> dict[tuple[str, ...], tuple[tuple[object, ...], ...]]:
return {
path: _transform_rows_key(transforms)
for path, transforms in refs.items()
}
def test_gdsii_lazy_arrow_loads_perf_fixture(tmp_path: Path) -> None:
gds_file = tmp_path / 'many_cells_lazy.gds'
manifest = write_fixture(gds_file, preset='many_cells', scale=0.001)
lib, info = gdsii_lazy_arrow.readfile(gds_file)
assert info['name'] == manifest.library_name
assert len(lib) == manifest.cells
assert lib.top() == 'TOP'
assert 'TOP' in lib.child_graph(dangling='ignore')
def test_gdsii_lazy_arrow_local_and_global_refs(tmp_path: Path) -> None:
gds_file = tmp_path / 'refs.gds'
src = _make_small_library()
gdsii.writefile(src, gds_file, meters_per_unit=1e-9, library_name='lazy-refs')
lib, _ = gdsii_lazy_arrow.readfile(gds_file)
local = lib.find_refs_local('leaf')
assert set(local) == {'mid'}
assert sum(arr.shape[0] for arr in local['mid']) == 5
global_refs = lib.find_refs_global('leaf')
assert set(global_refs) == {('top', 'mid', 'leaf')}
assert global_refs[('top', 'mid', 'leaf')].shape[0] == 5
def test_gdsii_lazy_arrow_graph_hooks_observe_cached_edits(tmp_path: Path) -> None:
gds_file = tmp_path / 'lazy_arrow_cached_graph.gds'
gdsii.writefile(_make_small_library(), gds_file, meters_per_unit=1e-9)
lib, _ = gdsii_lazy_arrow.readfile(gds_file)
del lib['mid'].refs['leaf']
assert lib.child_graph(dangling='ignore')['mid'] == set()
assert lib.find_refs_local('leaf') == {}
def test_gdsii_lazy_arrow_ref_queries_match_eager_reader(tmp_path: Path) -> None:
gds_file = tmp_path / 'complex_refs.gds'
src = _make_complex_ref_library()
gdsii.writefile(src, gds_file, meters_per_unit=1e-9, library_name='lazy-complex-refs')
eager, _ = gdsii.readfile(gds_file)
lazy, _ = gdsii_lazy_arrow.readfile(gds_file)
for name in ('leaf', 'child'):
assert _local_refs_key(lazy.find_refs_local(name)) == _local_refs_key(eager.find_refs_local(name))
assert _global_refs_key(lazy.find_refs_global(name)) == _global_refs_key(eager.find_refs_global(name))
def test_gdsii_lazy_arrow_detached_batch_preserves_native_batching(
tmp_path: Path,
monkeypatch: pytest.MonkeyPatch,
) -> None:
gds_file = tmp_path / 'lazy_arrow_detached_batch.gds'
gdsii.writefile(_make_small_library(), gds_file, meters_per_unit=1e-9)
raw, _ = gdsii_lazy_arrow.readfile(gds_file)
mapped = LayerMappedView(raw, lambda layer: layer)
original_read = gdsii_arrow._read_selected_cells_to_arrow
call_count = 0
def count_read(*args, **kwargs) -> object:
nonlocal call_count
call_count += 1
return original_read(*args, **kwargs)
monkeypatch.setattr(gdsii_arrow, '_read_selected_cells_to_arrow', count_read)
detached = mapped.materialize_many_detached(('leaf', 'mid', 'leaf'))
assert tuple(detached) == ('leaf', 'mid')
assert call_count == 1
assert not raw._cache
assert not mapped._cache
def test_gdsii_lazy_arrow_invalid_input_raises_klamath_error(tmp_path: Path) -> None:
gds_file = tmp_path / 'invalid.gds'
gds_file.write_bytes(b'not-a-gds')
script = textwrap.dedent(f"""
from masque.file.gdsii import lazy_arrow as gdsii_lazy_arrow
try:
gdsii_lazy_arrow.readfile({str(gds_file)!r})
except Exception as exc:
print(type(exc).__module__)
print(type(exc).__qualname__)
print(exc)
else:
raise SystemExit('expected gdsii_lazy_arrow.readfile() to fail')
""")
result = subprocess.run([sys.executable, '-c', script], capture_output=True, text=True, check=False)
assert result.returncode == 0, result.stderr
assert 'klamath.basic' in result.stdout
assert 'KlamathError' in result.stdout
def test_gdsii_lazy_arrow_invalid_path_type_raises_pattern_error(tmp_path: Path) -> None:
gds_file = tmp_path / 'invalid_path_type.gds'
_write_invalid_path_type_fixture(gds_file)
lib, _ = gdsii_lazy_arrow.readfile(gds_file)
with pytest.raises(PatternError, match='Unrecognized path type: 3'):
lib['top']
def test_gdsii_lazy_arrow_untouched_write_is_copy_through(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None:
gds_file = tmp_path / 'copy_source.gds'
src = _make_small_library()
gdsii.writefile(src, gds_file, meters_per_unit=1e-9, library_name='copy-through')
lib, info = gdsii_lazy_arrow.readfile(gds_file)
def forbid_materialization(*_args, **_kwargs) -> None:
pytest.fail('Untouched GDS writes must not materialize Arrow coordinates')
monkeypatch.setattr(gdsii_arrow, 'read_arrow', forbid_materialization)
out_file = tmp_path / 'copy_out.gds'
gdsii.writefile(
lib,
out_file,
meters_per_unit=info['meters_per_unit'],
logical_units_per_unit=info['logical_units_per_unit'],
library_name=info['name'],
)
assert out_file.read_bytes() == gds_file.read_bytes()
def test_gdsii_raw_copy_resolves_generic_borrowing_views(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None:
gds_file = tmp_path / 'provenance_source.gds'
gdsii.writefile(_make_small_library(), gds_file, meters_per_unit=1e-9, library_name='provenance')
raw, _ = gdsii_lazy_arrow.readfile(gds_file)
ports = PortLoadView(raw)
subtree = ports.subtree('top')
overlay = OverlayLibrary()
overlay.add_source(subtree)
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 / 'provenance_out.gds'
gdsii.writefile(overlay, out_file)
assert copied == ['leaf', 'mid', 'top']
assert out_file.read_bytes() == gds_file.read_bytes()
renamed = OverlayLibrary()
renamed.add_source(raw)
renamed.rename('top', 'renamed_top')
assert gdsii_writer._resolve_raw_struct(renamed, 'renamed_top') is None
remapped = OverlayLibrary()
remapped.add_source(raw)
remapped.rename('leaf', 'renamed_leaf', move_references=True)
assert gdsii_writer._resolve_raw_struct(remapped, 'mid') is None
def test_gdsii_layer_mapped_view_controls_raw_copy_through(
tmp_path: Path,
monkeypatch: pytest.MonkeyPatch,
) -> None:
gds_file = tmp_path / 'layer_mapped_source.gds'
gdsii.writefile(_make_small_library(), gds_file, meters_per_unit=1e-9, library_name='layer-mapped')
raw, _ = gdsii_lazy_arrow.readfile(gds_file)
copied: list[str] = []
raw_reader = raw.raw_struct_bytes
def record_raw_read(name: str) -> bytes:
copied.append(name)
return raw_reader(name)
def map_layer(layer): # noqa: ANN001,ANN202
return (20, 0) if layer == (1, 0) else layer
monkeypatch.setattr(raw, 'raw_struct_bytes', record_raw_read)
mapped = LayerMappedView(raw, map_layer)
mapped_file = tmp_path / 'layer_mapped_all.gds'
gdsii.writefile(mapped, mapped_file)
assert copied == []
assert not raw._cache
roundtrip, info = gdsii.readfile(mapped_file)
assert info['name'] == 'layer-mapped'
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.writefile(preflighted, copied_file)
assert copied == ['leaf', 'mid', 'top']
assert copied_file.read_bytes() == gds_file.read_bytes()
assert not raw._cache
copied.clear()
assert set(passthrough['leaf'].shapes) == {(20, 0)}
preflighted = preflight_source_aware(passthrough)
materialized_file = tmp_path / 'layer_mapped_materialized.gds'
gdsii.writefile(preflighted, materialized_file)
assert copied == ['mid', 'top']
assert not raw._cache
roundtrip, _ = gdsii.readfile(materialized_file)
assert set(roundtrip['leaf'].shapes) == {(20, 0)}
def test_gdsii_lazy_arrow_processed_cell_edit_disables_raw_copy(
tmp_path: Path,
monkeypatch: pytest.MonkeyPatch,
) -> None:
gds_file = tmp_path / 'processed_edit_source.gds'
src = _make_small_library()
gdsii.writefile(src, gds_file, meters_per_unit=1e-9, library_name='processed-edit')
raw, _ = gdsii_lazy_arrow.readfile(gds_file)
processed = PortLoadView(raw)
processed['top'].polygon((7, 0), vertices=[[0, 0], [4, 0], [0, 4]])
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 / 'processed_edit_out.gds'
gdsii.writefile(processed, out_file)
assert 'top' not in copied
roundtrip, _ = gdsii.readfile(out_file)
assert len(roundtrip['top'].shapes[(7, 0)]) == 1
def test_gdsii_lazy_arrow_subtree_preserves_raw_copy_and_ref_queries(tmp_path: Path) -> None:
gds_file = tmp_path / 'subtree_copy_source.gds'
src = _make_small_library()
src['unused'] = Pattern()
gdsii.writefile(src, gds_file, meters_per_unit=1e-9, library_name='subtree-copy')
raw, _ = gdsii_lazy_arrow.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', 'mid', 'top')
assert _global_refs_key(subtree.find_refs_global('leaf')) == _global_refs_key(raw.find_refs_global('leaf'))
assert not raw._cache
out_file = tmp_path / 'subtree_copy_out.gds'
gdsii.writefile(subtree, out_file)
assert not raw._cache
roundtrip, info = gdsii.readfile(out_file)
assert info['name'] == 'subtree-copy'
assert set(roundtrip) == {'leaf', 'mid', 'top'}
def test_gdsii_lazy_arrow_overlay_subtree_preserves_raw_copy(tmp_path: Path) -> None:
gds_file = tmp_path / 'overlay_subtree_source.gds'
src = _make_small_library()
src['unused'] = Pattern()
gdsii.writefile(src, gds_file, meters_per_unit=1e-9, library_name='overlay-subtree-copy')
raw, _ = gdsii_lazy_arrow.readfile(gds_file)
overlay = OverlayLibrary()
overlay.add_source(raw)
subtree = overlay.subtree('top')
assert isinstance(subtree, OverlayLibrary)
assert subtree.borrowed_sources() == (raw,)
assert not raw._cache
out_file = tmp_path / 'overlay_subtree_out.gds'
gdsii.writefile(subtree, out_file)
assert not raw._cache
roundtrip, info = gdsii.readfile(out_file)
assert info['name'] == 'overlay-subtree-copy'
assert set(roundtrip) == {'leaf', 'mid', 'top'}
def test_gdsii_lazy_arrow_gzipped_copy_through(tmp_path: Path) -> None:
gds_file = tmp_path / 'copy_source.gds.gz'
src = _make_small_library()
gdsii.writefile(src, gds_file, meters_per_unit=1e-9, library_name='copy-through-gz')
lib, info = gdsii_lazy_arrow.readfile(gds_file)
out_file = tmp_path / 'copy_out.gds.gz'
gdsii.writefile(
lib,
out_file,
meters_per_unit=info['meters_per_unit'],
logical_units_per_unit=info['logical_units_per_unit'],
library_name=info['name'],
)
assert out_file.read_bytes() == gds_file.read_bytes()
def test_gdsii_lazy_overlay_merge_and_write(tmp_path: Path) -> None:
base_a = Library()
leaf_a = Pattern()
leaf_a.polygon((1, 0), vertices=[[0, 0], [8, 0], [8, 8], [0, 8]])
base_a['leaf'] = leaf_a
top_a = Pattern()
top_a.ref('leaf', offset=(0, 0))
base_a['top_a'] = top_a
base_b = Library()
leaf_b = Pattern()
leaf_b.polygon((2, 0), vertices=[[0, 0], [5, 0], [5, 5], [0, 5]])
base_b['leaf'] = leaf_b
top_b = Pattern()
top_b.ref('leaf', offset=(20, 30))
base_b['top_b'] = top_b
gds_a = tmp_path / 'a.gds'
gds_b = tmp_path / 'b.gds'
gdsii.writefile(base_a, gds_a, meters_per_unit=1e-9, library_name='overlay')
gdsii.writefile(base_b, gds_b, meters_per_unit=1e-9, library_name='overlay')
lib_a, _ = gdsii_lazy_arrow.readfile(gds_a)
lib_b, _ = gdsii_lazy_arrow.readfile(gds_b)
overlay = OverlayLibrary()
overlay.add_source(lib_a)
rename_map = overlay.add_source(lib_b, rename_theirs=lambda lib, name: lib.get_name(name))
renamed_leaf = rename_map['leaf']
assert rename_map == {'leaf': renamed_leaf}
assert renamed_leaf != 'leaf'
assert len(lib_a._cache) == 0
assert len(lib_b._cache) == 0
overlay.move_references('leaf', renamed_leaf)
out_file = tmp_path / 'overlay_out.gds'
gdsii.writefile(overlay, out_file)
roundtrip, _ = gdsii.readfile(out_file)
assert set(roundtrip.keys()) == {'leaf', renamed_leaf, 'top_a', 'top_b'}
assert 'top_b' in roundtrip
assert list(roundtrip['top_b'].refs.keys()) == [renamed_leaf]
def test_gdsii_writer_accepts_overlay_library(tmp_path: Path) -> None:
gds_file = tmp_path / 'overlay_source.gds'
src = _make_small_library()
gdsii.writefile(src, gds_file, meters_per_unit=1e-9, library_name='overlay-src')
lib, info = gdsii_lazy_arrow.readfile(gds_file)
overlay = OverlayLibrary()
overlay.add_source(lib)
overlay.rename('leaf', 'leaf_copy', move_references=True)
out_file = tmp_path / 'overlay_via_eager_writer.gds'
gdsii.writefile(
overlay,
out_file,
meters_per_unit=info['meters_per_unit'],
logical_units_per_unit=info['logical_units_per_unit'],
library_name=info['name'],
)
roundtrip, _ = gdsii.readfile(out_file)
assert set(roundtrip.keys()) == {'leaf_copy', 'mid', 'top'}
assert list(roundtrip['mid'].refs.keys()) == ['leaf_copy']
def test_svg_writer_uses_detached_materialized_copy(tmp_path: Path) -> None:
pytest.importorskip('svgwrite')
from ..file import svg
from ..shapes import Path as MPath
gds_file = tmp_path / 'svg_source.gds'
src = _make_small_library()
src['top'].path((3, 0), vertices=[[0, 0], [0, 20]], width=4)
gdsii.writefile(src, gds_file, meters_per_unit=1e-9, library_name='svg-src')
lib, _ = gdsii_lazy_arrow.readfile(gds_file)
top_pat = lib['top']
assert list(top_pat.refs.keys()) == ['mid']
assert any(isinstance(shape, MPath) for shape in top_pat.shapes[(3, 0)])
svg_path = tmp_path / 'lazy.svg'
svg.writefile(lib, 'top', str(svg_path))
assert svg_path.exists()
assert list(top_pat.refs.keys()) == ['mid']
assert any(isinstance(shape, MPath) for shape in top_pat.shapes[(3, 0)])