[gdsii] enable lazy write-through

masque/test/test_gdsii_lazy.py

@@ -1,4 +1,5 @@
 from pathlib import Path
+import gzip
 import io
 
 import numpy
@@ -8,6 +9,7 @@ from numpy.testing import assert_allclose
 from ..file import gdsii
 from ..file.gdsii import lazy as gdsii_lazy
 from ..file.gdsii import lazy_write as gdsii_lazy_write
+from ..file.utils import preflight_source_aware
 from ..error import LibraryError
 from ..pattern import Pattern
 from ..ports import Port
@@ -404,6 +406,130 @@ def test_gdsii_lazy_processed_write_roundtrips_without_explicit_units(tmp_path:
     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_lazy.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_lazy.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_lazy.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_lazy.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_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_lazy.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()