[gdsii] cleanup

This commit is contained in:
Jan Petykiewicz 2026-07-14 21:24:34 -07:00
commit 29947c6c01
14 changed files with 153 additions and 901 deletions

View file

@ -764,6 +764,12 @@ An optional Arrow/native backend is available through
replacement unless their additional dependencies and native library are
available.
GDS writing now has one entry point for eager and lazy libraries. Use
`masque.file.gdsii.write()` or `masque.file.gdsii.writefile()` regardless of
which reader produced the library. The `lazy` and `lazy_arrow` modules no
longer re-export writer functions. Source-backed libraries continue to infer
their header metadata and copy untouched structures directly.
## Shape construction and geometry additions
The public `raw=True` constructor shortcut was removed from `Arc`, `Circle`,

View file

@ -1,5 +0,0 @@
from masque.file.gdsii.perf import main
if __name__ == '__main__':
raise SystemExit(main())

View file

@ -3,8 +3,6 @@ GDSII file format readers and writers.
"""
from .klamath import check_valid_names as check_valid_names
from .klamath import read as read
from .klamath import read_elements as read_elements
from .klamath import readfile as readfile
from .klamath import rint_cast as rint_cast
from .klamath import write as write
from .klamath import writefile as writefile
from .writer import write as write
from .writer import writefile as writefile

View file

@ -1,4 +1,4 @@
# ruff: noqa: ARG001, F401
# ruff: noqa: ARG001
"""
GDSII file format readers and writers using the `TODO` library.
@ -23,32 +23,37 @@ Notes:
TODO writing
TODO warn on boxes, nodes
"""
from typing import IO, cast, Any
from collections.abc import Iterable, Mapping, Callable
from __future__ import annotations
from typing import TYPE_CHECKING, Any
from functools import cache
from importlib.machinery import EXTENSION_SUFFIXES
import importlib.util
import mmap
import logging
import os
import pathlib
import gzip
import string
import sys
import tempfile
from pprint import pformat
from klamath.basic import KlamathError
import numpy
from numpy.typing import NDArray
import pyarrow
from pyarrow.cffi import ffi
from ..utils import is_gzipped, tmpfile
from ... import Pattern, Ref, PatternError, LibraryError, Label, Shape
from ..utils import is_gzipped
from ... import Pattern, Ref, PatternError, Label
from ...shapes import Polygon, Path, PolyCollection, RectCollection
from ...repetition import Grid
from ...utils import layer_t, annotations_t
from ...library import LazyLibrary, Library, ILibrary, ILibraryView
from ...library import Library
if TYPE_CHECKING:
from collections.abc import Callable
import mmap
from numpy.typing import NDArray
from ...utils import annotations_t
logger = logging.getLogger(__name__)
@ -69,10 +74,7 @@ ffi.cdef(
"""
)
clib: Any | None = None
path_cap_map = {
_PATH_CAP_MAP = {
0: Path.Cap.Flush,
1: Path.Cap.Circle,
2: Path.Cap.Square,
@ -140,7 +142,7 @@ def _repo_library_candidates() -> list[pathlib.Path]:
]
def find_klamath_rs_library() -> pathlib.Path | None:
def _find_klamath_rs_library() -> pathlib.Path | None:
env_path = os.environ.get('KLAMATH_RS_EXT_LIB')
if env_path:
candidate = pathlib.Path(env_path).expanduser()
@ -159,21 +161,19 @@ def find_klamath_rs_library() -> pathlib.Path | None:
def is_available() -> bool:
return find_klamath_rs_library() is not None
return _find_klamath_rs_library() is not None
@cache
def _get_clib() -> Any:
global clib # noqa: PLW0603
if clib is None:
lib_path = find_klamath_rs_library()
if lib_path is None:
raise ImportError(
'Could not locate klamath_rs_ext shared library. '
'Build klamath-rs with `cargo build --release --manifest-path klamath-rs/Cargo.toml` '
'or set KLAMATH_RS_EXT_LIB to the built library path.'
)
clib = ffi.dlopen(str(lib_path))
return clib
lib_path = _find_klamath_rs_library()
if lib_path is None:
raise ImportError(
'Could not locate klamath_rs_ext shared library. '
'Build klamath-rs with `cargo build --release --manifest-path klamath-rs/Cargo.toml` '
'or set KLAMATH_RS_EXT_LIB to the built library path.'
)
return ffi.dlopen(str(lib_path))
def _read_annotations(
@ -727,9 +727,9 @@ def _gpaths_to_mpaths(
vertices = xy_val[xy_offs[ee]:xy_offs[ee + 1]]
width = elem_widths[ee]
cap_int = int(elem_path_types[ee])
if cap_int not in path_cap_map:
if cap_int not in _PATH_CAP_MAP:
raise PatternError(f'Unrecognized path type: {cap_int}')
cap = path_cap_map[cap_int]
cap = _PATH_CAP_MAP[cap_int]
if cap_int == 4:
cap_extensions = elem_extensions[ee]
else:
@ -838,41 +838,3 @@ def _boundary_props_to_polygons(
annotations = _read_annotations(prop_offs, prop_key, prop_val, ee)
poly = Polygon._from_raw(vertices=vertices, annotations=annotations)
pat.shapes[layer].append(poly)
#def _properties_to_annotations(properties: pyarrow.Array) -> annotations_t:
# return {prop['key'].as_py(): prop['value'].as_py() for prop in properties}
def check_valid_names(
names: Iterable[str],
max_length: int = 32,
) -> None:
"""
Check all provided names to see if they're valid GDSII cell names.
Args:
names: Collection of names to check
max_length: Max allowed length
"""
allowed_chars = set(string.ascii_letters + string.digits + '_?$')
bad_chars = [
name for name in names
if not set(name).issubset(allowed_chars)
]
bad_lengths = [
name for name in names
if len(name) > max_length
]
if bad_chars:
logger.error('Names contain invalid characters:\n' + pformat(bad_chars))
if bad_lengths:
logger.error(f'Names too long (>{max_length}:\n' + pformat(bad_chars))
if bad_chars or bad_lengths:
raise LibraryError('Library contains invalid names, see log above')

View file

@ -19,7 +19,7 @@ Notes:
* GDS creation/modification/access times are set to 1900-01-01 for reproducibility.
* Gzip modification time is set to 0 (start of current epoch, usually 1970-01-01)
"""
from typing import IO, cast, Any
from typing import IO, Any
from collections.abc import Iterable, Mapping, Callable
from types import MappingProxyType
import logging
@ -33,7 +33,7 @@ from numpy.typing import ArrayLike, NDArray
import klamath
from klamath import records
from ..utils import is_gzipped, tmpfile
from ..utils import is_gzipped
from ... import Pattern, Ref, PatternError, LibraryError, Label, Shape
from ...shapes import Polygon, Path, RectCollection
from ...repetition import Grid
@ -44,17 +44,17 @@ from ...library import Library
logger = logging.getLogger(__name__)
path_cap_map = {
_PATH_CAP_MAP = {
0: Path.Cap.Flush,
1: Path.Cap.Circle,
2: Path.Cap.Square,
4: Path.Cap.SquareCustom,
}
RO_EMPTY_DICT: Mapping[int, bytes] = MappingProxyType({})
_EMPTY_PROPERTIES: Mapping[int, bytes] = MappingProxyType({})
def rint_cast(val: ArrayLike) -> NDArray[numpy.int32]:
def _rint_cast(val: ArrayLike) -> NDArray[numpy.int32]:
return numpy.rint(val).astype(numpy.int32)
@ -84,99 +84,6 @@ def _write_footer(stream: IO[bytes]) -> None:
records.ENDLIB.write(stream, None)
def write(
library: Mapping[str, Pattern],
stream: IO[bytes],
meters_per_unit: float,
logical_units_per_unit: float = 1,
library_name: str = 'masque-klamath',
) -> None:
"""
Convert a library to a GDSII stream, mapping data as follows:
Pattern -> GDSII structure
Ref -> GDSII SREF or AREF
Path -> GSDII path
Shape (other than path) -> GDSII boundary/ies
Label -> GDSII text
annnotations -> properties, where possible
For each shape,
layer is chosen to be equal to `shape.layer` if it is an int,
or `shape.layer[0]` if it is a tuple
datatype is chosen to be `shape.layer[1]` if available,
otherwise `0`
GDS does not support shape repetition (only cell repetition). Please call
`library.wrap_repeated_shapes()` before writing to file.
Other functions you may want to call:
- `masque.file.gdsii.check_valid_names(library.keys())` to check for invalid names
- `library.dangling_refs()` to check for references to missing patterns
- `pattern.polygonize()` for any patterns with shapes other
than `masque.shapes.Polygon` or `masque.shapes.Path`
Args:
library: A {name: Pattern} mapping of patterns to write.
meters_per_unit: Written into the GDSII file, meters per (database) length unit.
All distances are assumed to be an integer multiple of this unit, and are stored as such.
logical_units_per_unit: Written into the GDSII file. Allows the GDSII to specify a
"logical" unit which is different from the "database" unit, for display purposes.
Default `1`.
library_name: Library name written into the GDSII file.
Default 'masque-klamath'.
"""
_write_header(stream, meters_per_unit, logical_units_per_unit, library_name)
# Now create a structure for each pattern, and add in any Boundary and SREF elements
for name, pat in library.items():
_write_pattern_struct(stream, name, pat)
_write_footer(stream)
def _writefile_impl(
writer: Callable[..., None],
library: Any,
filename: str | pathlib.Path,
*args: Any,
**kwargs: Any,
) -> None:
path = pathlib.Path(filename)
with tmpfile(path) as base_stream:
streams: tuple[Any, ...] = (base_stream,)
if path.suffix == '.gz':
stream = cast('IO[bytes]', gzip.GzipFile(filename='', mtime=0, fileobj=base_stream, mode='wb', compresslevel=6))
streams = (stream,) + streams
else:
stream = base_stream
try:
writer(library, stream, *args, **kwargs)
finally:
for ss in streams:
ss.close()
def writefile(
library: Mapping[str, Pattern],
filename: str | pathlib.Path,
*args,
**kwargs,
) -> None:
"""
Wrapper for `write()` that takes a filename or path instead of a stream.
Will automatically compress the file if it has a .gz suffix.
Args:
library: {name: Pattern} pairs to save.
filename: Filename to save to.
*args: passed to `write()`
**kwargs: passed to `write()`
"""
_writefile_impl(write, library, filename, *args, **kwargs)
def readfile(
filename: str | pathlib.Path,
*args,
@ -233,7 +140,7 @@ def read(
found_struct = records.BGNSTR.skip_past(stream)
while found_struct:
name = records.STRNAME.skip_and_read(stream)
pat = read_elements(stream, raw_mode=raw_mode)
pat = _read_elements(stream, raw_mode=raw_mode)
mlib[name.decode('ASCII')] = pat
found_struct = records.BGNSTR.skip_past(stream)
@ -253,7 +160,7 @@ def _read_header(stream: IO[bytes]) -> dict[str, Any]:
return library_info
def read_elements(
def _read_elements(
stream: IO[bytes],
raw_mode: bool = True,
) -> Pattern:
@ -335,8 +242,8 @@ def _gref_to_mref(ref: klamath.library.Reference) -> tuple[str, Ref]:
def _gpath_to_mpath(gpath: klamath.library.Path, raw_mode: bool) -> tuple[layer_t, Path]:
if gpath.path_type in path_cap_map:
cap = path_cap_map[gpath.path_type]
if gpath.path_type in _PATH_CAP_MAP:
cap = _PATH_CAP_MAP[gpath.path_type]
else:
raise PatternError(f'Unrecognized path type: {gpath.path_type}')
@ -398,7 +305,7 @@ def _mrefs_to_grefs(refs: dict[str | None, list[Ref]]) -> list[klamath.library.R
])
aref = klamath.library.Reference(
struct_name=encoded_name,
xy=rint_cast(xy),
xy=_rint_cast(xy),
colrow=(numpy.rint(rep.a_count), numpy.rint(rep.b_count)),
angle_deg=angle_deg,
invert_y=ref.mirrored,
@ -409,7 +316,7 @@ def _mrefs_to_grefs(refs: dict[str | None, list[Ref]]) -> list[klamath.library.R
elif rep is None:
sref = klamath.library.Reference(
struct_name=encoded_name,
xy=rint_cast([ref.offset]),
xy=_rint_cast([ref.offset]),
colrow=None,
angle_deg=angle_deg,
invert_y=ref.mirrored,
@ -421,7 +328,7 @@ def _mrefs_to_grefs(refs: dict[str | None, list[Ref]]) -> list[klamath.library.R
new_srefs = [
klamath.library.Reference(
struct_name=encoded_name,
xy=rint_cast([ref.offset + dd]),
xy=_rint_cast([ref.offset + dd]),
colrow=None,
angle_deg=angle_deg,
invert_y=ref.mirrored,
@ -441,7 +348,7 @@ def _properties_to_annotations(properties: Mapping[int, bytes]) -> annotations_t
def _annotations_to_properties(annotations: annotations_t, max_len: int = 126) -> Mapping[int, bytes]:
if annotations is None:
return RO_EMPTY_DICT
return _EMPTY_PROPERTIES
cum_len = 0
props = {}
for key, vals in annotations.items():
@ -478,13 +385,13 @@ def _shapes_to_elements(
properties = _annotations_to_properties(shape.annotations, 128)
if isinstance(shape, Path) and not polygonize_paths:
xy = rint_cast(shape.vertices + shape.offset)
width = rint_cast(shape.width)
path_type = next(k for k, v in path_cap_map.items() if v == shape.cap) # reverse lookup
xy = _rint_cast(shape.vertices + shape.offset)
width = _rint_cast(shape.width)
path_type = next(k for k, v in _PATH_CAP_MAP.items() if v == shape.cap) # reverse lookup
extension: tuple[int, int]
if shape.cap == Path.Cap.SquareCustom and shape.cap_extensions is not None:
extension = tuple(rint_cast(shape.cap_extensions))
extension = tuple(_rint_cast(shape.cap_extensions))
else:
extension = (0, 0)
@ -500,10 +407,10 @@ def _shapes_to_elements(
elif isinstance(shape, RectCollection):
for rect in shape.rects:
xy_closed = numpy.empty((5, 2), dtype=numpy.int32)
xy_closed[0] = rint_cast((rect[0], rect[1]))
xy_closed[1] = rint_cast((rect[0], rect[3]))
xy_closed[2] = rint_cast((rect[2], rect[3]))
xy_closed[3] = rint_cast((rect[2], rect[1]))
xy_closed[0] = _rint_cast((rect[0], rect[1]))
xy_closed[1] = _rint_cast((rect[0], rect[3]))
xy_closed[2] = _rint_cast((rect[2], rect[3]))
xy_closed[3] = _rint_cast((rect[2], rect[1]))
xy_closed[4] = xy_closed[0]
boundary = klamath.elements.Boundary(
layer=(layer, data_type),
@ -542,7 +449,7 @@ def _labels_to_texts(labels: dict[layer_t, list[Label]]) -> list[klamath.element
layer, text_type = _mlayer2gds(mlayer)
for label in lseq:
properties = _annotations_to_properties(label.annotations, 128)
xy = rint_cast([label.offset])
xy = _rint_cast([label.offset])
text = klamath.elements.Text(
layer=(layer, text_type),
xy=xy,

View file

@ -29,8 +29,7 @@ import pathlib
import klamath
from klamath import records
from . import klamath as gdsii
from .lazy_write import write as write, writefile as writefile
from . import klamath as gdsii_klamath
from ..utils import is_gzipped
from ...error import LibraryError
from ...library import (
@ -100,7 +99,7 @@ def _open_source_stream(
def _scan_library(
stream: IO[bytes],
) -> tuple[dict[str, Any], list[str], dict[str, _CellScan]]:
library_info = gdsii._read_header(stream)
library_info = gdsii_klamath._read_header(stream)
order: list[str] = []
cells: dict[str, _CellScan] = {}
@ -134,7 +133,7 @@ class GdsLibrarySource(ILibraryView, IMaterializable):
Cells are scanned once up front to discover order, byte ranges, and child
edges. Untouched structures can be copied directly, while accessed cells
are materialized through the classic `gdsii.read_elements` path.
are materialized through the classic GDS decoder.
The source owns the stream lifetime, preserves on-disk ordering through
`source_order()`, and answers graph queries from scan metadata whenever
@ -217,7 +216,7 @@ class GdsLibrarySource(ILibraryView, IMaterializable):
self._lookups_in_progress.append(name)
try:
self._source.stream.seek(self._cells[name].offset)
pat = gdsii.read_elements(self._source.stream, raw_mode=True)
pat = gdsii_klamath._read_elements(self._source.stream, raw_mode=True)
finally:
self._lookups_in_progress.pop()

View file

@ -16,7 +16,6 @@ import pathlib
import numpy
from . import arrow
from .lazy_write import write as write, writefile as writefile
from ..utils import is_gzipped
from ...library import (
ILibraryView,
@ -83,10 +82,6 @@ class _ScanPayload:
cells: dict[str, _CellScan]
refs: _ScanRefs
def is_available() -> bool:
return arrow.is_available()
def _open_source_buffer(path: pathlib.Path) -> _SourceBuffer:
if is_gzipped(path):
with gzip.open(path, mode='rb') as stream:
@ -372,9 +367,3 @@ def readfile(
) -> tuple[ArrowLibrary, dict[str, Any]]:
lib = ArrowLibrary.from_file(filename)
return lib, lib.library_info
def load_libraryfile(
filename: str | pathlib.Path,
) -> tuple[ArrowLibrary, dict[str, Any]]:
return readfile(filename)

View file

@ -1,626 +0,0 @@
"""
Synthetic GDS fixture generation for reader/writer performance testing.
The presets here are intentionally hierarchical and deterministic. They aim to
approximate a pair of real-world layout families discussed during GDS reader and
writer work:
* `many_cells`: tens of thousands of cells, moderate reference count, very heavy
box usage after flattening, and moderate polygon density.
* `many_instances`: a much smaller cell library with very high reference count,
similar box density, and far fewer polygons.
Fixtures are written by streaming structures through `klamath` directly so large
benchmark files can be produced without first materializing an equally large
`masque.Library` in Python.
"""
from __future__ import annotations
from dataclasses import asdict, dataclass
from pathlib import Path
from typing import Any
import argparse
import json
import math
import numpy
import klamath
from klamath import elements
EMPTY_PROPERTIES: dict[int, bytes] = {}
METERS_PER_DB_UNIT = 1e-9
USER_UNITS_PER_DB_UNIT = 1e-3
TOTAL_LAYERS = 200
@dataclass(frozen=True)
class FixturePreset:
name: str
total_layers: int
box_layers: int
heavy_box_layers: int
polygon_layers: int
box_cells: int
poly_cells: int
box_wrappers: int
poly_wrappers: int
box_clusters: int
poly_clusters: int
box_cluster_refs: int
poly_cluster_refs: int
top_direct_box_refs: int
top_direct_poly_refs: int
heavy_boxes_per_cell: int
regular_boxes_per_cell: int
polygons_per_cell: int
path_stride: int
text_stride: int
box_cluster_array: tuple[int, int]
top_box_array: tuple[int, int]
poly_cluster_array: tuple[int, int]
top_poly_array: tuple[int, int]
rare_annotation_stride: int
PRESETS: dict[str, FixturePreset] = {
'many_cells': FixturePreset(
name='many_cells',
total_layers=TOTAL_LAYERS,
box_layers=20,
heavy_box_layers=3,
polygon_layers=20,
box_cells=17_000,
poly_cells=6_000,
box_wrappers=18_000,
poly_wrappers=6_000,
box_clusters=2_000,
poly_clusters=999,
box_cluster_refs=24,
poly_cluster_refs=16,
top_direct_box_refs=21_000,
top_direct_poly_refs=7_000,
heavy_boxes_per_cell=6,
regular_boxes_per_cell=2,
polygons_per_cell=50,
path_stride=2,
text_stride=3,
box_cluster_array=(24, 16),
top_box_array=(8, 8),
poly_cluster_array=(4, 2),
top_poly_array=(3, 2),
rare_annotation_stride=1_250,
),
'many_instances': FixturePreset(
name='many_instances',
total_layers=TOTAL_LAYERS,
box_layers=25,
heavy_box_layers=3,
polygon_layers=10,
box_cells=2_500,
poly_cells=500,
box_wrappers=1_000,
poly_wrappers=500,
box_clusters=1_000,
poly_clusters=499,
box_cluster_refs=1_200,
poly_cluster_refs=400,
top_direct_box_refs=102_001,
top_direct_poly_refs=0,
heavy_boxes_per_cell=40,
regular_boxes_per_cell=16,
polygons_per_cell=60,
path_stride=1,
text_stride=2,
box_cluster_array=(1, 1),
top_box_array=(1, 1),
poly_cluster_array=(1, 1),
top_poly_array=(1, 1),
rare_annotation_stride=250,
),
}
@dataclass(frozen=True)
class FixtureManifest:
preset: str
scale: float
gds_path: str
library_name: str
cells: int
refs: int
layers: int
box_layers: int
heavy_box_layers: list[list[int]]
polygon_layers: list[list[int]]
hierarchical_boxes_per_heavy_layer: int
hierarchical_boxes_per_regular_layer: int
hierarchical_polygons_total: int
hierarchical_paths_total: int
hierarchical_texts_total: int
flattened_box_placements: int
flattened_poly_placements: int
estimated_flat_boxes_per_heavy_layer: int
estimated_flat_polygons_per_active_polygon_layer: int
def _scaled_count(value: int, scale: float, minimum: int = 0) -> int:
if value == 0:
return 0
scaled = int(math.ceil(value * scale))
return max(minimum, scaled)
def _scaled_preset(preset: FixturePreset, scale: float) -> FixturePreset:
if scale <= 0:
raise ValueError(f'scale must be positive, got {scale!r}')
return FixturePreset(
name=preset.name,
total_layers=preset.total_layers,
box_layers=min(preset.box_layers, preset.total_layers),
heavy_box_layers=min(preset.heavy_box_layers, preset.box_layers),
polygon_layers=min(preset.polygon_layers, preset.total_layers),
box_cells=_scaled_count(preset.box_cells, scale, minimum=1),
poly_cells=_scaled_count(preset.poly_cells, scale, minimum=1),
box_wrappers=_scaled_count(preset.box_wrappers, scale),
poly_wrappers=_scaled_count(preset.poly_wrappers, scale),
box_clusters=_scaled_count(preset.box_clusters, scale, minimum=1),
poly_clusters=_scaled_count(preset.poly_clusters, scale, minimum=1),
box_cluster_refs=_scaled_count(preset.box_cluster_refs, scale, minimum=1),
poly_cluster_refs=_scaled_count(preset.poly_cluster_refs, scale, minimum=1),
top_direct_box_refs=_scaled_count(preset.top_direct_box_refs, scale),
top_direct_poly_refs=_scaled_count(preset.top_direct_poly_refs, scale),
heavy_boxes_per_cell=max(1, preset.heavy_boxes_per_cell),
regular_boxes_per_cell=max(1, preset.regular_boxes_per_cell),
polygons_per_cell=max(1, preset.polygons_per_cell),
path_stride=max(1, preset.path_stride),
text_stride=max(1, preset.text_stride),
box_cluster_array=preset.box_cluster_array,
top_box_array=preset.top_box_array,
poly_cluster_array=preset.poly_cluster_array,
top_poly_array=preset.top_poly_array,
rare_annotation_stride=max(1, _scaled_count(preset.rare_annotation_stride, scale, minimum=1)),
)
def _rect_xy(xmin: int, ymin: int, xmax: int, ymax: int) -> numpy.ndarray[Any, numpy.dtype[numpy.int32]]:
return numpy.array(
[[xmin, ymin], [xmin, ymax], [xmax, ymax], [xmax, ymin], [xmin, ymin]],
dtype=numpy.int32,
)
def _poly_xy(points: list[tuple[int, int]]) -> numpy.ndarray[Any, numpy.dtype[numpy.int32]]:
closed = points + [points[0]]
return numpy.array(closed, dtype=numpy.int32)
def _sref(
target: str,
xy: tuple[int, int],
properties: dict[int, bytes] | None = None,
) -> elements.Reference:
return klamath.library.Reference(
struct_name=target.encode('ASCII'),
invert_y=False,
mag=1.0,
angle_deg=0.0,
xy=numpy.array([xy], dtype=numpy.int32),
colrow=None,
properties=EMPTY_PROPERTIES if properties is None else properties,
)
def _aref(
target: str,
origin: tuple[int, int],
counts: tuple[int, int],
step: tuple[int, int],
properties: dict[int, bytes] | None = None,
) -> elements.Reference:
cols, rows = counts
dx, dy = step
xy = numpy.array(
[
origin,
(origin[0] + cols * dx, origin[1]),
(origin[0], origin[1] + rows * dy),
],
dtype=numpy.int32,
)
return klamath.library.Reference(
struct_name=target.encode('ASCII'),
invert_y=False,
mag=1.0,
angle_deg=0.0,
xy=xy,
colrow=(cols, rows),
properties=EMPTY_PROPERTIES if properties is None else properties,
)
def _annotation(index: int) -> dict[int, bytes]:
return {1: f'perf-{index}'.encode('ASCII')}
def _make_box_cell(index: int, cfg: FixturePreset) -> list[elements.Element]:
cell_elements: list[elements.Element] = []
xbase = (index % 17) * 600
ybase = (index // 17) * 180
for layer in range(cfg.heavy_box_layers):
for box_idx in range(cfg.heavy_boxes_per_cell):
x0 = xbase + box_idx * 22
y0 = ybase + layer * 40
width = 10 + ((index + box_idx + layer) % 7) * 6
height = 10 + ((index * 3 + box_idx + layer) % 5) * 8
properties = _annotation(index) if index % cfg.rare_annotation_stride == 0 and box_idx == 0 and layer == 0 else EMPTY_PROPERTIES
cell_elements.append(elements.Boundary(
layer=(layer, 0),
xy=_rect_xy(x0, y0, x0 + width, y0 + height),
properties=properties,
))
for layer in range(cfg.heavy_box_layers, cfg.box_layers):
for box_idx in range(cfg.regular_boxes_per_cell):
x0 = xbase + box_idx * 38
y0 = ybase + (layer - cfg.heavy_box_layers) * 28 + 400
width = 18 + ((index + layer + box_idx) % 9) * 4
height = 12 + ((index + 2 * layer + box_idx) % 6) * 5
cell_elements.append(elements.Boundary(
layer=(layer, 0),
xy=_rect_xy(x0, y0, x0 + width, y0 + height),
properties=EMPTY_PROPERTIES,
))
return cell_elements
def _make_poly_cell(index: int, cfg: FixturePreset) -> list[elements.Element]:
cell_elements: list[elements.Element] = []
xbase = (index % 19) * 900
ybase = (index // 19) * 260
for poly_idx in range(cfg.polygons_per_cell):
layer = poly_idx % cfg.polygon_layers
dx = xbase + (poly_idx % 5) * 120
dy = ybase + (poly_idx // 5) * 80
size = 18 + ((index + poly_idx + layer) % 11) * 7
points = [
(dx, dy),
(dx + size, dy + size // 5),
(dx + size + size // 3, dy + size),
(dx + size // 2, dy + size + size // 2),
(dx - size // 4, dy + size // 2),
]
properties = _annotation(index) if poly_idx == 0 and index % cfg.rare_annotation_stride == 0 else EMPTY_PROPERTIES
cell_elements.append(elements.Boundary(
layer=(layer, 0),
xy=_poly_xy(points),
properties=properties,
))
if index % cfg.path_stride == 0:
layer = index % cfg.polygon_layers
cell_elements.append(elements.Path(
layer=(layer, 1),
path_type=2,
width=12 + (index % 5) * 4,
extension=(0, 0),
xy=numpy.array(
[
[xbase, ybase + 900],
[xbase + 240, ybase + 930],
[xbase + 420, ybase + 960],
],
dtype=numpy.int32,
),
properties=EMPTY_PROPERTIES,
))
if index % cfg.text_stride == 0:
layer = index % cfg.polygon_layers
properties = _annotation(index) if index % cfg.rare_annotation_stride == 0 else EMPTY_PROPERTIES
cell_elements.append(elements.Text(
layer=(layer, 2),
presentation=0,
path_type=0,
width=0,
invert_y=False,
mag=1.0,
angle_deg=0.0,
xy=numpy.array([[xbase + 64, ybase + 1536]], dtype=numpy.int32),
string=f'T{index:05d}'.encode('ASCII'),
properties=properties,
))
return cell_elements
def _write_struct(stream: Any, name: str, cell_elements: list[elements.Element]) -> None:
klamath.library.write_struct(stream, name=name.encode('ASCII'), elements=cell_elements)
def _box_name(index: int) -> str:
return f'box_{index:05d}'
def _poly_name(index: int) -> str:
return f'poly_{index:05d}'
def _box_wrapper_name(index: int) -> str:
return f'box_wrap_{index:05d}'
def _poly_wrapper_name(index: int) -> str:
return f'poly_wrap_{index:05d}'
def _box_cluster_name(index: int) -> str:
return f'box_cluster_{index:05d}'
def _poly_cluster_name(index: int) -> str:
return f'poly_cluster_{index:05d}'
def _write_box_cells(stream: Any, cfg: FixturePreset) -> None:
for idx in range(cfg.box_cells):
_write_struct(stream, _box_name(idx), _make_box_cell(idx, cfg))
def _write_poly_cells(stream: Any, cfg: FixturePreset) -> None:
for idx in range(cfg.poly_cells):
_write_struct(stream, _poly_name(idx), _make_poly_cell(idx, cfg))
def _write_wrappers(stream: Any, cfg: FixturePreset) -> None:
for idx in range(cfg.box_wrappers):
target = _box_name(idx % cfg.box_cells)
origin = ((idx % 97) * 2_000, (idx // 97) * 2_000)
_write_struct(stream, _box_wrapper_name(idx), [_sref(target, origin)])
for idx in range(cfg.poly_wrappers):
target = _poly_name(idx % cfg.poly_cells)
origin = ((idx % 61) * 3_200, (idx // 61) * 3_200)
_write_struct(stream, _poly_wrapper_name(idx), [_sref(target, origin)])
def _write_box_clusters(stream: Any, cfg: FixturePreset) -> None:
array_refs = min(cfg.box_cluster_refs, max(1, (3 * cfg.box_cluster_refs) // 4))
for idx in range(cfg.box_clusters):
cell_elements: list[elements.Element] = []
for ref_idx in range(cfg.box_cluster_refs):
target = _box_name((idx * cfg.box_cluster_refs + ref_idx) % cfg.box_cells)
origin = (
(ref_idx % 6) * 48_000,
(ref_idx // 6) * 48_000,
)
if ref_idx < array_refs:
cell_elements.append(_aref(target, origin, cfg.box_cluster_array, (720, 900)))
else:
cell_elements.append(_sref(target, origin))
_write_struct(stream, _box_cluster_name(idx), cell_elements)
def _write_poly_clusters(stream: Any, cfg: FixturePreset) -> None:
array_refs = min(cfg.poly_cluster_refs, cfg.poly_cluster_refs // 2)
for idx in range(cfg.poly_clusters):
cell_elements: list[elements.Element] = []
for ref_idx in range(cfg.poly_cluster_refs):
target = _poly_name((idx * cfg.poly_cluster_refs + ref_idx) % cfg.poly_cells)
origin = (
(ref_idx % 10) * 96_000,
(ref_idx // 10) * 96_000,
)
if ref_idx < array_refs:
cell_elements.append(_aref(target, origin, cfg.poly_cluster_array, (12_000, 8_500)))
else:
cell_elements.append(_sref(target, origin))
_write_struct(stream, _poly_cluster_name(idx), cell_elements)
def _top_box_refs(cfg: FixturePreset) -> list[elements.Reference]:
refs: list[elements.Reference] = []
for idx in range(cfg.box_wrappers):
refs.append(_sref(
_box_wrapper_name(idx),
((idx % 240) * 240_000, (idx // 240) * 240_000),
))
for idx in range(cfg.box_clusters):
refs.append(_sref(
_box_cluster_name(idx),
((idx % 100) * 800_000, (idx // 100) * 800_000 + 14_000_000),
))
for idx in range(cfg.top_direct_box_refs):
target = _box_name(idx % cfg.box_cells)
origin = (
(idx % 150) * 160_000,
(idx // 150) * 160_000 + 26_000_000,
)
if cfg.top_box_array == (1, 1):
refs.append(_sref(target, origin))
else:
refs.append(_aref(target, origin, cfg.top_box_array, (1_100, 1_350)))
return refs
def _top_poly_refs(cfg: FixturePreset) -> list[elements.Reference]:
refs: list[elements.Reference] = []
for idx in range(cfg.poly_wrappers):
refs.append(_sref(
_poly_wrapper_name(idx),
((idx % 180) * 360_000, (idx // 180) * 360_000 + 44_000_000),
))
for idx in range(cfg.poly_clusters):
refs.append(_sref(
_poly_cluster_name(idx),
((idx % 70) * 1_100_000, (idx // 70) * 1_100_000 + 58_000_000),
))
for idx in range(cfg.top_direct_poly_refs):
target = _poly_name(idx % cfg.poly_cells)
origin = (
(idx % 110) * 420_000,
(idx // 110) * 420_000 + 72_000_000,
)
if cfg.top_poly_array == (1, 1):
refs.append(_sref(target, origin))
else:
refs.append(_aref(target, origin, cfg.top_poly_array, (16_000, 14_000)))
return refs
def _write_top(stream: Any, cfg: FixturePreset) -> None:
cell_elements: list[elements.Element] = []
cell_elements.extend(_top_box_refs(cfg))
cell_elements.extend(_top_poly_refs(cfg))
_write_struct(stream, 'TOP', cell_elements)
def fixture_manifest(path: str | Path, preset: str, scale: float = 1.0) -> FixtureManifest:
base = PRESETS[preset]
cfg = _scaled_preset(base, scale)
box_cluster_array_refs = min(cfg.box_cluster_refs, max(1, (3 * cfg.box_cluster_refs) // 4))
box_cluster_array_mult = cfg.box_cluster_array[0] * cfg.box_cluster_array[1]
box_cluster_ref_instances = (
box_cluster_array_refs * box_cluster_array_mult
+ (cfg.box_cluster_refs - box_cluster_array_refs)
)
poly_cluster_array_refs = min(cfg.poly_cluster_refs, cfg.poly_cluster_refs // 2)
poly_cluster_array_mult = cfg.poly_cluster_array[0] * cfg.poly_cluster_array[1]
poly_cluster_ref_instances = (
poly_cluster_array_refs * poly_cluster_array_mult
+ (cfg.poly_cluster_refs - poly_cluster_array_refs)
)
flattened_box_placements = (
cfg.box_wrappers
+ cfg.box_clusters * box_cluster_ref_instances
+ cfg.top_direct_box_refs * cfg.top_box_array[0] * cfg.top_box_array[1]
)
flattened_poly_placements = (
cfg.poly_wrappers
+ cfg.poly_clusters * poly_cluster_ref_instances
+ cfg.top_direct_poly_refs * cfg.top_poly_array[0] * cfg.top_poly_array[1]
)
polygon_layers = max(1, cfg.polygon_layers)
polys_per_layer = (cfg.poly_cells * cfg.polygons_per_cell) // polygon_layers
return FixtureManifest(
preset=cfg.name,
scale=scale,
gds_path=str(Path(path)),
library_name=f'masque-perf-{cfg.name}',
cells=cfg.box_cells + cfg.poly_cells + cfg.box_wrappers + cfg.poly_wrappers + cfg.box_clusters + cfg.poly_clusters + 1,
refs=(
cfg.box_wrappers
+ cfg.poly_wrappers
+ cfg.box_clusters * cfg.box_cluster_refs
+ cfg.poly_clusters * cfg.poly_cluster_refs
+ cfg.box_wrappers + cfg.poly_wrappers + cfg.box_clusters + cfg.poly_clusters
+ cfg.top_direct_box_refs + cfg.top_direct_poly_refs
),
layers=cfg.total_layers,
box_layers=cfg.box_layers,
heavy_box_layers=[[layer, 0] for layer in range(cfg.heavy_box_layers)],
polygon_layers=[[layer, 0] for layer in range(cfg.polygon_layers)],
hierarchical_boxes_per_heavy_layer=cfg.box_cells * cfg.heavy_boxes_per_cell,
hierarchical_boxes_per_regular_layer=cfg.box_cells * cfg.regular_boxes_per_cell,
hierarchical_polygons_total=cfg.poly_cells * cfg.polygons_per_cell,
hierarchical_paths_total=(cfg.poly_cells - 1) // cfg.path_stride + 1,
hierarchical_texts_total=(cfg.poly_cells - 1) // cfg.text_stride + 1,
flattened_box_placements=flattened_box_placements,
flattened_poly_placements=flattened_poly_placements,
estimated_flat_boxes_per_heavy_layer=flattened_box_placements * cfg.heavy_boxes_per_cell,
estimated_flat_polygons_per_active_polygon_layer=flattened_poly_placements * polys_per_layer // cfg.poly_cells if cfg.poly_cells else 0,
)
def write_fixture(
path: str | Path,
*,
preset: str,
scale: float = 1.0,
write_manifest: bool = True,
) -> FixtureManifest:
if preset not in PRESETS:
known = ', '.join(sorted(PRESETS))
raise KeyError(f'unknown preset {preset!r}; expected one of: {known}')
manifest = fixture_manifest(path, preset, scale)
cfg = _scaled_preset(PRESETS[preset], scale)
output = Path(path)
output.parent.mkdir(parents=True, exist_ok=True)
with output.open('wb') as stream:
header = klamath.library.FileHeader(
name=manifest.library_name.encode('ASCII'),
user_units_per_db_unit=USER_UNITS_PER_DB_UNIT,
meters_per_db_unit=METERS_PER_DB_UNIT,
)
header.write(stream)
_write_box_cells(stream, cfg)
_write_poly_cells(stream, cfg)
_write_wrappers(stream, cfg)
_write_box_clusters(stream, cfg)
_write_poly_clusters(stream, cfg)
_write_top(stream, cfg)
klamath.records.ENDLIB.write(stream, None)
if write_manifest:
manifest_path = output.with_suffix(output.suffix + '.json')
manifest_path.write_text(json.dumps(asdict(manifest), indent=2, sort_keys=True) + '\n')
return manifest
def build_arg_parser() -> argparse.ArgumentParser:
parser = argparse.ArgumentParser(description='Generate synthetic GDS fixtures for GDS reader/writer performance work.')
parser.add_argument(
'preset',
nargs='?',
default='many_cells',
choices=sorted(PRESETS),
help='Fixture family to generate.',
)
parser.add_argument(
'output',
nargs='?',
help='Output .gds path. Defaults to build/gds_perf/<preset>.gds',
)
parser.add_argument(
'--scale',
type=float,
default=1.0,
help='Scale the preset counts down or up while keeping the same shape mix. Default: 1.0',
)
parser.add_argument(
'--no-manifest',
action='store_true',
help='Do not write the sidecar JSON manifest.',
)
return parser
def main(argv: list[str] | None = None) -> int:
parser = build_arg_parser()
args = parser.parse_args(argv)
output = Path(args.output) if args.output is not None else Path('build/gds_perf') / f'{args.preset}.gds'
manifest = write_fixture(output, preset=args.preset, scale=args.scale, write_manifest=not args.no_manifest)
print(json.dumps(asdict(manifest), indent=2, sort_keys=True))
return 0
if __name__ == '__main__':
raise SystemExit(main())

View file

@ -1,5 +1,5 @@
"""
GDS write helpers for source-backed lazy GDS views.
GDSII writer for eager and source-backed libraries.
The generic mutable overlay and ports-importing view live in `masque.library`.
This module preserves source-backed GDS copy-through behavior where possible,
@ -8,16 +8,18 @@ or remapped.
"""
from __future__ import annotations
from typing import IO, TYPE_CHECKING, Any, Protocol, runtime_checkable
from typing import IO, TYPE_CHECKING, Any, Protocol, cast, runtime_checkable
import gzip
import logging
import pathlib
from . import klamath as gdsii
from . import klamath
from ..utils import tmpfile
from ...error import LibraryError
from ...library import IBorrowing, ILibraryView, IMaterializable
if TYPE_CHECKING:
from collections.abc import Mapping
import pathlib
from ...pattern import Pattern
@ -104,19 +106,21 @@ def _get_write_info(
name = info['name'] if library_name is None else library_name
return meters, logical, name
if meters_per_unit is None or logical_units_per_unit is None or library_name is None:
raise LibraryError('meters_per_unit, logical_units_per_unit, and library_name are required for non-GDS-backed lazy writes')
return meters_per_unit, logical_units_per_unit, library_name
if meters_per_unit is None:
raise LibraryError('meters_per_unit is required when writing a library without GDS metadata')
logical = 1 if logical_units_per_unit is None else logical_units_per_unit
name = 'masque-klamath' if library_name is None else library_name
return meters_per_unit, logical, name
def write(
library: Mapping[str, Pattern] | ILibraryView,
stream: IO[bytes],
*,
meters_per_unit: float | None = None,
logical_units_per_unit: float | None = None,
library_name: str | None = None,
) -> None:
"""Write an eager or source-backed library to a GDSII stream."""
meters_per_unit, logical_units_per_unit, library_name = _get_write_info(
library,
meters_per_unit=meters_per_unit,
@ -124,7 +128,7 @@ def write(
library_name=library_name,
)
gdsii._write_header(stream, meters_per_unit, logical_units_per_unit, library_name)
klamath._write_header(stream, meters_per_unit, logical_units_per_unit, library_name)
names = library.source_order() if isinstance(library, ILibraryView) else tuple(library)
for name in names:
@ -139,24 +143,32 @@ def write(
pat = library.materialize(name, persist=False)
else:
pat = library[name]
gdsii._write_pattern_struct(stream, name, pat)
klamath._write_pattern_struct(stream, name, pat)
gdsii._write_footer(stream)
klamath._write_footer(stream)
def writefile(
library: Mapping[str, Pattern] | ILibraryView,
filename: str | pathlib.Path,
*,
meters_per_unit: float | None = None,
logical_units_per_unit: float | None = None,
library_name: str | None = None,
) -> None:
gdsii._writefile_impl(
write,
library,
filename,
meters_per_unit=meters_per_unit,
logical_units_per_unit=logical_units_per_unit,
library_name=library_name,
)
"""Write an eager or source-backed library to a path, compressing `.gz` files."""
path = pathlib.Path(filename)
with tmpfile(path) as base_stream:
if path.suffix == '.gz':
stream = cast('IO[bytes]', gzip.GzipFile(
filename='',
mtime=0,
fileobj=base_stream,
mode='wb',
compresslevel=6,
))
try:
write(library, stream, meters_per_unit, logical_units_per_unit, library_name)
finally:
stream.close()
else:
write(library, base_stream, meters_per_unit, logical_units_per_unit, library_name)

View file

@ -78,3 +78,8 @@ def test_gdsii_check_valid_names_validates_generator_lengths() -> None:
with pytest.raises(LibraryError, match="invalid names"):
gdsii.check_valid_names(names)
def test_gdsii_does_not_export_codec_helpers() -> None:
assert not hasattr(gdsii, 'read_elements')
assert not hasattr(gdsii, 'rint_cast')

View file

@ -16,7 +16,7 @@ from ..repetition import Grid
from ..shapes import Path as MPath, Polygon, PolyCollection, RectCollection
from ..file import gdsii
from ..file.gdsii import arrow as gdsii_arrow
from ..file.gdsii.perf import write_fixture
from tools.generate_gds_perf import write_fixture
if not gdsii_arrow.is_available():

View file

@ -8,7 +8,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.gdsii import writer as gdsii_writer
from ..file.utils import preflight_source_aware
from ..error import LibraryError
from ..pattern import Pattern
@ -37,29 +37,25 @@ def _make_lazy_port_library() -> Library:
return lib
def test_gdsii_lazy_write_ignores_non_mapping_library_info() -> None:
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='required for non-GDS-backed lazy writes'):
gdsii_lazy.write(lib, io.BytesIO())
with pytest.raises(LibraryError, match='meters_per_unit is required'):
gdsii.write(lib, io.BytesIO())
def test_gdsii_lazy_write_plain_library_matches_eager_writer() -> None:
def test_gdsii_write_plain_library_defaults() -> None:
lib = _make_lazy_port_library()
eager_stream = io.BytesIO()
lazy_stream = io.BytesIO()
gdsii.write(lib, eager_stream, meters_per_unit=1e-9, library_name='writer-match')
gdsii_lazy.write(
lib,
lazy_stream,
meters_per_unit=1e-9,
logical_units_per_unit=1,
library_name='writer-match',
)
assert lazy_stream.getvalue() == eager_stream.getvalue()
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:
@ -67,7 +63,7 @@ def test_gdsii_lazy_write_materializes_transiently() -> None:
lib['top'] = Pattern()
stream = io.BytesIO()
gdsii_lazy.write(
gdsii.write(
lib,
stream,
meters_per_unit=1e-9,
@ -92,7 +88,7 @@ def test_gdsii_raw_copy_provenance_cycle_falls_back() -> None:
view = SelfBorrowingView({'top': Pattern()})
assert gdsii_lazy_write._resolve_raw_struct(view, 'top') is None
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:
@ -158,7 +154,7 @@ def test_gdsii_lazy_subtree_stays_borrowed_and_preserves_write_metadata(tmp_path
assert not raw._cache
out_file = tmp_path / 'lazy_subtree_out.gds'
gdsii_lazy.writefile(subtree, out_file)
gdsii.writefile(subtree, out_file)
assert not raw._cache
roundtrip, info = gdsii.readfile(out_file)
@ -180,7 +176,7 @@ def test_gdsii_lazy_borrowed_sources_require_matching_units(tmp_path: Path) -> N
overlay.add_source(lazy_b, rename_theirs=lambda lib, name: lib.get_name(name))
with pytest.raises(LibraryError, match='identical units'):
gdsii_lazy.write(overlay, io.BytesIO())
gdsii.write(overlay, io.BytesIO())
def test_gdsii_lazy_overlay_subtree_preserves_source_laziness(tmp_path: Path) -> None:
@ -198,7 +194,7 @@ def test_gdsii_lazy_overlay_subtree_preserves_source_laziness(tmp_path: Path) ->
assert subtree.borrowed_sources() == (raw,)
out_file = tmp_path / 'lazy_overlay_subtree_out.gds'
gdsii_lazy.writefile(subtree, out_file)
gdsii.writefile(subtree, out_file)
assert not raw._cache
roundtrip, info = gdsii.readfile(out_file)
@ -401,7 +397,7 @@ def test_gdsii_lazy_processed_write_roundtrips_without_explicit_units(tmp_path:
processed = PortLoadView(raw, layers=[(10, 0)], max_depth=2)
out_file = tmp_path / 'lazy_roundtrip_out.gds'
gdsii_lazy.writefile(processed, out_file)
gdsii.writefile(processed, out_file)
assert out_file.read_bytes() == gds_file.read_bytes()
@ -431,7 +427,7 @@ def test_gdsii_lazy_source_aware_write_copies_untouched_structures(
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)
gdsii.writefile(prepared, out_file)
assert copied == ['leaf', 'child', 'top']
assert not raw._cache
@ -448,7 +444,7 @@ def test_gdsii_lazy_raw_copy_stream_backends(tmp_path: Path, use_mmap: bool) ->
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)
gdsii.writefile(raw, out_file)
assert not raw._cache
assert out_file.read_bytes() == gds_file.read_bytes()
@ -467,7 +463,7 @@ def test_gdsii_lazy_raw_copy_gzipped_source(tmp_path: Path) -> None:
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)
gdsii.writefile(raw, out_file)
assert not raw._cache
assert out_file.read_bytes() == gds_file.read_bytes()
@ -494,7 +490,7 @@ def test_gdsii_lazy_cached_source_cell_disables_only_its_raw_copy(
monkeypatch.setattr(raw, 'raw_struct_bytes', record_raw_read)
out_file = tmp_path / 'classic_cached_out.gds'
gdsii_lazy.writefile(raw, out_file)
gdsii.writefile(raw, out_file)
assert copied == ['child', 'top']
roundtrip, _ = gdsii.readfile(out_file)
@ -522,7 +518,7 @@ def test_gdsii_lazy_materialized_cell_disables_only_its_raw_copy(
monkeypatch.setattr(raw, 'raw_struct_bytes', record_raw_read)
out_file = tmp_path / 'classic_partial_copy_out.gds'
gdsii_lazy.writefile(prepared, out_file)
gdsii.writefile(prepared, out_file)
assert copied == ['child', 'top']
assert not raw._cache
@ -538,7 +534,7 @@ def test_gdsii_lazy_layer_mapped_write_materializes_without_source_cache(tmp_pat
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_lazy.writefile(mapped, out_file)
gdsii.writefile(mapped, out_file)
assert not raw._cache
roundtrip, info = gdsii.readfile(out_file)
@ -550,3 +546,5 @@ def test_gdsii_lazy_layer_mapped_write_materializes_without_source_cache(tmp_pat
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')

View file

@ -16,14 +16,21 @@ 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
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_lazy_arrow.is_available():
if not gdsii_arrow.is_available():
pytest.skip('klamath_rs_ext shared library is not available', allow_module_level=True)
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()
@ -225,7 +232,7 @@ def test_gdsii_lazy_arrow_untouched_write_is_copy_through(tmp_path: Path) -> Non
lib, info = gdsii_lazy_arrow.readfile(gds_file)
out_file = tmp_path / 'copy_out.gds'
gdsii_lazy_arrow.writefile(
gdsii.writefile(
lib,
out_file,
meters_per_unit=info['meters_per_unit'],
@ -255,7 +262,7 @@ def test_gdsii_raw_copy_resolves_generic_borrowing_views(tmp_path: Path, monkeyp
monkeypatch.setattr(raw, 'raw_struct_bytes', record_raw_read)
out_file = tmp_path / 'provenance_out.gds'
gdsii_lazy_arrow.writefile(overlay, out_file)
gdsii.writefile(overlay, out_file)
assert copied == ['leaf', 'mid', 'top']
assert out_file.read_bytes() == gds_file.read_bytes()
@ -263,12 +270,12 @@ def test_gdsii_raw_copy_resolves_generic_borrowing_views(tmp_path: Path, monkeyp
renamed = OverlayLibrary()
renamed.add_source(raw)
renamed.rename('top', 'renamed_top')
assert gdsii_lazy_write._resolve_raw_struct(renamed, 'renamed_top') is None
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_lazy_write._resolve_raw_struct(remapped, 'mid') is None
assert gdsii_writer._resolve_raw_struct(remapped, 'mid') is None
def test_gdsii_layer_mapped_view_controls_raw_copy_through(
@ -293,7 +300,7 @@ def test_gdsii_layer_mapped_view_controls_raw_copy_through(
mapped = LayerMappedView(raw, map_layer)
mapped_file = tmp_path / 'layer_mapped_all.gds'
gdsii_lazy_arrow.writefile(mapped, mapped_file)
gdsii.writefile(mapped, mapped_file)
assert copied == []
assert not raw._cache
@ -305,7 +312,7 @@ def test_gdsii_layer_mapped_view_controls_raw_copy_through(
preflighted = preflight_source_aware(passthrough)
assert isinstance(preflighted, OverlayLibrary)
copied_file = tmp_path / 'layer_mapped_copied.gds'
gdsii_lazy_arrow.writefile(preflighted, copied_file)
gdsii.writefile(preflighted, copied_file)
assert copied == ['leaf', 'mid', 'top']
assert copied_file.read_bytes() == gds_file.read_bytes()
@ -315,7 +322,7 @@ def test_gdsii_layer_mapped_view_controls_raw_copy_through(
assert set(passthrough['leaf'].shapes) == {(20, 0)}
preflighted = preflight_source_aware(passthrough)
materialized_file = tmp_path / 'layer_mapped_materialized.gds'
gdsii_lazy_arrow.writefile(preflighted, materialized_file)
gdsii.writefile(preflighted, materialized_file)
assert copied == ['mid', 'top']
assert not raw._cache
@ -345,7 +352,7 @@ def test_gdsii_lazy_arrow_processed_cell_edit_disables_raw_copy(
monkeypatch.setattr(raw, 'raw_struct_bytes', record_raw_read)
out_file = tmp_path / 'processed_edit_out.gds'
gdsii_lazy_arrow.writefile(processed, out_file)
gdsii.writefile(processed, out_file)
assert 'top' not in copied
roundtrip, _ = gdsii.readfile(out_file)
@ -370,7 +377,7 @@ def test_gdsii_lazy_arrow_subtree_preserves_raw_copy_and_ref_queries(tmp_path: P
assert not raw._cache
out_file = tmp_path / 'subtree_copy_out.gds'
gdsii_lazy_arrow.writefile(subtree, out_file)
gdsii.writefile(subtree, out_file)
assert not raw._cache
roundtrip, info = gdsii.readfile(out_file)
@ -394,7 +401,7 @@ def test_gdsii_lazy_arrow_overlay_subtree_preserves_raw_copy(tmp_path: Path) ->
assert not raw._cache
out_file = tmp_path / 'overlay_subtree_out.gds'
gdsii_lazy_arrow.writefile(subtree, out_file)
gdsii.writefile(subtree, out_file)
assert not raw._cache
roundtrip, info = gdsii.readfile(out_file)
@ -409,7 +416,7 @@ def test_gdsii_lazy_arrow_gzipped_copy_through(tmp_path: Path) -> None:
lib, info = gdsii_lazy_arrow.readfile(gds_file)
out_file = tmp_path / 'copy_out.gds.gz'
gdsii_lazy_arrow.writefile(
gdsii.writefile(
lib,
out_file,
meters_per_unit=info['meters_per_unit'],
@ -458,7 +465,7 @@ def test_gdsii_lazy_overlay_merge_and_write(tmp_path: Path) -> None:
overlay.move_references('leaf', renamed_leaf)
out_file = tmp_path / 'overlay_out.gds'
gdsii_lazy_arrow.writefile(overlay, out_file)
gdsii.writefile(overlay, out_file)
roundtrip, _ = gdsii.readfile(out_file)
assert set(roundtrip.keys()) == {'leaf', renamed_leaf, 'top_a', 'top_b'}

View file

@ -3,7 +3,7 @@ import json
from pathlib import Path
from ..file import gdsii
from ..file.gdsii.perf import fixture_manifest, write_fixture
from tools.generate_gds_perf import fixture_manifest, write_fixture
def test_gdsii_perf_fixture_smoke(tmp_path: Path) -> None: