Compare commits

..

10 commits

36 changed files with 2137 additions and 1631 deletions

View file

@ -1,4 +1,4 @@
from masque.file.gdsii_perf import main
from masque.file.gdsii.perf import main
if __name__ == '__main__':

View file

@ -12,7 +12,7 @@ from masque import LibraryError
READERS: dict[str, tuple[str, tuple[str, ...]]] = {
'gdsii': ('masque.file.gdsii', ('readfile',)),
'gdsii_arrow': ('masque.file.gdsii_arrow', ('readfile', 'arrow_import', 'arrow_convert')),
'gdsii_arrow': ('masque.file.gdsii.arrow', ('readfile', 'arrow_import', 'arrow_convert')),
}

View file

@ -29,7 +29,7 @@ Contents
* Define a custom `Tool` that exposes primitive routing offers
* Use primitive offers to automatically transition between path types
- [renderpather](renderpather.py)
* Use `Pather(auto_render=False)` and `PathTool` to build a layout similar to the one in [pather](pather.py),
* Use `Pather(render='deferred')` and `PathTool` to build a layout similar to the one in [pather](pather.py),
but using `Path` shapes instead of `Polygon`s.
- [port_pather](port_pather.py)
* Use `PortPather` and the `.at()` syntax for more concise routing

View file

@ -13,7 +13,7 @@ from pprint import pformat
from masque import BuildLibrary, Pather, Pattern, cell
from masque.file.gdsii import writefile
from masque.file.gdsii_lazy import readfile
from masque.file.gdsii.lazy import readfile
import basic_shapes
import devices

View file

@ -13,7 +13,7 @@ def main() -> None:
library, M1_tool, M2_tool = prepare_tools()
# Create a deferred Pather and place some initial pads (same as Pather tutorial)
rpather = Pather(library, tools=M2_tool, auto_render=False)
rpather = Pather(library, tools=M2_tool, render='deferred')
rpather.place('pad', offset=(18_000, 30_000), port_map={'wire_port': 'VCC'})
rpather.place('pad', offset=(18_000, 60_000), port_map={'wire_port': 'GND'})
@ -157,7 +157,7 @@ def main() -> None:
#
# Rendering and Saving
#
# Since we deferred auto-rendering, we must call .render() to generate the geometry.
# Since routing is deferred, we must call .render() to generate the geometry.
rpather.render()
library['PortPather_Tutorial'] = rpather.pattern

View file

@ -13,7 +13,7 @@ def main() -> None:
#
# To illustrate deferred routing with `Pather`, we use `PathTool` instead
# of `AutoTool`. `PathTool` lacks some sophistication (e.g. no automatic transitions)
# but when used with `Pather(auto_render=False)`, it can consolidate multiple routing steps into
# but when used with `Pather(render='deferred')`, it can consolidate multiple routing steps into
# a single `Path` shape.
#
# We'll try to nearly replicate the layout from the `Pather` tutorial; see `pather.py`
@ -39,7 +39,7 @@ def main() -> None:
# and what port type to present.
M1_ptool = PathTool(layer='M1', width=M1_WIDTH, ptype='m1wire')
M2_ptool = PathTool(layer='M2', width=M2_WIDTH, ptype='m2wire')
rpather = Pather(tools=M2_ptool, library=library, auto_render=False)
rpather = Pather(tools=M2_ptool, library=library, render='deferred')
# As in the pather tutorial, we make some pads and labels...
rpather.place('pad', offset=(18_000, 30_000), port_map={'wire_port': 'VCC'})

View file

@ -63,6 +63,8 @@ from .library import (
ILibrary as ILibrary,
LibraryView as LibraryView,
Library as Library,
OverlayLibrary as OverlayLibrary,
PortsLibraryView as PortsLibraryView,
BuildLibrary as BuildLibrary,
BuildReport as BuildReport,
CellProvenance as CellProvenance,

View file

@ -3,7 +3,7 @@ Unified Pattern assembly and routing (`Pather`).
`Pather` is the public object that owns layout state. It holds the working
Pattern, Library, per-port Tool assignments, pending `RenderStep`s, and routing
side effects such as plug consumption, port renames, and auto-render. The
side effects such as plug consumption, port renames, and automatic rendering. The
planner package is intentionally internal: custom route generators should
extend `Tool.primitive_offers()` and `Tool.render()` rather than depending on
planner classes or search details.
@ -16,7 +16,7 @@ Routing is split into four ownership phases:
`RenderStep.data`, and returns prepared actions,
- application: `Pather` appends the prepared steps to its pending queue,
replaces live output ports, consumes plug destinations, applies deferred
trace-thru renames, and batches auto-render around the whole route,
trace-thru renames, and batches immediate rendering around the whole route,
- rendering: pending steps are grouped by live port, Tool, and continuity before
`Tool.render()` builds geometry for each compatible batch.
@ -27,7 +27,7 @@ prepared actions are applied, later commit, plug, rename, render, or insertion
failures may leave partial output; that mutation boundary belongs to Pather,
not to Tool implementations or the route solver.
"""
from typing import Self, Any, overload
from typing import Self, Any, Literal, overload
from collections.abc import Iterator, Iterable, Mapping, MutableMapping, Sequence
import copy
import logging
@ -36,6 +36,7 @@ from functools import wraps
from pprint import pformat
from contextlib import contextmanager
from itertools import chain
from types import TracebackType
import numpy
from numpy import pi
@ -64,6 +65,8 @@ from .logging import PatherLogger
logger = logging.getLogger(__name__)
RenderPolicy = Literal['auto', 'immediate', 'deferred', 'warn', 'error', 'ignore']
RENDER_POLICIES: tuple[RenderPolicy, ...] = ('auto', 'immediate', 'deferred', 'warn', 'error', 'ignore')
class Pather(PortList):
@ -76,9 +79,9 @@ class Pather(PortList):
pattern, and a set of `Tool`s for generating routing segments.
Routing operations (`trace`, `jog`, `uturn`, etc.) select primitive offers
from the active `Tool` and compose them into deferred `RenderStep`s.
Set `auto_render=False` to defer geometry generation until an explicit call
to `render()`.
from the active `Tool` and compose them into `RenderStep`s. By default,
geometry is rendered after each route, unless the `Pather` is used as a
context manager.
Examples: Creating a Pather
===========================
@ -94,12 +97,12 @@ class Pather(PortList):
connects port 'A' of the current pattern to port 'C' of `subdevice`.
- `pather.trace('my_port', ccw=True, length=100)` plans a 100-unit bend
starting at 'my_port'. Geometry is added immediately by default.
Set `auto_render=False` to defer and call `pather.render()` later.
starting at 'my_port'. If the `Pather` is used as a context manager,
geometry is generated on clean context exit.
"""
__slots__ = (
'pattern', 'library', 'tools', 'planner', '_paths',
'_dead', '_logger', '_auto_render', '_auto_render_append'
'_dead', '_logger', '_render_policy', '_render_append', '_context_depth'
)
pattern: Pattern
@ -128,8 +131,14 @@ class Pather(PortList):
_logger: PatherLogger
""" Handles diagnostic logging of operations """
_auto_render: bool
""" If True, routing operations call render() immediately """
_render_policy: RenderPolicy
""" Routing render behavior """
_render_append: bool
""" If True, automatic render calls append geometry instead of adding references """
_context_depth: int
""" Number of active context-manager entries """
_paths: defaultdict[str, list[RenderStep]]
""" Per-port pending render steps, consumed by `render()` """
@ -168,8 +177,8 @@ class Pather(PortList):
tools: Tool | MutableMapping[str | None, Tool] | None = None,
name: str | None = None,
debug: bool = False,
auto_render: bool = True,
auto_render_append: bool = True,
render: RenderPolicy = 'auto',
render_append: bool = True,
planner: RoutingPlanner | None = None,
) -> None:
"""
@ -181,16 +190,26 @@ class Pather(PortList):
tools: Tool(s) to use for routing segments.
name: If specified, `library[name]` is set to `self.pattern`.
debug: If True, enables detailed logging.
auto_render: If True, enables immediate rendering of routing steps.
auto_render_append: If `auto_render` is True, determines whether
to append geometry or add a reference.
render: Routing render policy. `'auto'` renders after each route
outside a context manager and defers until clean context exit
inside one. Use `'immediate'` to always render after each route,
`'deferred'` to keep paths pending until `render()` or clean
context exit, `'warn'` to log pending paths on clean context
exit, `'error'` to reject pending paths on clean context exit,
or `'ignore'` to leave pending paths silent.
render_append: If an automatic render is triggered, determines
whether to append geometry or add a reference.
planner: Optional stateless route-selection planner. If omitted,
a new `RoutingPlanner` is used.
"""
if render not in RENDER_POLICIES:
raise BuildError(f'Invalid render policy {render!r}; expected one of {RENDER_POLICIES}')
self._dead = False
self._logger = PatherLogger(debug=debug)
self._auto_render = auto_render
self._auto_render_append = auto_render_append
self._render_policy = render
self._render_append = render_append
self._context_depth = 0
self.library = library
self.pattern = pattern if pattern is not None else Pattern()
self.planner = RoutingPlanner() if planner is None else planner
@ -213,9 +232,40 @@ class Pather(PortList):
if name is not None:
library[name] = self.pattern
def __del__(self) -> None:
if any(self._paths.values()):
logger.warning(f'Pather {self} had pending render steps', stack_info=True)
def __enter__(self) -> Self:
self._context_depth += 1
return self
def __exit__(
self,
exc_type: type[BaseException] | None,
exc_value: BaseException | None,
traceback: TracebackType | None,
) -> bool:
_ = exc_value, traceback
self._context_depth -= 1
if exc_type is not None or self._context_depth != 0 or not any(self._paths.values()):
return False
if self._render_policy in ('auto', 'deferred'):
self.render(append=self._render_append)
elif self._render_policy == 'warn':
logger.warning(
'Pather context exited with %s; call render() or use render="deferred"',
self._pending_render_summary(),
)
elif self._render_policy == 'error':
raise BuildError(f'Pather context exited with {self._pending_render_summary()}')
return False
def _pending_render_summary(self) -> str:
ports = [(portspec, len(steps)) for portspec, steps in self._paths.items() if steps]
port_count = len(ports)
step_count = sum(count for _portspec, count in ports)
return (
f'{step_count} pending render step{"s" if step_count != 1 else ""} '
f'on {port_count} port{"s" if port_count != 1 else ""}'
)
def __repr__(self) -> str:
s = f'<Pather {self.pattern} L({len(self.library)}) {pformat(self.tools)}>'
@ -397,12 +447,9 @@ class Pather(PortList):
Apply every action and deferred rename in a prepared route result.
Route actions may contain several primitive render steps and port
mutations. Temporarily disabling auto-render prevents each step from
becoming its own rendered cell.
mutations. Immediate rendering happens once after the whole prepared
result has been applied.
"""
saved_auto_render = self._auto_render
self._auto_render = False
try:
for action in result.actions:
if not action.render_steps:
raise BuildError('Prepared route action has no render steps')
@ -416,11 +463,12 @@ class Pather(PortList):
self.plugged({action.portspec: action.plug_into})
for old_name, new_name in result.renames:
self.rename_ports({old_name: new_name})
self._auto_render = saved_auto_render
if saved_auto_render and any(self._paths.values()):
self.render(append = self._auto_render_append)
finally:
self._auto_render = saved_auto_render
render_immediately = (
self._render_policy == 'immediate'
or (self._render_policy == 'auto' and self._context_depth == 0)
)
if render_immediately and any(self._paths.values()):
self.render(append=self._render_append)
def _apply_dead_fallback(
self,

View file

@ -24,7 +24,7 @@ from __future__ import annotations
from collections.abc import Iterable, Mapping, Sequence
from dataclasses import dataclass
from itertools import combinations
from math import isclose as math_isclose
from math import cos, isclose as math_isclose, sin
from typing import Any, Literal
import numpy
@ -33,7 +33,7 @@ from numpy.typing import ArrayLike
from ...error import BuildError, PortError
from ...ports import Port
from ...utils import PTypeMatch, SupportsBool, ptype_match, ptypes_compatible, rotation_matrix_2d
from ...utils import PTypeMatch, SupportsBool, ptype_match, ptypes_compatible
from ..tools import (
BendOffer,
PrimitiveKind,
@ -50,6 +50,7 @@ from .interface import (
PreparedRouteResult,
RoutePlanningError,
RoutePortContext,
route_error_is_fatal,
)
@ -288,16 +289,28 @@ class Solver:
def __init__(self, request: RouteRequest) -> None:
self.request = request
self.eval_cache: dict[tuple[int, float, str | None, str | None, str, str, PrimitiveKind | None], SelectedPrimitive] = {}
self.offer_cache: dict[
tuple[PrimitiveKind, str | None, str | None, tuple[tuple[str, Any], ...]],
tuple[PrimitiveOffer, ...],
] = {}
self.seen_candidate_keys: set[tuple[Any, ...]] = set()
self.order = 0
def solve(self) -> Candidate:
"""
Enumerate, finalize, deduplicate, and rank legal candidates.
@staticmethod
def route_bend_count(steps: Sequence[SelectedPrimitive]) -> int:
"""Return the route bend budget consumed by non-adapter primitives."""
count = 0
for step in steps:
if step.role == 'adapter':
continue
if step.route_kind == 'bend':
count += 1
elif step.route_kind in ('s', 'u'):
count += 2
return count
Non-fatal candidate errors are accumulated so the failure message can
preserve useful Tool feedback. Fatal offer-contract errors stop the
solve immediately.
"""
def candidate_key(self, candidate: Candidate) -> tuple[Any, ...]:
"""Return a deterministic key for duplicate solved candidates."""
def endpoint_key(port: Port) -> tuple[float, float, float | None, str | None]:
return (
round(float(port.x), 9),
@ -306,7 +319,6 @@ class Solver:
port.ptype,
)
def candidate_key(candidate: Candidate) -> tuple[Any, ...]:
def offer_key(offer: PrimitiveOffer) -> tuple[Any, ...]:
return (
type(offer).__qualname__,
@ -330,12 +342,28 @@ class Solver:
) for step in candidate.steps),
)
def solve(
self,
*,
min_bends: int = 0,
max_bends: int | None = None,
) -> Candidate:
"""
Enumerate, finalize, deduplicate, and rank legal candidates.
Non-fatal candidate errors are accumulated so the failure message can
preserve useful Tool feedback. Fatal offer-contract errors stop the
solve immediately.
"""
if max_bends is None:
max_bends = self.request.bend_budget
candidates: list[Candidate] = []
errors: list[Exception] = []
seen: set[tuple[Any, ...]] = set()
for steps in self.enumerate_grammar():
for steps in self.enumerate_grammar(max_bends):
if not steps:
continue
if self.route_bend_count(steps) < min_bends:
continue
if any(first.role == 'adapter' and second.role == 'adapter' for first, second in zip(steps, steps[1:], strict=False)):
continue
try:
@ -344,10 +372,10 @@ class Solver:
raise_if_fatal(err)
errors.append(err)
continue
key = candidate_key(candidate)
if key in seen:
key = self.candidate_key(candidate)
if key in self.seen_candidate_keys:
continue
seen.add(key)
self.seen_candidate_keys.add(key)
candidates.append(candidate)
if not candidates:
@ -384,8 +412,19 @@ class Solver:
kwargs.pop('out_ptype', None)
if extra:
kwargs.update(extra)
try:
cache_key = (kind, in_ptype, out_ptype, tuple(sorted(kwargs.items())))
hash(cache_key)
except TypeError:
return self.request.tool.primitive_offers(kind, in_ptype=in_ptype, out_ptype=out_ptype, **kwargs)
cached = self.offer_cache.get(cache_key)
if cached is not None:
return cached
offers = self.request.tool.primitive_offers(kind, in_ptype=in_ptype, out_ptype=out_ptype, **kwargs)
self.offer_cache[cache_key] = offers
return offers
def evaluate(
self,
offer: PrimitiveOffer,
@ -450,17 +489,21 @@ class Solver:
points back into the primitive, so each step advances orientation by
the primitive output rotation plus pi.
"""
offset = numpy.zeros(2)
x = 0.0
y = 0.0
angle = 0.0
ptype: str | None = None
for step in steps:
out_port = step.out_port
if out_port.rotation is None:
raise BuildError('Primitive endpoints must have rotation')
offset += rotation_matrix_2d(angle) @ out_port.offset
angle_cos = cos(angle)
angle_sin = sin(angle)
x += angle_cos * float(out_port.x) - angle_sin * float(out_port.y)
y += angle_sin * float(out_port.x) + angle_cos * float(out_port.y)
angle += out_port.rotation + pi
ptype = out_port.ptype
return Port(offset, rotation=angle - pi, ptype=ptype)
return Port((x, y), rotation=angle - pi, ptype=ptype)
def current_ptype(self, steps: Sequence[SelectedPrimitive]) -> str | None:
return self.request.in_ptype if not steps else self.compose_endpoint(steps).ptype
@ -656,13 +699,13 @@ class Solver:
options.extend((turn, 2) for turn in self.su_primitive_options(steps, 'u', jog))
return tuple(options)
def enumerate_grammar(self) -> Iterable[tuple[SelectedPrimitive, ...]]:
def enumerate_grammar(self, max_bends: int) -> Iterable[tuple[SelectedPrimitive, ...]]:
"""Yield raw primitive sequences allowed by the bounded route grammar."""
residual_jog = 0.0 if self.request.jog is None else float(self.request.jog)
base: tuple[SelectedPrimitive, ...] = ()
for prefix_adapter in self.adapter_options(base, residual_jog=residual_jog):
prefix = (*base, *prefix_adapter)
yield from self.enumerate_segments(prefix, self.request.bend_budget, residual_jog=residual_jog)
yield from self.enumerate_segments(prefix, max_bends, residual_jog=residual_jog)
def enumerate_segments(
self,
@ -814,6 +857,22 @@ class Solver:
return False
return self.request.out_ptype is None or ptypes_compatible(end_port.ptype, self.request.out_ptype)
def rotation_matches_request(self, steps: Sequence[SelectedPrimitive]) -> bool:
"""
Return true when a sequence can produce the requested output rotation.
Primitive parameters do not affect the rotation contract, so
rotation-impossible candidates can be rejected before parameter solving.
"""
angle = 0.0
for step in steps:
step_rotation = step.out_port.rotation
if step_rotation is None:
raise BuildError('Primitive endpoints must have rotation')
angle += float(step_rotation) + pi
rotation_delta = ((angle - pi) - self.request.out_rotation) % (2 * pi)
return is_close(rotation_delta, 0) or is_close(rotation_delta, 2 * pi)
def finalize(self, steps: Sequence[SelectedPrimitive]) -> Candidate:
"""
Try all small solve sets for one raw sequence and return the first match.
@ -831,6 +890,8 @@ class Solver:
raise BuildError(f'{self.request.route_name} route requires a jog constraint')
constraints.append(('y', float(self.request.jog)))
route_constraints = tuple(constraints)
if not self.rotation_matches_request(steps):
raise BuildError(f'{self.request.route_name} composed primitive route is unsupported')
adjustable = self.adjustable_indices(steps)
solve_sets: list[tuple[int, ...]] = [()]
max_solve = min(len(route_constraints), len(adjustable))
@ -889,6 +950,64 @@ class RoutingPlanner:
TRACE_INTO_MAX_BENDS: int = 4
def trace_into_bend_bands(self, family: PrimitiveKind) -> tuple[tuple[int, int], ...]:
"""Return non-overlapping trace_into bend-budget bands for staged solving."""
max_bends = self.TRACE_INTO_MAX_BENDS
if family == 'bend':
return tuple(band for band in ((1, 1), (3, 3)) if band[1] <= max_bends)
bands: list[tuple[int, int]] = []
if max_bends >= 0:
bands.append((0, 2 if max_bends >= 2 else 0))
if max_bends >= 4:
bands.append((4, 4))
return tuple(bands)
def route_request(
self,
family: PrimitiveKind,
context: RoutePortContext,
*,
length: float | None = None,
jog: float | None = None,
ccw: SupportsBool | None = None,
constrain_jog: bool = False,
max_bends: int | None = None,
**kwargs: Any,
) -> RouteRequest:
"""Build normalized solver input for one route leg."""
return RouteRequest(
family=family,
tool=context.tool,
in_ptype=context.port.ptype,
route_kwargs=kwargs,
length=length,
jog=jog,
ccw=ccw,
out_ptype=kwargs.get('out_ptype'),
constrain_jog=constrain_jog,
max_bends=max_bends,
)
def solver_for_request(self, request: RouteRequest) -> Solver:
"""Construct the solver for a route request."""
return Solver(request)
def route_leg_from_candidate(
self,
context: RoutePortContext,
candidate: Candidate,
*,
plug_into: str | None,
) -> RouteLeg:
"""Attach a solved candidate to its source Pather context."""
return RouteLeg(
portspec=context.portspec,
start_port=context.port.copy(),
tool=context.tool,
candidate=candidate,
plug_into=plug_into,
)
def plan_leg(
self,
family: PrimitiveKind,
@ -903,31 +1022,23 @@ class RoutingPlanner:
**kwargs: Any,
) -> RouteLeg:
"""Solve one route leg and attach it to its source Pather context."""
request = RouteRequest(
request = self.route_request(
family=family,
tool=context.tool,
in_ptype=context.port.ptype,
route_kwargs=kwargs,
context=context,
length=length,
jog=jog,
ccw=ccw,
out_ptype=kwargs.get('out_ptype'),
constrain_jog=constrain_jog,
max_bends=max_bends,
**kwargs,
)
try:
candidate = Solver(request).solve()
candidate = self.solver_for_request(request).solve()
except BuildError as err:
if family == 'u' and length is None and not getattr(err, 'fatal', False):
raise BuildError('No legal primitive offer for omitted-length U-turn') from err
raise
return RouteLeg(
portspec=context.portspec,
start_port=context.port.copy(),
tool=context.tool,
candidate=candidate,
plug_into=plug_into,
)
return self.route_leg_from_candidate(context, candidate, plug_into=plug_into)
def prepared_route_action_from_leg(
self,
@ -1178,17 +1289,36 @@ class RoutingPlanner:
desired.rotation = port_dst.rotation - pi
desired.ptype = out_ptype
family, length, jog, ccw = self.trace_into_spec(context_src.port, desired)
leg = self.plan_leg(
request = self.route_request(
family,
context_src,
length=length,
jog=jog,
ccw=ccw,
plug_into=portspec_dst if plug_destination else None,
constrain_jog=family == 'bend',
max_bends=self.TRACE_INTO_MAX_BENDS,
**(dict(kwargs) | {'out_ptype': out_ptype}),
)
solver = self.solver_for_request(request)
candidate = None
last_error: Exception | None = None
for min_bends, max_bends in self.trace_into_bend_bands(family):
try:
candidate = solver.solve(min_bends=min_bends, max_bends=max_bends)
break
except (BuildError, NotImplementedError) as err:
if route_error_is_fatal(err):
raise
last_error = err
if candidate is None:
if last_error is not None:
raise last_error
raise BuildError('No legal primitive offer for trace_into route')
leg = self.route_leg_from_candidate(
context_src,
candidate,
plug_into=portspec_dst if plug_destination else None,
)
renames = ((thru, context_src.portspec),) if thru is not None else ()
return self.prepared_result_from_legs(
(leg,),

View file

@ -560,7 +560,7 @@ class RenderStep:
"""
A single deferred routing operation.
`Pather(auto_render=False)` stores these records while routing and later
`Pather(render='deferred')` stores these records while routing and later
passes batches of compatible steps to `Tool.render()` when `Pather.render()`
is called.
"""

View file

@ -0,0 +1,10 @@
"""
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

View file

@ -39,16 +39,16 @@ from pprint import pformat
from klamath.basic import KlamathError
import numpy
from numpy.typing import ArrayLike, NDArray
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 ..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 ..utils import is_gzipped, tmpfile
from ... import Pattern, Ref, PatternError, LibraryError, Label, Shape
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
logger = logging.getLogger(__name__)
@ -80,10 +80,6 @@ path_cap_map = {
}
def rint_cast(val: ArrayLike) -> NDArray[numpy.int32]:
return numpy.rint(val).astype(numpy.int32)
def _packed_layer_u32_to_pairs(values: NDArray[numpy.unsignedinteger[Any]]) -> NDArray[numpy.int16]:
layer = (values >> numpy.uint32(16)).astype(numpy.uint16).view(numpy.int16)
dtype = (values & numpy.uint32(0xffff)).astype(numpy.uint16).view(numpy.int16)
@ -136,7 +132,7 @@ def _installed_library_candidates() -> list[pathlib.Path]:
def _repo_library_candidates() -> list[pathlib.Path]:
repo_root = pathlib.Path(__file__).resolve().parents[2]
repo_root = pathlib.Path(__file__).resolve().parents[3]
library_name = _local_library_filename()
return [
repo_root / 'klamath-rs' / 'target' / 'release' / library_name,
@ -167,7 +163,7 @@ def is_available() -> bool:
def _get_clib() -> Any:
global clib
global clib # noqa: PLW0603
if clib is None:
lib_path = find_klamath_rs_library()
if lib_path is None:

View file

@ -33,12 +33,12 @@ from numpy.typing import ArrayLike, NDArray
import klamath
from klamath import records
from .utils import is_gzipped, tmpfile
from .. import Pattern, Ref, PatternError, LibraryError, Label, Shape
from ..shapes import Polygon, Path, RectCollection
from ..repetition import Grid
from ..utils import layer_t, annotations_t
from ..library import Library, ILibrary
from ..utils import is_gzipped, tmpfile
from ... import Pattern, Ref, PatternError, LibraryError, Label, Shape
from ...shapes import Polygon, Path, RectCollection
from ...repetition import Grid
from ...utils import layer_t, annotations_t
from ...library import Library, ILibrary
logger = logging.getLogger(__name__)

View file

@ -8,18 +8,17 @@ This module provides the non-Arrow half of Masque's lazy GDS architecture:
- cells are materialized on demand through the classic `gdsii` decoder
whenever a caller indexes the lazy view
- the source can be wrapped in `PortsLibraryView` or merged through
`OverlayLibrary`, both of which live in `gdsii_lazy_core`
`OverlayLibrary`
The public surface intentionally parallels `gdsii_lazy_arrow` closely so that
The public surface intentionally parallels `gdsii.lazy_arrow` closely so that
callers can swap between the classic and Arrow-backed implementations with
minimal changes.
"""
from __future__ import annotations
from dataclasses import dataclass
from typing import IO, Any, cast
from typing import IO, TYPE_CHECKING, Any, cast
from collections import defaultdict
from collections.abc import Iterator, Sequence
import gzip
import io
import logging
@ -28,16 +27,26 @@ import pathlib
import klamath
import numpy
from numpy.typing import NDArray
from klamath import records
from . import gdsii
from .utils import is_gzipped
from .gdsii_lazy_core import OverlayLibrary, PortsLibraryView, _pattern_children, write, writefile
from ..error import LibraryError
from ..library import ILibraryView, LibraryView, dangling_mode_t
from ..pattern import Pattern
from ..utils import apply_transforms
from . import klamath as gdsii
from .lazy_write import write as write, writefile as writefile
from ..utils import is_gzipped
from ...error import LibraryError
from ...library import (
ILibraryView,
LibraryView,
PortsLibraryView,
dangling_mode_t,
)
from ...utils import apply_transforms
if TYPE_CHECKING:
from collections.abc import Iterator, Sequence
from numpy.typing import NDArray
from ...pattern import Pattern
logger = logging.getLogger(__name__)
@ -76,7 +85,7 @@ def _open_source_stream(
with gzip.open(path, mode='rb') as stream:
data = stream.read()
return _SourceHandle(path=path, stream=io.BytesIO(data))
stream = cast('IO[bytes]', gzip.open(path, mode='rb'))
stream = cast('IO[bytes]', gzip.open(path, mode='rb')) # noqa: SIM115
return _SourceHandle(path=path, stream=stream)
if use_mmap:
@ -216,7 +225,7 @@ class GdsLibrarySource(ILibraryView):
graph: dict[str, set[str]] = {}
for name in self._cell_order:
if name in self._cache:
graph[name] = _pattern_children(self._cache[name])
graph[name] = {child for child, refs in self._cache[name].refs.items() if child is not None and refs}
else:
graph[name] = self._raw_children(name)

View file

@ -1,30 +1,39 @@
"""
Lazy GDSII readers and writers backed by native Arrow scan/materialize paths.
This module is intentionally separate from `gdsii_arrow` so the eager read path
This module is intentionally separate from `gdsii.arrow` so the eager read path
keeps its current behavior and performance profile.
"""
from __future__ import annotations
from dataclasses import dataclass
from typing import IO, Any, cast
from typing import IO, TYPE_CHECKING, Any, cast
from collections import defaultdict
from collections.abc import Iterator, Sequence
import gzip
import logging
import mmap
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,
LibraryView,
PortsLibraryView,
dangling_mode_t,
)
from ...utils import apply_transforms
if TYPE_CHECKING:
from collections.abc import Iterator, Sequence
from numpy.typing import NDArray
import pyarrow
from . import gdsii_arrow
from .utils import is_gzipped
from .gdsii_lazy_core import OverlayLibrary, PortsLibraryView, _pattern_children, write, writefile
from ..library import ILibraryView, LibraryView, dangling_mode_t
from ..pattern import Pattern
from ..utils import apply_transforms
from ...pattern import Pattern
logger = logging.getLogger(__name__)
@ -78,11 +87,7 @@ class _ScanPayload:
refs: _ScanRefs
def is_available() -> bool:
return gdsii_arrow.is_available()
def _read_header(libarr: pyarrow.StructScalar) -> dict[str, Any]:
return gdsii_arrow._read_header(libarr)
return arrow.is_available()
def _open_source_buffer(path: pathlib.Path) -> _SourceBuffer:
@ -97,7 +102,7 @@ def _open_source_buffer(path: pathlib.Path) -> _SourceBuffer:
def _extract_scan_payload(libarr: pyarrow.StructScalar) -> _ScanPayload:
library_info = _read_header(libarr)
library_info = arrow._read_header(libarr)
cell_names = libarr['cell_names'].as_py()
cells = libarr['cells']
@ -110,10 +115,10 @@ def _extract_scan_payload(libarr: pyarrow.StructScalar) -> _ScanPayload:
ref_values = refs.values
ref_offsets = refs.offsets.to_numpy()
targets = ref_values.field('target').to_numpy()
xy = gdsii_arrow._packed_xy_u64_to_pairs(ref_values.field('xy').to_numpy())
xy0 = gdsii_arrow._packed_xy_u64_to_pairs(ref_values.field('xy0').to_numpy())
xy1 = gdsii_arrow._packed_xy_u64_to_pairs(ref_values.field('xy1').to_numpy())
counts = gdsii_arrow._packed_counts_u32_to_pairs(ref_values.field('counts').to_numpy())
xy = arrow._packed_xy_u64_to_pairs(ref_values.field('xy').to_numpy())
xy0 = arrow._packed_xy_u64_to_pairs(ref_values.field('xy0').to_numpy())
xy1 = arrow._packed_xy_u64_to_pairs(ref_values.field('xy1').to_numpy())
counts = arrow._packed_counts_u32_to_pairs(ref_values.field('counts').to_numpy())
invert_y = ref_values.field('invert_y').to_numpy(zero_copy_only=False)
angle_rad = ref_values.field('angle_rad').to_numpy()
scale = ref_values.field('scale').to_numpy()
@ -219,7 +224,7 @@ class ArrowLibrary(ILibraryView):
def from_file(cls, filename: str | pathlib.Path) -> ArrowLibrary:
path = pathlib.Path(filename).expanduser().resolve()
source = _open_source_buffer(path)
scan_arr = gdsii_arrow._scan_buffer_to_arrow(source.data)
scan_arr = arrow._scan_buffer_to_arrow(source.data)
assert len(scan_arr) == 1
payload = _extract_scan_payload(scan_arr[0])
return cls(path=path, payload=payload, source=source)
@ -279,9 +284,9 @@ class ArrowLibrary(ILibraryView):
],
dtype=numpy.uint64,
)
arrow_arr = gdsii_arrow._read_selected_cells_to_arrow(self._source.data, ranges)
arrow_arr = arrow._read_selected_cells_to_arrow(self._source.data, ranges)
assert len(arrow_arr) == 1
selected_lib, _info = gdsii_arrow.read_arrow(arrow_arr[0])
selected_lib, _info = arrow.read_arrow(arrow_arr[0])
for name in uncached:
pat = selected_lib[name]
materialized[name] = pat
@ -343,7 +348,7 @@ class ArrowLibrary(ILibraryView):
graph: dict[str, set[str]] = {}
for name in self._payload.cell_order:
if name in self._cache:
graph[name] = _pattern_children(self._cache[name])
graph[name] = {child for child, refs in self._cache[name].refs.items() if child is not None and refs}
else:
graph[name] = self._raw_children(name)

View file

@ -0,0 +1,170 @@
"""
GDS write helpers for source-backed lazy GDS views.
The generic mutable overlay and ports-importing view live in `masque.library`.
This module preserves source-backed GDS copy-through behavior where possible,
falling back to normal pattern serialization when a cell has been materialized
or remapped.
"""
from __future__ import annotations
from typing import IO, TYPE_CHECKING, Any, cast
import gzip
import logging
import pathlib
import klamath
from . import klamath as gdsii
from ..utils import tmpfile
from ...error import LibraryError
from ...library import ILibraryView, OverlayLibrary
from ...library.overlay import _SourceEntry, _materialize_detached_pattern
if TYPE_CHECKING:
from collections.abc import Mapping
from ...pattern import Pattern
logger = logging.getLogger(__name__)
def _get_write_info(
library: Mapping[str, Pattern] | ILibraryView,
*,
meters_per_unit: float | None,
logical_units_per_unit: float | None,
library_name: str | None,
) -> tuple[float, float, str]:
if meters_per_unit is not None and logical_units_per_unit is not None and library_name is not None:
return meters_per_unit, logical_units_per_unit, library_name
infos: list[dict[str, Any]] = []
stack: list[Mapping[str, Pattern] | ILibraryView] = [library]
while stack:
current = stack.pop()
info = getattr(current, 'library_info', None)
if isinstance(info, dict):
infos.append(info)
if isinstance(current, OverlayLibrary):
stack.extend(reversed([layer.library for layer in current._layers]))
if infos:
unit_pairs = {(info['meters_per_unit'], info['logical_units_per_unit']) for info in infos}
if len(unit_pairs) > 1:
raise LibraryError('Merged lazy GDS sources must have identical units before writing')
info = infos[0]
meters = info['meters_per_unit'] if meters_per_unit is None else meters_per_unit
logical = info['logical_units_per_unit'] if logical_units_per_unit is None else logical_units_per_unit
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
def _can_copy_raw_cell(library: Mapping[str, Pattern] | ILibraryView, name: str) -> bool:
can_copy = getattr(library, 'can_copy_raw_struct', None)
if not callable(can_copy):
return False
return bool(can_copy(name))
def _raw_struct_bytes(library: Mapping[str, Pattern] | ILibraryView, name: str) -> bytes:
reader = getattr(library, 'raw_struct_bytes', None)
if not callable(reader):
raise TypeError('raw_struct_bytes')
return cast('bytes', reader(name))
def _write_pattern_struct(stream: IO[bytes], name: str, pat: Pattern) -> None:
elements: list[klamath.elements.Element] = []
elements += gdsii._shapes_to_elements(pat.shapes)
elements += gdsii._labels_to_texts(pat.labels)
elements += gdsii._mrefs_to_grefs(pat.refs)
klamath.library.write_struct(stream, name=name.encode('ASCII'), elements=elements)
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:
meters_per_unit, logical_units_per_unit, library_name = _get_write_info(
library,
meters_per_unit=meters_per_unit,
logical_units_per_unit=logical_units_per_unit,
library_name=library_name,
)
header = klamath.library.FileHeader(
name=library_name.encode('ASCII'),
user_units_per_db_unit=logical_units_per_unit,
meters_per_db_unit=meters_per_unit,
)
header.write(stream)
if isinstance(library, OverlayLibrary):
for name in library.source_order():
entry = library._entries[name]
can_copy_overlay = False
if isinstance(entry, _SourceEntry) and name == entry.source_name:
layer = library._layers[entry.layer_index]
children = layer.child_graph.get(entry.source_name, set())
can_copy_overlay = (
_can_copy_raw_cell(layer.library, entry.source_name)
and all(library._effective_target(layer, child) == child for child in children)
)
if can_copy_overlay:
stream.write(_raw_struct_bytes(layer.library, entry.source_name))
else:
_write_pattern_struct(stream, name, library._materialize_pattern(name, persist=False))
klamath.records.ENDLIB.write(stream, None)
return
if hasattr(library, 'raw_struct_bytes'):
for name in library.source_order():
if _can_copy_raw_cell(library, name):
stream.write(_raw_struct_bytes(library, name))
else:
_write_pattern_struct(stream, name, _materialize_detached_pattern(cast('ILibraryView', library), name))
klamath.records.ENDLIB.write(stream, None)
return
gdsii.write(cast('Mapping[str, Pattern]', library), stream, meters_per_unit, logical_units_per_unit, library_name)
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:
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:
write(
library,
stream,
meters_per_unit=meters_per_unit,
logical_units_per_unit=logical_units_per_unit,
library_name=library_name,
)
finally:
for ss in streams:
ss.close()

View file

@ -244,7 +244,7 @@ def _annotation(index: int) -> dict[int, bytes]:
return {1: f'perf-{index}'.encode('ASCII')}
def _make_box_cell(name: str, index: int, cfg: FixturePreset) -> list[elements.Element]:
def _make_box_cell(index: int, cfg: FixturePreset) -> list[elements.Element]:
cell_elements: list[elements.Element] = []
xbase = (index % 17) * 600
ybase = (index // 17) * 180
@ -277,7 +277,7 @@ def _make_box_cell(name: str, index: int, cfg: FixturePreset) -> list[elements.E
return cell_elements
def _make_poly_cell(name: str, index: int, cfg: FixturePreset) -> list[elements.Element]:
def _make_poly_cell(index: int, cfg: FixturePreset) -> list[elements.Element]:
cell_elements: list[elements.Element] = []
xbase = (index % 19) * 900
ybase = (index // 19) * 260
@ -368,12 +368,12 @@ def _poly_cluster_name(index: int) -> str:
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(_box_name(idx), idx, cfg))
_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(_poly_name(idx), idx, cfg))
_write_struct(stream, _poly_name(idx), _make_poly_cell(idx, cfg))
def _write_wrappers(stream: Any, cfg: FixturePreset) -> None:
@ -487,38 +487,31 @@ def _write_top(stream: Any, cfg: FixturePreset) -> None:
_write_struct(stream, 'TOP', cell_elements)
def _poly_paths_total(cfg: FixturePreset) -> int:
return (cfg.poly_cells - 1) // cfg.path_stride + 1
def _poly_texts_total(cfg: FixturePreset) -> int:
return (cfg.poly_cells - 1) // cfg.text_stride + 1
def _ref_instances_per_box_cluster(cfg: FixturePreset) -> int:
array_refs = min(cfg.box_cluster_refs, max(1, (3 * cfg.box_cluster_refs) // 4))
array_mult = cfg.box_cluster_array[0] * cfg.box_cluster_array[1]
return array_refs * array_mult + (cfg.box_cluster_refs - array_refs)
def _ref_instances_per_poly_cluster(cfg: FixturePreset) -> int:
array_refs = min(cfg.poly_cluster_refs, cfg.poly_cluster_refs // 2)
array_mult = cfg.poly_cluster_array[0] * cfg.poly_cluster_array[1]
return array_refs * array_mult + (cfg.poly_cluster_refs - array_refs)
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 * _ref_instances_per_box_cluster(cfg)
+ 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 * _ref_instances_per_poly_cluster(cfg)
+ 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)
@ -545,8 +538,8 @@ def fixture_manifest(path: str | Path, preset: str, scale: float = 1.0) -> Fixtu
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=_poly_paths_total(cfg),
hierarchical_texts_total=_poly_texts_total(cfg),
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,

View file

@ -0,0 +1,29 @@
"""Library classes for managing name-to-pattern mappings."""
from .utils import (
SINGLE_USE_PREFIX as SINGLE_USE_PREFIX,
Tree as Tree,
TreeView as TreeView,
b64suffix as b64suffix,
dangling_mode_t as dangling_mode_t,
visitor_function_t as visitor_function_t,
)
from .base import (
AbstractView as AbstractView,
ILibrary as ILibrary,
ILibraryView as ILibraryView,
)
from .mapping import (
Library as Library,
LibraryView as LibraryView,
)
from .overlay import (
OverlayLibrary as OverlayLibrary,
PortsLibraryView as PortsLibraryView,
)
from .build import (
BuildLibrary as BuildLibrary,
BuildReport as BuildReport,
CellProvenance as CellProvenance,
cell as cell,
)
from .lazy import LazyLibrary as LazyLibrary

File diff suppressed because it is too large Load diff

742
masque/library/build.py Normal file
View file

@ -0,0 +1,742 @@
"""Two-phase build library implementation."""
from __future__ import annotations
from collections import defaultdict
from contextvars import ContextVar
from dataclasses import dataclass, replace
from functools import wraps
from pprint import pformat
from typing import TYPE_CHECKING, Any, Literal, Self, cast
import copy
from ..error import BuildError, LibraryError
from .base import ILibrary, ILibraryView
from .utils import TreeView, dangling_mode_t, _plan_source_names, _rename_patterns, _source_rename_map
from .mapping import Library, LibraryView
from .overlay import OverlayLibrary
if TYPE_CHECKING:
from collections.abc import Callable, Iterator, Mapping, Sequence
from ..abstract import Abstract
from ..pattern import Pattern
_ACTIVE_BUILD_SESSIONS: ContextVar[dict[int, _BuildSessionLibrary] | None] = ContextVar(
'masque_active_build_sessions',
default=None,
)
@dataclass(frozen=True)
class CellProvenance:
"""
Provenance record for one cell in a completed build output.
Each output name in a `BuildReport` maps to one `CellProvenance`. The
record captures both where the cell came from and how its visible name was
chosen.
Attributes:
requested_name: First name requested for this cell during the build.
kind: Whether the cell came from a declaration, helper emission, or an
imported source library.
owner_declared_name: Declared cell responsible for this output cell, if
any. Imported source cells leave this as `None`.
build_chain: Declared-cell dependency chain that was active when the
cell was emitted.
"""
requested_name: str
kind: Literal['declared', 'helper', 'source']
owner_declared_name: str | None
build_chain: tuple[str, ...]
@dataclass(frozen=True)
class BuildReport:
"""
Immutable summary of one `BuildLibrary.validate()` or `.build()` run.
The report is designed to answer two questions after a build completes:
which declared cells depended on which other declared cells, and where each
output cell came from.
Attributes:
requested_roots: Roots explicitly requested for the run. A full
`build()` uses all declared cells.
provenance: Mapping from final output name to provenance metadata.
dependency_graph: Declared-cell dependency graph discovered through
library-mediated reads and explicit recipe hints.
"""
requested_roots: tuple[str, ...]
provenance: Mapping[str, CellProvenance]
dependency_graph: Mapping[str, frozenset[str]]
@dataclass
class _BuildRecipe:
""" Captured deferred call to a pattern factory. """
func: Callable[..., Pattern]
args: tuple[Any, ...]
kwargs: dict[str, Any]
explicit_dependencies: tuple[str, ...] = ()
def depends_on(self, *names: str) -> _BuildRecipe:
self.explicit_dependencies += tuple(names)
return self
def cell(func: Callable[..., Pattern]) -> Callable[..., _BuildRecipe]:
"""
Wrap a plain pattern factory so calls return deferred build recipes.
Use as either `cell(fn)(...)` or `@cell`.
"""
@wraps(func)
def wrapper(*args: Any, **kwargs: Any) -> _BuildRecipe:
return _BuildRecipe(func=func, args=args, kwargs=kwargs)
return wrapper
class BuildCellsView:
"""
Attribute-based declaration namespace for `BuildLibrary`.
This is the ergonomic authoring surface exposed as `builder.cells`. It is
intentionally write-focused: attribute assignment and deletion register
declarations, while attribute reads fail with guidance to build first and
use the returned library.
"""
def __init__(self, library: BuildLibrary) -> None:
object.__setattr__(self, '_library', library)
def __getattr__(self, name: str) -> Pattern:
raise BuildError(
f'BuildLibrary.cells.{name} is write-only during authoring. '
'Call build() and index the returned library instead.'
)
def __setattr__(self, name: str, value: Pattern | _BuildRecipe) -> None:
if name.startswith('_'):
object.__setattr__(self, name, value)
return
self._library[name] = value
def __delattr__(self, name: str) -> None:
if name.startswith('_'):
raise AttributeError(name)
del self._library[name]
class BuildLibrary(ILibrary):
"""
Two-phase declaration surface for mixed imported/generated libraries.
A `BuildLibrary` collects three kinds of inputs:
- direct declared `Pattern` objects
- deferred recipes created with `cell(...)`
- imported source-backed library views added with `add_source(...)`
The builder itself is not a normal readable library during authoring.
Instead, `validate()` and `build()` create a temporary build-session library
that recipes can read from and write helper cells into while dependencies
are resolved. `build()` then freezes the builder on success and returns the
built library plus a `BuildReport`.
"""
def __init__(self) -> None:
self.cells = BuildCellsView(self)
self._frozen = False
self._declarations: dict[str, Pattern | _BuildRecipe] = {}
self._sources: list[tuple[ILibraryView, dict[str, str]]] = []
self._names: dict[str, None] = {}
def _active_session(self) -> _BuildSessionLibrary | None:
sessions = _ACTIVE_BUILD_SESSIONS.get()
if sessions is None:
return None
return sessions.get(id(self))
def _require_active_session(self, operation: str) -> _BuildSessionLibrary:
session = self._active_session()
if session is None:
raise BuildError(
f'BuildLibrary.{operation}() is only available while validate() or build() is running. '
'Use the built output library for reads.'
)
return session
def _assert_editable(self) -> None:
if self._frozen:
raise BuildError('This BuildLibrary has already been built successfully and is now frozen.')
def __iter__(self) -> Iterator[str]:
session = self._active_session()
if session is not None:
return iter(session)
return iter(self._names)
def __len__(self) -> int:
session = self._active_session()
if session is not None:
return len(session)
return len(self._names)
def __contains__(self, key: object) -> bool:
session = self._active_session()
if session is not None:
return key in session
return key in self._names
def __getitem__(self, key: str) -> Pattern:
return self._require_active_session('__getitem__')[key]
def __setitem__(
self,
key: str,
value: Pattern | _BuildRecipe,
) -> None:
session = self._active_session()
if session is not None:
session[key] = value
return
self._assert_editable()
if key in self._names:
raise LibraryError(f'"{key}" already exists in the builder. Overwriting is not allowed!')
if isinstance(value, _BuildRecipe):
declaration = value
else:
if callable(value):
raise TypeError('BuildLibrary recipes must be wrapped with cell(fn)(...) or @cell.')
declaration = value
self._declarations[key] = declaration
self._names[key] = None
def __delitem__(self, key: str) -> None:
session = self._active_session()
if session is not None:
del session[key]
return
self._assert_editable()
if key not in self._declarations:
raise KeyError(key)
del self._declarations[key]
del self._names[key]
def _merge(self, key_self: str, other: Mapping[str, Pattern], key_other: str) -> None:
session = self._active_session()
if session is not None:
session._merge(key_self, other, key_other)
return
self[key_self] = copy.deepcopy(other[key_other])
def add(
self,
other: Mapping[str, Pattern],
rename_theirs: Callable[[ILibraryView, str], str] = _rename_patterns,
mutate_other: bool = False,
) -> dict[str, str]:
from ..pattern import map_targets # noqa: PLC0415
session = self._active_session()
if session is not None:
return session.add(other, rename_theirs=rename_theirs, mutate_other=mutate_other)
self._assert_editable()
source_backed = isinstance(other, ILibraryView) and not isinstance(other, Library | LibraryView)
if source_backed:
if mutate_other:
raise BuildError('BuildLibrary.add(..., mutate_other=True) is not supported for source-backed inputs.')
return self.add_source(
other,
rename_theirs = rename_theirs,
rename_when = 'conflict',
)
source_order = tuple(other.keys())
source_to_visible = _plan_source_names(
self,
source_order,
self._names,
rename_theirs = rename_theirs,
rename_when = 'conflict',
)
rename_map = _source_rename_map(source_to_visible)
if mutate_other:
temp = other
else:
temp = Library(copy.deepcopy(dict(other)))
for source_name in source_order:
visible_name = source_to_visible[source_name]
pattern = temp[source_name]
if rename_map:
pattern.refs = map_targets(
pattern.refs,
lambda target: cast('dict[str | None, str | None]', rename_map).get(target, target),
)
self[visible_name] = pattern
return rename_map
def __lshift__(self, other: TreeView) -> str:
session = self._active_session()
if session is not None:
return session << other
self._assert_editable()
if len(other) == 1:
name = next(iter(other))
elif isinstance(other, ILibraryView) and not isinstance(other, Library | LibraryView):
source_order = other.source_order()
child_graph = other.child_graph(dangling='include')
referenced = set().union(*child_graph.values()) if child_graph else set()
tops = [candidate for candidate in source_order if candidate not in referenced]
if len(tops) != 1:
raise LibraryError(f'Asked for the single topcell, but found the following: {pformat(tops)}')
name = tops[0]
else:
return super().__lshift__(other)
rename_map = self.add(other)
return rename_map.get(name, name)
def __le__(self, other: Mapping[str, Pattern]) -> Abstract:
if self._active_session() is not None:
return super().__le__(other)
raise BuildError('BuildLibrary.__le__() is only available while validate() or build() is running.')
def rename(
self,
old_name: str,
new_name: str,
move_references: bool = False,
) -> Self:
"""
Rename a helper cell during an active build session.
During authoring, declared cells must be registered under their
intended final names and imported source cells must be renamed through
`add_source(...)`.
"""
session = self._active_session()
if session is not None:
session.rename(old_name, new_name, move_references=move_references)
return self
self._assert_editable()
if old_name == new_name:
return self
if old_name in self._declarations:
raise BuildError(
f'Cannot rename declared build cell "{old_name}" during authoring. '
'Register it under the intended final name instead.'
)
if old_name not in self._names:
raise LibraryError(f'"{old_name}" does not exist in the builder.')
raise BuildError(
f'Cannot rename imported source cell "{old_name}" during authoring. '
'Choose visible source names with add_source(..., rename_theirs=..., rename_when=...).'
)
def abstract(self, name: str) -> Abstract:
return self._require_active_session('abstract').abstract(name)
def resolve(
self,
other: Abstract | str | Pattern | TreeView,
append: bool = False,
) -> Abstract | Pattern:
return self._require_active_session('resolve').resolve(other, append=append)
def add_source(
self,
source: Mapping[str, Pattern] | ILibraryView,
*,
rename_theirs: Callable[[ILibraryView, str], str] | None = None,
rename_when: Literal['conflict', 'always'] = 'conflict',
) -> dict[str, str]:
"""
Register an imported source-backed library with the builder.
The source is not materialized immediately. Its names are scanned once
to reserve visible builder names, then the source is read again when a
build session starts. The source's cell membership must not be
structurally mutated between `add_source()` and `build()`/`validate()`.
Source cells may be renamed on entry to avoid collisions with existing
declarations or other imported sources.
Args:
rename_theirs: Function used to choose visible names for imported
source cells.
rename_when: If `'conflict'`, only conflicting names are renamed.
If `'always'`, every imported source name is passed through
`rename_theirs`.
Returns:
Mapping of `{source_name: visible_name}` for imported names that
were renamed while being added.
"""
if self._active_session() is not None:
raise BuildError('BuildLibrary.add_source() is only available while authoring, not during validate() or build().')
self._assert_editable()
view = source if isinstance(source, ILibraryView) else LibraryView(source)
source_order = tuple(view.source_order())
source_to_visible = _plan_source_names(
self,
source_order,
self._names,
rename_theirs = rename_theirs,
rename_when = rename_when,
)
self._sources.append((view, dict(source_to_visible)))
for source_name in source_order:
visible = source_to_visible[source_name]
self._names[visible] = None
return _source_rename_map(source_to_visible)
def validate(
self,
names: Sequence[str] | None = None,
*,
allow_dangling: bool = False,
) -> BuildReport:
"""
Run the full build logic and return a `BuildReport` without producing output.
This is a dry run over the same dependency resolution and recipe
execution path used by `build()`. Any generated library is discarded
after validation completes.
"""
_session, report = self._run_build(names=names, allow_dangling=allow_dangling)
return report
def build(
self,
*,
output: Literal['overlay', 'library'] = 'overlay',
allow_dangling: bool = False,
) -> tuple[ILibrary, BuildReport]:
"""
Materialize declarations and return a usable output library plus report.
Args:
output: `'overlay'` preserves imported source-backed cells where
possible, while `'library'` eagerly materializes the full
result.
allow_dangling: If `False`, fail the build when the completed
library still contains dangling references.
"""
if output not in ('overlay', 'library'):
raise ValueError(f'Unknown build output mode: {output!r}')
self._assert_editable()
session, report = self._run_build(names=None, allow_dangling=allow_dangling)
if output == 'library':
built_output = session.to_library()
else:
built_output = session.to_overlay()
self._frozen = True
return built_output, report
def _run_build(
self,
*,
names: Sequence[str] | None,
allow_dangling: bool,
) -> tuple[_BuildSessionLibrary, BuildReport]:
roots = tuple(dict.fromkeys(names if names is not None else self._declarations.keys()))
unknown = [name for name in roots if name not in self._names]
if unknown:
raise BuildError(f'Unknown build roots requested: {unknown}')
session = _BuildSessionLibrary(self)
sessions = dict(_ACTIVE_BUILD_SESSIONS.get() or {})
sessions[id(self)] = session
token = _ACTIVE_BUILD_SESSIONS.set(sessions)
try:
session.materialize_many(roots)
if not allow_dangling:
session.child_graph(dangling='error')
finally:
_ACTIVE_BUILD_SESSIONS.reset(token)
report = session.build_report(roots)
return session, report
class _BuildSessionLibrary(ILibrary):
"""
Internal overlay-backed library used while a `BuildLibrary` is executing.
This object provides the mutable-library surface that recipes expect while
also tracking declared-cell dependencies, helper-cell provenance, and
imported source cells. It exists only for the duration of a validation or
build run.
"""
def __init__(self, builder: BuildLibrary) -> None:
self._builder = builder
self._overlay = OverlayLibrary()
self._built: set[str] = set()
self._declared_stack: list[str] = []
self._names = dict(builder._names)
self._provenance: dict[str, CellProvenance] = {}
self._dependency_graph: defaultdict[str, set[str]] = defaultdict(set)
self._install_sources()
def _install_sources(self) -> None:
for source_library, source_to_visible in self._builder._sources:
source_order = source_library.source_order()
expected_names = set(source_to_visible)
actual_names = set(source_order)
if actual_names != expected_names:
added_names = sorted(actual_names - expected_names)
removed_names = sorted(expected_names - actual_names)
detail = []
if added_names:
detail.append(f'added={added_names}')
if removed_names:
detail.append(f'removed={removed_names}')
raise BuildError(
'Imported source library changed after add_source() was called '
f'({", ".join(detail)}). '
'Do not structurally mutate source libraries between add_source() and build()/validate().'
)
def rename_source(_lib: ILibraryView, name: str, *, mapping: Mapping[str, str] = source_to_visible) -> str:
return mapping[name]
self._overlay.add_source(
source_library,
rename_theirs = rename_source,
rename_when = 'always',
)
for source_name in source_order:
visible_name = source_to_visible[source_name]
self._provenance[visible_name] = CellProvenance(
requested_name = source_name,
kind = 'source',
owner_declared_name = None,
build_chain = (),
)
def __iter__(self) -> Iterator[str]:
return iter(self._names)
def __len__(self) -> int:
return len(self._names)
def __contains__(self, key: object) -> bool:
return key in self._names
def _current_declared(self) -> str | None:
if not self._declared_stack:
return None
return self._declared_stack[-1]
def _record_dependency(self, target: str) -> None:
current = self._current_declared()
if current is None or current == target or target not in self._builder._declarations:
return
self._dependency_graph[current].add(target)
def _guard_mutable_output_name(self, key: str, *, operation: str) -> None:
if key in self._builder._declarations:
raise BuildError(f'Cannot {operation} declared build cell "{key}" during an active build session.')
provenance = self._provenance.get(key)
if provenance is not None and provenance.kind == 'source':
raise BuildError(f'Cannot {operation} imported source cell "{key}" during an active build session.')
def rename(
self,
old_name: str,
new_name: str,
move_references: bool = False,
) -> Self:
if old_name == new_name:
return self
if old_name not in self._overlay:
if old_name in self._builder._declarations:
self._guard_mutable_output_name(old_name, operation='rename')
raise LibraryError(f'"{old_name}" does not exist in the library.')
self._guard_mutable_output_name(old_name, operation='rename')
if new_name in self._names:
raise LibraryError(f'"{new_name}" already exists in the library.')
self._overlay.rename(old_name, new_name, move_references=move_references)
self._names = {
new_name if name == old_name else name: None
for name in self._names
}
provenance = self._provenance.pop(old_name)
self._provenance[new_name] = provenance
return self
def __getitem__(self, key: str) -> Pattern:
if key in self._builder._declarations:
self._record_dependency(key)
self._ensure_declared(key)
return self._overlay[key]
def __setitem__(
self,
key: str,
value: Pattern | Callable[[], Pattern],
) -> None:
if key in self._overlay:
raise LibraryError(f'"{key}" already exists in the library. Overwriting is not allowed!')
current = self._current_declared()
if key in self._builder._declarations and key != current:
raise LibraryError(f'"{key}" is reserved for a declared cell and cannot be used as a helper name.')
pattern = value() if callable(value) else value
self._overlay[key] = pattern
self._names.setdefault(key, None)
kind: Literal['declared', 'helper']
if current is not None and key == current:
kind = 'declared'
else:
kind = 'helper'
self._provenance[key] = CellProvenance(
requested_name = key,
kind = kind,
owner_declared_name = current if kind == 'helper' else key,
build_chain = tuple(self._declared_stack),
)
def __delitem__(self, key: str) -> None:
if key not in self._overlay:
if key in self._builder._declarations:
self._guard_mutable_output_name(key, operation='delete')
raise KeyError(key)
self._guard_mutable_output_name(key, operation='delete')
if key in self._overlay:
del self._overlay[key]
self._names.pop(key, None)
self._provenance.pop(key, None)
def _merge(self, key_self: str, other: Mapping[str, Pattern], key_other: str) -> None:
self[key_self] = copy.deepcopy(other[key_other])
def add(
self,
other: Mapping[str, Pattern],
rename_theirs: Callable[[ILibraryView, str], str] = _rename_patterns,
mutate_other: bool = False,
) -> dict[str, str]:
rename_map = super().add(other, rename_theirs=rename_theirs, mutate_other=mutate_other)
current = self._current_declared()
for old_name, new_name in rename_map.items():
if new_name in self._provenance:
self._provenance[new_name] = replace(
self._provenance[new_name],
requested_name = old_name,
owner_declared_name = current if current is not None else self._provenance[new_name].owner_declared_name,
)
return rename_map
def _wrap_error(self, name: str, exc: Exception) -> BuildError:
chain = tuple(self._declared_stack)
msg = [f'Failed while building declared cell "{name}"']
if chain:
msg.append(f'Dependency chain: {" -> ".join(chain)}')
msg.append(f'Cause: {exc}')
return BuildError('\n'.join(msg))
def _ensure_named(self, name: str) -> None:
if name in self._builder._declarations:
self._record_dependency(name)
self._ensure_declared(name)
return
if name in self._overlay:
return
raise BuildError(f'Missing dependency "{name}"')
def _ensure_declared(self, name: str) -> None:
from ..pattern import Pattern # noqa: PLC0415
if name in self._built:
return
if name in self._declared_stack:
chain = ' -> '.join(self._declared_stack + [name])
raise BuildError(f'Cycle detected while building declared cells: {chain}')
declaration = self._builder._declarations[name]
self._declared_stack.append(name)
try:
if isinstance(declaration, _BuildRecipe):
for dep in declaration.explicit_dependencies:
self._ensure_named(dep)
pattern = declaration.func(*declaration.args, **declaration.kwargs)
if not isinstance(pattern, Pattern):
raise BuildError(f'Recipe for "{name}" returned {type(pattern).__name__}, expected Pattern') # noqa: TRY301
else:
pattern = declaration.deepcopy()
if name in self._overlay:
if self._overlay[name] is not pattern:
raise BuildError( # noqa: TRY301
f'Recipe for "{name}" wrote a different pattern into the session under its own name.'
)
else:
self[name] = pattern
self._built.add(name)
except Exception as exc:
raise self._wrap_error(name, exc) from exc
finally:
self._declared_stack.pop()
def materialize_many(self, names: Sequence[str]) -> None:
for name in dict.fromkeys(names):
self._ensure_named(name)
def source_order(self) -> tuple[str, ...]:
return self._overlay.source_order()
def child_graph(
self,
dangling: dangling_mode_t = 'error',
) -> dict[str, set[str]]:
return self._overlay.child_graph(dangling=dangling)
def parent_graph(
self,
dangling: dangling_mode_t = 'error',
) -> dict[str, set[str]]:
return self._overlay.parent_graph(dangling=dangling)
def build_report(self, requested_roots: Sequence[str]) -> BuildReport:
dependency_graph = {
name: frozenset(self._dependency_graph.get(name, set()))
for name in self._builder._declarations
if name in self._dependency_graph or name in requested_roots
}
return BuildReport(
requested_roots = tuple(dict.fromkeys(requested_roots)),
provenance = dict(self._provenance),
dependency_graph = dependency_graph,
)
def to_overlay(self) -> ILibrary:
return self._overlay
def to_library(self) -> Library:
mapping = {name: self._overlay[name] for name in self._overlay.source_order()}
return Library(mapping)

169
masque/library/lazy.py Normal file
View file

@ -0,0 +1,169 @@
"""Closure-backed lazy library implementation."""
from __future__ import annotations
from pprint import pformat
from typing import TYPE_CHECKING, Self, cast
import logging
from ..error import LibraryError
from .base import ILibrary
if TYPE_CHECKING:
from collections.abc import Callable, Iterator, Mapping
from ..pattern import Pattern
logger = logging.getLogger(__name__)
class LazyLibrary(ILibrary):
"""
This class is usually used to create a library of Patterns by mapping names to
functions which generate or load the relevant `Pattern` object as-needed.
TODO: lots of stuff causes recursive loads (e.g. data_to_ports?). What should you avoid?
"""
mapping: dict[str, Callable[[], Pattern]]
cache: dict[str, Pattern]
_lookups_in_progress: list[str]
def __init__(self) -> None:
self.mapping = {}
self.cache = {}
self._lookups_in_progress = []
def __setitem__(
self,
key: str,
value: Pattern | Callable[[], Pattern],
) -> None:
if key in self.mapping:
raise LibraryError(f'"{key}" already exists in the library. Overwriting is not allowed!')
if callable(value):
value_func = value
else:
value_func = lambda: cast('Pattern', value) # noqa: E731
self.mapping[key] = value_func
if key in self.cache:
del self.cache[key]
def __delitem__(self, key: str) -> None:
del self.mapping[key]
if key in self.cache:
del self.cache[key]
def __getitem__(self, key: str) -> Pattern:
logger.debug(f'loading {key}')
if key in self.cache:
logger.debug(f'found {key} in cache')
return self.cache[key]
if key in self._lookups_in_progress:
chain = ' -> '.join(self._lookups_in_progress + [key])
raise LibraryError(
f'Detected circular reference or recursive lookup of "{key}".\n'
f'Lookup chain: {chain}\n'
'This may be caused by an invalid (cyclical) reference, or buggy code.\n'
'If you are lazy-loading a file, try a non-lazy load and check for reference cycles.'
)
self._lookups_in_progress.append(key)
try:
func = self.mapping[key]
pat = func()
finally:
self._lookups_in_progress.pop()
self.cache[key] = pat
return pat
def __iter__(self) -> Iterator[str]:
return iter(self.mapping)
def __len__(self) -> int:
return len(self.mapping)
def __contains__(self, key: object) -> bool:
return key in self.mapping
def _merge(self, key_self: str, other: Mapping[str, Pattern], key_other: str) -> None:
if isinstance(other, LazyLibrary):
self.mapping[key_self] = other.mapping[key_other]
if key_other in other.cache:
self.cache[key_self] = other.cache[key_other]
else:
self[key_self] = other[key_other]
def __repr__(self) -> str:
return '<LazyLibrary with keys\n' + pformat(list(self.keys())) + '>'
def rename(
self,
old_name: str,
new_name: str,
move_references: bool = False,
) -> Self:
"""
Rename a pattern.
Args:
old_name: Current name for the pattern
new_name: New name for the pattern
move_references: Whether to scan all refs in the pattern and
move them to point to `new_name` as necessary.
Default `False`.
Returns:
self
"""
if old_name not in self.mapping:
raise LibraryError(f'"{old_name}" does not exist in the library.')
if old_name == new_name:
return self
self[new_name] = self.mapping[old_name] # copy over function
if old_name in self.cache:
self.cache[new_name] = self.cache[old_name]
del self[old_name]
if move_references:
self.move_references(old_name, new_name)
return self
def move_references(self, old_target: str, new_target: str) -> Self:
"""
Change all references pointing at `old_target` into references pointing at `new_target`.
Args:
old_target: Current reference target
new_target: New target for the reference
Returns:
self
"""
if old_target == new_target:
return self
self.precache()
for pattern in self.cache.values():
if old_target in pattern.refs:
pattern.refs[new_target].extend(pattern.refs[old_target])
del pattern.refs[old_target]
return self
def precache(self) -> Self:
"""
Force all patterns into the cache
Returns:
self
"""
for key in self.mapping:
_ = self[key] # want to trigger our own __getitem__
return self
def __deepcopy__(self, memo: dict | None = None) -> LazyLibrary:
raise LibraryError('LazyLibrary cannot be deepcopied (deepcopy doesn\'t descend into closures)')

112
masque/library/mapping.py Normal file
View file

@ -0,0 +1,112 @@
"""Concrete mapping-backed library implementations."""
from __future__ import annotations
from pprint import pformat
from typing import TYPE_CHECKING, Self
from ..error import LibraryError
from .base import ILibrary, ILibraryView
if TYPE_CHECKING:
from collections.abc import Callable, Iterator, Mapping, MutableMapping
from ..pattern import Pattern
class LibraryView(ILibraryView):
"""
Default implementation for a read-only library.
A library is a mapping from unique names (str) to collections of geometry (`Pattern`).
This library is backed by an arbitrary python object which implements the `Mapping` interface.
"""
mapping: Mapping[str, Pattern]
def __init__(
self,
mapping: Mapping[str, Pattern],
) -> None:
self.mapping = mapping
def __getitem__(self, key: str) -> Pattern:
return self.mapping[key]
def __iter__(self) -> Iterator[str]:
return iter(self.mapping)
def __len__(self) -> int:
return len(self.mapping)
def __contains__(self, key: object) -> bool:
return key in self.mapping
def __repr__(self) -> str:
return f'<LibraryView ({type(self.mapping)}) with keys\n' + pformat(list(self.keys())) + '>'
class Library(ILibrary):
"""
Default implementation for a writeable library.
A library is a mapping from unique names (str) to collections of geometry (`Pattern`).
This library is backed by an arbitrary python object which implements the `MutableMapping` interface.
"""
mapping: MutableMapping[str, Pattern]
def __init__(
self,
mapping: MutableMapping[str, Pattern] | None = None,
) -> None:
if mapping is None:
self.mapping = {}
else:
self.mapping = mapping
def __getitem__(self, key: str) -> Pattern:
return self.mapping[key]
def __iter__(self) -> Iterator[str]:
return iter(self.mapping)
def __len__(self) -> int:
return len(self.mapping)
def __contains__(self, key: object) -> bool:
return key in self.mapping
def __setitem__(
self,
key: str,
value: Pattern | Callable[[], Pattern],
) -> None:
if key in self.mapping:
raise LibraryError(f'"{key}" already exists in the library. Overwriting is not allowed!')
value = value() if callable(value) else value
self.mapping[key] = value
def __delitem__(self, key: str) -> None:
del self.mapping[key]
def _merge(self, key_self: str, other: Mapping[str, Pattern], key_other: str) -> None:
self[key_self] = other[key_other]
def __repr__(self) -> str:
return f'<Library ({type(self.mapping)}) with keys\n' + pformat(list(self.keys())) + '>'
@classmethod
def mktree(cls: type[Self], name: str) -> tuple[Self, Pattern]:
"""
Create a new Library and immediately add a pattern
Args:
name: The name for the new pattern (usually the name of the topcell).
Returns:
The newly created `Library` and the newly created `Pattern`
"""
from ..pattern import Pattern # noqa: PLC0415
tree = cls()
pat = Pattern()
tree[name] = pat
return tree, pat

View file

@ -1,44 +1,24 @@
"""
Shared helpers for source-backed lazy GDS views.
This module contains the reusable pieces that sit between lazy source readers
and ordinary mutable library usage:
- `PortsLibraryView` layers a processed, ports-importing cache on top of a raw
source view without mutating the source itself
- `OverlayLibrary` exposes a mutable library surface that can mix source-backed
cells with overlay-owned materialized patterns
- the write helpers preserve source-backed copy-through behavior where
possible, falling back to normal pattern serialization when a cell has been
materialized or remapped
Both the classic and Arrow-backed lazy GDS readers rely on these helpers.
"""
"""Overlay and ports-importing library views."""
from __future__ import annotations
from dataclasses import dataclass
from typing import IO, Any, Literal, cast
from collections import defaultdict
from collections.abc import Callable, Iterator, Mapping, Sequence
from dataclasses import dataclass
from typing import TYPE_CHECKING, Any, Literal, cast
import copy
import gzip
import logging
import pathlib
import klamath
import numpy
from numpy.typing import NDArray
from . import gdsii
from .utils import tmpfile
from ..error import LibraryError
from ..library import ILibrary, ILibraryView, LibraryView, _plan_source_names, _source_rename_map, dangling_mode_t
from ..pattern import Pattern, map_targets
from ..utils import apply_transforms
from ..utils.ports2data import data_to_ports
from ..utils import apply_transforms, layer_t
from .base import ILibrary, ILibraryView
from .utils import dangling_mode_t, _plan_source_names, _source_rename_map
from .mapping import LibraryView
if TYPE_CHECKING:
from collections.abc import Callable, Iterator, Mapping, Sequence
logger = logging.getLogger(__name__)
from numpy.typing import NDArray
@dataclass
@ -58,23 +38,6 @@ class _SourceEntry:
source_name: str
def _pattern_children(pat: Pattern) -> set[str]:
return {child for child, refs in pat.refs.items() if child is not None and refs}
def _remap_pattern_targets(pat: Pattern, remap: Callable[[str | None], str | None]) -> Pattern:
if not pat.refs:
return pat
pat.refs = map_targets(pat.refs, remap)
return pat
def _coerce_library_view(source: Mapping[str, Pattern] | ILibraryView) -> ILibraryView:
if isinstance(source, ILibraryView):
return source
return LibraryView(source)
def _materialize_detached_pattern(view: ILibraryView, name: str) -> Pattern:
func = getattr(view, '_materialize_pattern', None)
if callable(func):
@ -98,7 +61,7 @@ class PortsLibraryView(ILibraryView):
self,
source: ILibraryView,
*,
layers: Sequence[gdsii.layer_t],
layers: Sequence[layer_t],
max_depth: int = 0,
skip_subcells: bool = True,
) -> None:
@ -109,7 +72,7 @@ class PortsLibraryView(ILibraryView):
self._cache: dict[str, Pattern] = {}
self._lookups_in_progress: list[str] = []
if hasattr(source, 'library_info'):
self.library_info = cast('dict[str, Any]', getattr(source, 'library_info'))
self.library_info = cast('dict[str, Any]', source.library_info)
def __getitem__(self, key: str) -> Pattern:
return self._materialize_pattern(key, persist=True)
@ -124,6 +87,8 @@ class PortsLibraryView(ILibraryView):
return key in self._source
def _materialize_pattern(self, name: str, *, persist: bool) -> Pattern:
from ..utils.ports2data import data_to_ports # noqa: PLC0415
if name in self._cache:
return self._cache[name]
@ -222,7 +187,7 @@ class PortsLibraryView(ILibraryView):
def raw_struct_bytes(self, name: str) -> bytes:
reader = getattr(self._source, 'raw_struct_bytes', None)
if not callable(reader):
raise AttributeError('raw_struct_bytes')
raise TypeError('raw_struct_bytes')
return cast('bytes', reader(name))
def can_copy_raw_struct(self, name: str) -> bool:
@ -250,14 +215,6 @@ class OverlayLibrary(ILibrary):
Source-backed cells remain lazy until accessed through `__getitem__`, at
which point that visible cell is promoted into an overlay-owned materialized
`Pattern`.
This is the main mutable integration surface for lazy GDS content. It lets
callers:
- expose one or more source-backed libraries behind a normal `ILibrary`
interface
- add or replace cells with overlay-owned patterns
- rename visible source cells
- remap references without immediately rewriting untouched source structs
"""
def __init__(self) -> None:
@ -315,7 +272,7 @@ class OverlayLibrary(ILibrary):
If `'always'`, every imported source name is passed through
`rename_theirs`.
"""
view = _coerce_library_view(source)
view = source if isinstance(source, ILibraryView) else LibraryView(source)
source_order = list(view.source_order())
child_graph = view.child_graph(dangling='include')
@ -414,8 +371,13 @@ class OverlayLibrary(ILibrary):
layer = self._layers[entry.layer_index]
source_pat = layer.library[entry.source_name].deepcopy()
remap = lambda target: None if target is None else self._effective_target(layer, target)
pat = _remap_pattern_targets(source_pat, remap)
def remap(target: str | None) -> str | None:
return None if target is None else self._effective_target(layer, target)
if source_pat.refs:
source_pat.refs = map_targets(source_pat.refs, remap)
pat = source_pat
if persist:
self._entries[name] = pat
return pat
@ -430,7 +392,7 @@ class OverlayLibrary(ILibrary):
continue
entry = self._entries[name]
if isinstance(entry, Pattern):
graph[name] = _pattern_children(entry)
graph[name] = {child for child, refs in entry.refs.items() if child is not None and refs}
continue
layer = self._layers[entry.layer_index]
children = {self._effective_target(layer, child) for child in layer.child_graph.get(entry.source_name, set())}
@ -551,146 +513,3 @@ class OverlayLibrary(ILibrary):
def source_order(self) -> tuple[str, ...]:
return tuple(name for name in self._order if name in self._entries)
def _iter_library_infos(library: Mapping[str, Pattern] | ILibraryView) -> Iterator[dict[str, Any]]:
info = getattr(library, 'library_info', None)
if isinstance(info, dict):
yield info
if isinstance(library, OverlayLibrary):
for layer in library._layers:
yield from _iter_library_infos(layer.library)
def _get_write_info(
library: Mapping[str, Pattern] | ILibraryView,
*,
meters_per_unit: float | None,
logical_units_per_unit: float | None,
library_name: str | None,
) -> tuple[float, float, str]:
if meters_per_unit is not None and logical_units_per_unit is not None and library_name is not None:
return meters_per_unit, logical_units_per_unit, library_name
infos = list(_iter_library_infos(library))
if infos:
unit_pairs = {(info['meters_per_unit'], info['logical_units_per_unit']) for info in infos}
if len(unit_pairs) > 1:
raise LibraryError('Merged lazy GDS sources must have identical units before writing')
info = infos[0]
meters = info['meters_per_unit'] if meters_per_unit is None else meters_per_unit
logical = info['logical_units_per_unit'] if logical_units_per_unit is None else logical_units_per_unit
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
def _can_copy_raw_cell(library: Mapping[str, Pattern] | ILibraryView, name: str) -> bool:
can_copy = getattr(library, 'can_copy_raw_struct', None)
if not callable(can_copy):
return False
return bool(can_copy(name))
def _raw_struct_bytes(library: Mapping[str, Pattern] | ILibraryView, name: str) -> bytes:
reader = getattr(library, 'raw_struct_bytes', None)
if not callable(reader):
raise AttributeError('raw_struct_bytes')
return cast('bytes', reader(name))
def _can_copy_overlay_cell(library: OverlayLibrary, name: str, entry: _SourceEntry) -> bool:
layer = library._layers[entry.layer_index]
if name != entry.source_name:
return False
if not _can_copy_raw_cell(layer.library, entry.source_name):
return False
children = layer.child_graph.get(entry.source_name, set())
return all(library._effective_target(layer, child) == child for child in children)
def _write_pattern_struct(stream: IO[bytes], name: str, pat: Pattern) -> None:
elements: list[klamath.elements.Element] = []
elements += gdsii._shapes_to_elements(pat.shapes)
elements += gdsii._labels_to_texts(pat.labels)
elements += gdsii._mrefs_to_grefs(pat.refs)
klamath.library.write_struct(stream, name=name.encode('ASCII'), elements=elements)
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:
meters_per_unit, logical_units_per_unit, library_name = _get_write_info(
library,
meters_per_unit=meters_per_unit,
logical_units_per_unit=logical_units_per_unit,
library_name=library_name,
)
header = klamath.library.FileHeader(
name=library_name.encode('ASCII'),
user_units_per_db_unit=logical_units_per_unit,
meters_per_db_unit=meters_per_unit,
)
header.write(stream)
if isinstance(library, OverlayLibrary):
for name in library.source_order():
entry = library._entries[name]
if isinstance(entry, _SourceEntry) and _can_copy_overlay_cell(library, name, entry):
layer = library._layers[entry.layer_index]
stream.write(_raw_struct_bytes(layer.library, entry.source_name))
else:
_write_pattern_struct(stream, name, library._materialize_pattern(name, persist=False))
klamath.records.ENDLIB.write(stream, None)
return
if hasattr(library, 'raw_struct_bytes'):
for name in library.source_order():
if _can_copy_raw_cell(library, name):
stream.write(_raw_struct_bytes(library, name))
else:
_write_pattern_struct(stream, name, _materialize_detached_pattern(cast('ILibraryView', library), name))
klamath.records.ENDLIB.write(stream, None)
return
gdsii.write(cast('Mapping[str, Pattern]', library), stream, meters_per_unit, logical_units_per_unit, library_name)
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:
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:
write(
library,
stream,
meters_per_unit=meters_per_unit,
logical_units_per_unit=logical_units_per_unit,
library_name=library_name,
)
finally:
for ss in streams:
ss.close()

124
masque/library/utils.py Normal file
View file

@ -0,0 +1,124 @@
"""Shared types and helpers for library implementations."""
from __future__ import annotations
from typing import TYPE_CHECKING, Literal, Protocol, TypeAlias
from collections.abc import Callable, Container, Mapping, MutableMapping, Sequence
from ..error import LibraryError
if TYPE_CHECKING:
import numpy
from numpy.typing import NDArray
from ..pattern import Pattern
from .base import ILibraryView
class visitor_function_t(Protocol):
""" Signature for `Library.dfs()` visitor functions. """
def __call__(
self,
pattern: Pattern,
hierarchy: tuple[str | None, ...],
memo: dict,
transform: NDArray[numpy.float64] | Literal[False],
) -> Pattern:
...
TreeView: TypeAlias = Mapping[str, 'Pattern']
""" A name-to-`Pattern` mapping which is expected to have only one top-level cell """
Tree: TypeAlias = MutableMapping[str, 'Pattern']
""" A mutable name-to-`Pattern` mapping which is expected to have only one top-level cell """
dangling_mode_t: TypeAlias = Literal['error', 'ignore', 'include']
""" How helpers should handle refs whose targets are not present in the library. """
SINGLE_USE_PREFIX = '_'
"""
Names starting with this prefix are assumed to refer to single-use patterns,
which may be renamed automatically by `ILibrary.add()` (via
`rename_theirs=_rename_patterns()` )
"""
# TODO what are the consequences of making '_' special? maybe we can make this decision everywhere?
def _rename_patterns(lib: ILibraryView, name: str) -> str:
"""
The default `rename_theirs` function for `ILibrary.add`.
Treats names starting with `SINGLE_USE_PREFIX` (default: one underscore) as
"one-offs" for which name conflicts should be automatically resolved.
Conflicts are resolved by calling `lib.get_name(SINGLE_USE_PREFIX + stem)`
where `stem = name.removeprefix(SINGLE_USE_PREFIX).split('$')[0]`.
Names lacking the prefix are directly returned (not renamed).
Args:
lib: The library into which `name` is to be added (but is presumed to conflict)
name: The original name, to be modified
Returns:
The new name, not guaranteed to be conflict-free!
"""
if not name.startswith(SINGLE_USE_PREFIX):
return name
stem = name.removeprefix(SINGLE_USE_PREFIX).split('$')[0]
return lib.get_name(SINGLE_USE_PREFIX + stem)
def _plan_source_names(
target: ILibraryView,
source_order: Sequence[str],
existing_names: Container[str],
*,
rename_theirs: Callable[[ILibraryView, str], str] | None = None,
rename_when: Literal['conflict', 'always'] = 'conflict',
) -> dict[str, str]:
if rename_when not in ('conflict', 'always'):
raise ValueError(f'Unknown source rename mode: {rename_when!r}')
if rename_when == 'always' and rename_theirs is None:
raise TypeError('rename_theirs is required when rename_when="always"')
source_to_visible: dict[str, str] = {}
visible_names: set[str] = set()
for name in source_order:
visible = name
if rename_when == 'always':
assert rename_theirs is not None
visible = rename_theirs(target, name)
elif visible in existing_names or visible in visible_names:
if rename_theirs is None:
raise LibraryError(f'Conflicting name while adding source: {name!r}')
visible = rename_theirs(target, name)
if visible in existing_names or visible in visible_names:
raise LibraryError(f'Unresolved duplicate key encountered while adding source: {name!r} -> {visible!r}')
source_to_visible[name] = visible
visible_names.add(visible)
return source_to_visible
def _source_rename_map(source_to_visible: Mapping[str, str]) -> dict[str, str]:
return {
source_name: visible_name
for source_name, visible_name in source_to_visible.items()
if source_name != visible_name
}
def b64suffix(ii: int) -> str:
"""
Turn an integer into a base64-equivalent suffix.
This could be done with base64.b64encode, but this way is faster for many small `ii`.
"""
def i2a(nn: int) -> str:
return 'ABCDEFGHIJKLMNOPQRSTUVWXYZabcdefghijklmnopqrstuvwxyz0123456789$?'[nn]
parts = ['$', i2a(ii % 64)]
ii >>= 6
while ii:
parts.append(i2a(ii % 64))
ii >>= 6
return ''.join(parts)

View file

@ -122,7 +122,7 @@ def test_autotool_straight_offer_supports_requested_output_transition(
) -> None:
_p, tool, lib = autotool_setup
p = Pather(lib, tools=tool, auto_render=False)
p = Pather(lib, tools=tool, render='deferred')
p.ports["start"] = Port((0, 0), pi, ptype="wire_m1")
p.straight("start", 10, out_ptype="wire_m2")
@ -158,7 +158,7 @@ def test_pather_straight_topology_allows_transition_offset_cancellation() -> Non
.add_transition(lib.abstract("trans_out"), "EXT", "CORE")
)
p = Pather(lib, tools=tool, auto_render=False)
p = Pather(lib, tools=tool, render='deferred')
p.ports["A"] = Port((0, 0), pi, ptype="ext_in")
p.trace("A", None, length=10, out_ptype="ext_out")
@ -550,7 +550,7 @@ def test_autotool_s_offer_bbox_matches_rendered_sbend() -> None:
assert_offer_bbox_matches_rendered_offer(tool, offer, 4, lib)
pather = Pather(lib, tools=tool, auto_render=False)
pather = Pather(lib, tools=tool, render='deferred')
pather.ports["A"] = Port((0, 0), 0, ptype="ext")
pather.jog("A", 4)
@ -581,7 +581,7 @@ def test_autotool_s_offer_uses_absolute_jog_range_for_both_signs() -> None:
assert valid_positive
assert valid_negative
pather = Pather(Library(), tools=tool, auto_render=False)
pather = Pather(Library(), tools=tool, render='deferred')
pather.ports["A"] = Port((0, 0), 0, ptype="core")
pather.jog("A", -4)
assert_allclose(pather.ports["A"].offset, [-10, 4])
@ -623,7 +623,7 @@ def test_autotool_s_offer_singleton_jog_range_includes_both_signs() -> None:
assert [offer.endpoint_at(-4).y for offer in offers if offer.jog_domain == (-4.0, -4.0)] == [-4]
for jog, expected_y in ((4, -4), (-4, 4)):
pather = Pather(Library(), tools=tool, auto_render=False)
pather = Pather(Library(), tools=tool)
pather.ports["A"] = Port((0, 0), 0, ptype="core")
pather.jog("A", jog)
assert_allclose(pather.ports["A"].offset, [-10, expected_y])
@ -669,7 +669,7 @@ def test_autotool_uturn_offer_mirror_exposes_both_signs() -> None:
def test_pather_autotool_uses_prebaked_uturn_offer() -> None:
tool, lib = make_uturn_tool()
p = Pather(lib, tools=tool, auto_render=False)
p = Pather(lib, tools=tool, render='deferred')
p.ports["A"] = Port((0, 0), 0, ptype="core")
p.uturn("A", 4, length=10)
@ -695,7 +695,7 @@ def test_pather_autotool_omitted_uturn_composes_l_offers(
multi_bend_tool: tuple[AutoTool, Library],
) -> None:
tool, lib = multi_bend_tool
p = Pather(lib, tools=tool, auto_render=False)
p = Pather(lib, tools=tool, render='deferred')
p.ports["A"] = Port((0, 0), 0, ptype="wire")
p.uturn("A", 20)
@ -714,7 +714,7 @@ def test_pather_autotool_explicit_uturn_composes_l_offers(
multi_bend_tool: tuple[AutoTool, Library],
) -> None:
tool, lib = multi_bend_tool
p = Pather(lib, tools=tool, auto_render=False)
p = Pather(lib, tools=tool, render='deferred')
p.ports["A"] = Port((0, 0), 0, ptype="wire")
p.uturn("A", 20, length=10)
@ -739,7 +739,7 @@ def test_autotool_offer_bbox_rejects_invalid_parameter(multi_bend_tool: tuple[Au
def test_pather_autotool_uses_l_offer_domains(multi_bend_tool: tuple[AutoTool, Library]) -> None:
tool, lib = multi_bend_tool
p = Pather(lib, tools=tool, auto_render=False)
p = Pather(lib, tools=tool, render='deferred')
p.ports["A"] = Port((0, 0), 0, ptype="wire")
p.trace("A", True, length=15)
@ -760,7 +760,7 @@ def test_autotool_generated_primitives_do_not_capture_route_kwargs() -> None:
return make_straight(length, ptype="wire")
tool = AutoTool().add_straight("wire", make_marked_straight, "A", length_range=(0, 1e8))
p = Pather(Library(), tools=tool, auto_render=False)
p = Pather(Library(), tools=tool, render='deferred')
p.ports["A"] = Port((0, 0), 0, ptype="wire")
p.straight("A", 5, marker="route")
@ -790,7 +790,7 @@ def test_autotool_bend_offer_supports_requested_output_transition(ccw: bool) ->
.add_transition(lib.abstract("out_trans"), "EXT", "CORE")
)
p = Pather(lib, tools=tool, auto_render=False)
p = Pather(lib, tools=tool, render='deferred')
p.ports["A"] = Port((0, 0), 0, ptype="core")
p.trace("A", ccw, length=10, out_ptype="ext")
@ -824,7 +824,7 @@ def test_autotool_bend_offer_supports_bend_input_transition(ccw: bool) -> None:
.add_transition(lib.abstract("bend_trans"), "MID", "CORE")
)
p = Pather(lib, tools=tool, auto_render=False)
p = Pather(lib, tools=tool, render='deferred')
p.ports["A"] = Port((0, 0), 0, ptype="core")
p.trace("A", ccw, length=10, out_ptype="mid")
@ -857,7 +857,7 @@ def test_pather_accepts_bend_offer_with_zero_lateral_endpoint() -> None:
.add_transition(lib.abstract("bend_trans"), "MID", "CORE")
)
p = Pather(lib, tools=tool, auto_render=False)
p = Pather(lib, tools=tool, render='deferred')
p.ports["A"] = Port((0, 0), 0, ptype="core")
p.trace("A", True, length=10, out_ptype="mid")
@ -897,7 +897,7 @@ def test_autotool_bend_offer_supports_bend_and_output_transitions(ccw: bool) ->
.add_transition(lib.abstract("out_trans"), "EXT", "MID")
)
p = Pather(lib, tools=tool, auto_render=False)
p = Pather(lib, tools=tool, render='deferred')
p.ports["A"] = Port((0, 0), 0, ptype="core")
p.trace("A", ccw, length=12, out_ptype="ext")
@ -955,7 +955,7 @@ def test_pather_autotool_pure_sbend_with_transition_dx() -> None:
assert isinstance(s_data, AutoTool.GeneratedData)
assert s_offer.out_ptype == "core"
p = Pather(lib, tools=tool, auto_render=False)
p = Pather(lib, tools=tool, render='deferred')
p.ports["A"] = Port((0, 0), 0, ptype="ext")
p.jog("A", 4)

View file

@ -14,8 +14,9 @@ from ..library import Library
from ..pattern import Pattern
from ..repetition import Grid
from ..shapes import Path as MPath, Polygon, PolyCollection, RectCollection
from ..file import gdsii, gdsii_arrow
from ..file.gdsii_perf import write_fixture
from ..file import gdsii
from ..file.gdsii import arrow as gdsii_arrow
from ..file.gdsii.perf import write_fixture
if not gdsii_arrow.is_available():
@ -267,7 +268,7 @@ def test_gdsii_arrow_invalid_input_raises_klamath_error(tmp_path: Path) -> None:
gds_file.write_bytes(b'not-a-gds')
script = textwrap.dedent(f"""
from masque.file import gdsii_arrow
from masque.file.gdsii import arrow as gdsii_arrow
try:
gdsii_arrow.readfile({str(gds_file)!r})
except Exception as exc:

View file

@ -4,9 +4,10 @@ import numpy
import pytest
from numpy.testing import assert_allclose
from ..file import gdsii, gdsii_lazy
from ..file import gdsii
from ..file.gdsii import lazy as gdsii_lazy
from ..pattern import Pattern
from ..library import Library
from ..library import Library, OverlayLibrary
def _make_lazy_port_library() -> Library:
@ -73,7 +74,7 @@ def test_gdsii_lazy_overlay_add_source_stays_lazy_for_processed_view(tmp_path: P
raw, _ = gdsii_lazy.readfile(gds_file)
processed = raw.with_ports_from_data(layers=[(10, 0)], max_depth=2)
overlay = gdsii_lazy.OverlayLibrary()
overlay = OverlayLibrary()
overlay.add_source(processed)
assert not raw._cache
@ -85,7 +86,7 @@ def test_gdsii_lazy_overlay_add_source_stays_lazy_for_processed_view(tmp_path: P
def test_gdsii_lazy_overlay_add_source_can_rename_every_source_cell() -> None:
src = _make_lazy_port_library()
overlay = gdsii_lazy.OverlayLibrary()
overlay = OverlayLibrary()
rename_map = overlay.add_source(
src,
@ -106,10 +107,10 @@ def test_gdsii_lazy_overlay_add_source_rename_when_validation() -> None:
src = _make_lazy_port_library()
with pytest.raises(TypeError, match='rename_theirs'):
gdsii_lazy.OverlayLibrary().add_source(src, rename_when='always')
OverlayLibrary().add_source(src, rename_when='always')
with pytest.raises(ValueError, match='rename mode'):
gdsii_lazy.OverlayLibrary().add_source(src, rename_when='sometimes') # type: ignore[arg-type]
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:

View file

@ -10,11 +10,12 @@ import pytest
pytest.importorskip('pyarrow')
from .. import PatternError
from ..library import Library
from ..library import Library, OverlayLibrary
from ..pattern import Pattern
from ..repetition import Grid
from ..file import gdsii, gdsii_lazy_arrow
from ..file.gdsii_perf import write_fixture
from ..file import gdsii
from ..file.gdsii import lazy_arrow as gdsii_lazy_arrow
from ..file.gdsii.perf import write_fixture
if not gdsii_lazy_arrow.is_available():
@ -155,7 +156,7 @@ def test_gdsii_lazy_arrow_local_and_global_refs(tmp_path: Path) -> None:
assert sum(arr.shape[0] for arr in local['mid']) == 5
global_refs = lib.find_refs_global('leaf')
assert {path for path in global_refs} == {('top', 'mid', 'leaf')}
assert set(global_refs) == {('top', 'mid', 'leaf')}
assert global_refs[('top', 'mid', 'leaf')].shape[0] == 5
@ -177,7 +178,7 @@ def test_gdsii_lazy_arrow_invalid_input_raises_klamath_error(tmp_path: Path) ->
gds_file.write_bytes(b'not-a-gds')
script = textwrap.dedent(f"""
from masque.file import gdsii_lazy_arrow
from masque.file.gdsii import lazy_arrow as gdsii_lazy_arrow
try:
gdsii_lazy_arrow.readfile({str(gds_file)!r})
except Exception as exc:
@ -265,7 +266,7 @@ def test_gdsii_lazy_overlay_merge_and_write(tmp_path: Path) -> None:
lib_a, _ = gdsii_lazy_arrow.readfile(gds_a)
lib_b, _ = gdsii_lazy_arrow.readfile(gds_b)
overlay = gdsii_lazy_arrow.OverlayLibrary()
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']
@ -293,7 +294,7 @@ def test_gdsii_writer_accepts_overlay_library(tmp_path: Path) -> None:
lib, info = gdsii_lazy_arrow.readfile(gds_file)
overlay = gdsii_lazy_arrow.OverlayLibrary()
overlay = OverlayLibrary()
overlay.add_source(lib)
overlay.rename('leaf', 'leaf_copy', move_references=True)

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 ..file.gdsii.perf import fixture_manifest, write_fixture
def test_gdsii_perf_fixture_smoke(tmp_path: Path) -> None:

View file

@ -70,7 +70,7 @@ def test_autotool_uturn() -> None:
.add_bend(lib.abstract('bend'), 'in', 'out', clockwise=True)
)
p = Pather(lib, tools=tool, auto_render=False)
p = Pather(lib, tools=tool)
p.pattern.ports['A'] = Port((0, 0), 0)
p.at('A').uturn(offset=-2000, length=1000)
@ -82,7 +82,7 @@ def test_autotool_uturn() -> None:
def test_deferred_render_autotool_double_L(multi_bend_tool: tuple[AutoTool, Library]) -> None:
tool, lib = multi_bend_tool
rp = Pather(lib, tools=tool, auto_render=False)
rp = Pather(lib, tools=tool)
rp.ports["A"] = Port((0,0), 0, ptype="wire")
rp.jog("A", 10, length=20)

View file

@ -117,7 +117,7 @@ class CountingPathTool(PathTool):
def test_pather_jog_failed_two_bend_route_is_atomic() -> None:
lib = Library()
tool = PathTool(layer='M1', width=2, ptype='wire')
p = Pather(lib, tools=tool)
p = Pather(lib, tools=tool, render='immediate')
p.pattern.ports['A'] = Port((0, 0), rotation=0, ptype='wire')
with pytest.raises(BuildError, match='S-bend'):
@ -130,7 +130,7 @@ def test_pather_jog_failed_two_bend_route_is_atomic() -> None:
def test_pather_jog_accepts_sub_width_offset_when_length_is_sufficient() -> None:
lib = Library()
tool = PathTool(layer='M1', width=2, ptype='wire')
p = Pather(lib, tools=tool)
p = Pather(lib, tools=tool, render='immediate')
p.pattern.ports['A'] = Port((0, 0), rotation=0, ptype='wire')
p.jog('A', 1.5, length=5)
@ -139,10 +139,10 @@ def test_pather_jog_accepts_sub_width_offset_when_length_is_sufficient() -> None
assert p.pattern.ports['A'].rotation == 0
assert len(p._paths['A']) == 0
def test_pather_auto_render_batches_multi_step_selected_route_once() -> None:
def test_pather_immediate_render_batches_multi_step_selected_route_once() -> None:
lib = Library()
tool = CountingPathTool(layer='M1', width=2, ptype='wire')
p = Pather(lib, tools=tool, auto_render=True)
p = Pather(lib, tools=tool, render='immediate')
p.pattern.ports['A'] = Port((0, 0), rotation=0, ptype='wire')
p.jog('A', 4, length=10)
@ -169,7 +169,7 @@ def test_pather_jog_length_solved_from_single_position_bound() -> None:
def test_pather_positional_bound_requires_port_rotation() -> None:
p = Pather(Library(), tools=PathTool(layer='M1', width=1, ptype='wire'), auto_render=False)
p = Pather(Library(), tools=PathTool(layer='M1', width=1, ptype='wire'), render='deferred')
p.pattern.ports['A'] = Port((0, 0), rotation=None, ptype='wire')
with pytest.raises(BuildError, match='Ports must have rotation'):
@ -179,7 +179,7 @@ def test_pather_positional_bound_requires_port_rotation() -> None:
def test_pather_jog_omitted_length_uses_minimum_length_route() -> None:
lib = Library()
tool = PathTool(layer='M1', width=1, ptype='wire')
p = Pather(lib, tools=tool, auto_render=False)
p = Pather(lib, tools=tool, render='deferred')
p.pattern.ports['A'] = Port((0, 0), rotation=0, ptype='wire')
p.jog('A', 2)
@ -195,7 +195,7 @@ def test_pather_jog_omitted_length_uses_minimum_length_route() -> None:
def test_pather_trace_omitted_length_uses_minimum_offer() -> None:
lib = Library()
tool = PathTool(layer='M1', width=2, ptype='wire')
p = Pather(lib, tools=tool, auto_render=False)
p = Pather(lib, tools=tool, render='deferred')
p.pattern.ports['A'] = Port((0, 0), rotation=0, ptype='wire')
p.trace('A', None)
@ -211,7 +211,7 @@ def test_pather_trace_omitted_length_uses_minimum_offer() -> None:
def test_pather_trace_to_without_bound_uses_single_port_trace_minimum() -> None:
lib = Library()
tool = PathTool(layer='M1', width=2, ptype='wire')
p = Pather(lib, tools=tool, auto_render=False)
p = Pather(lib, tools=tool, render='deferred')
p.pattern.ports['A'] = Port((0, 0), rotation=0, ptype='wire')
p.trace_to('A', False)
@ -229,7 +229,7 @@ def test_pather_trace_to_rejects_conflicting_position_bounds() -> None:
with pytest.raises(BuildError, match='exactly one positional bound'):
p.trace_to('A', None, **kwargs)
p = Pather(Library(), tools=tool)
p = Pather(Library(), tools=tool, render='deferred')
p.pattern.ports['A'] = Port((0, 0), rotation=0, ptype='wire')
with pytest.raises(BuildError, match='length cannot be combined'):
p.trace_to('A', None, x=-5, length=3)
@ -261,7 +261,7 @@ def test_planner_constrained_bend_requires_jog() -> None:
def test_pather_trace_each_plans_all_ports_before_mutation() -> None:
tool = FirstPortOnlyTraceTool()
p = Pather(Library(), tools=tool, auto_render=False)
p = Pather(Library(), tools=tool)
p.pattern.ports['A'] = Port((0, 0), rotation=0, ptype='wire')
p.pattern.ports['B'] = Port((-2, 5), rotation=0, ptype='blocked')
@ -277,7 +277,7 @@ def test_pather_trace_each_plans_all_ports_before_mutation() -> None:
def test_pather_bundle_trace_plans_all_ports_before_mutation_or_render() -> None:
tool = FirstPortOnlyTraceTool()
p = Pather(Library(), tools=tool, auto_render=True)
p = Pather(Library(), tools=tool, render='immediate')
p.pattern.ports['A'] = Port((0, 0), rotation=0, ptype='wire')
p.pattern.ports['B'] = Port((0, 4), rotation=0, ptype='blocked')
@ -343,7 +343,7 @@ def test_pather_route_commit_failure_is_atomic_for_multi_port_trace() -> None:
return tree
tool = CommitFailureTool()
p = Pather(Library(), tools=tool, auto_render=True)
p = Pather(Library(), tools=tool, render='immediate')
p.pattern.ports['A'] = Port((0, 0), rotation=0, ptype='wire')
p.pattern.ports['B'] = Port((0, 4), rotation=0, ptype='bad')
@ -433,7 +433,7 @@ def test_pather_uturn_does_not_use_direct_planl_fallback() -> None:
jog = -1
return Port((length, jog), rotation=rotation, ptype=in_ptype or 'wire'), {'ccw': ccw, 'length': length}
p = Pather(Library(), tools=PlanLOnlyTool(), auto_render=False)
p = Pather(Library(), tools=PlanLOnlyTool(), render='deferred')
p.pattern.ports['A'] = Port((0, 0), rotation=0, ptype='wire')
with pytest.raises(BuildError, match='No legal primitive offer for omitted-length U-turn'):
@ -476,7 +476,7 @@ def test_pather_su_topology_rejects_out_ptype_sensitive_planl_jog() -> None:
ptype = out_ptype or in_ptype or 'wire'
return Port((length, jog), rotation=rotation, ptype=ptype), {'ccw': ccw, 'length': length}
p = Pather(Library(), tools=OutPtypeSensitiveTool(), auto_render=False)
p = Pather(Library(), tools=OutPtypeSensitiveTool(), render='deferred')
p.pattern.ports['A'] = Port((0, 0), rotation=0, ptype='wire')
with pytest.raises((BuildError, NotImplementedError)):
@ -509,7 +509,7 @@ def test_pather_two_l_planl_only_uturn_is_not_supported() -> None:
jog = -1
return Port((length, jog), rotation=rotation, ptype=in_ptype or 'wire'), {'ccw': ccw, 'length': length}
p = Pather(Library(), tools=PlanLOnlyTool(), auto_render=False)
p = Pather(Library(), tools=PlanLOnlyTool(), render='deferred')
p.pattern.ports['A'] = Port((0, 0), rotation=0, ptype='wire')
with pytest.raises((BuildError, NotImplementedError)):
@ -543,7 +543,7 @@ def test_pather_two_l_planl_only_jog_is_not_supported() -> None:
jog = -1
return Port((length, jog), rotation=rotation, ptype=in_ptype or 'wire'), {'ccw': ccw, 'length': length}
p = Pather(Library(), tools=PlanLOnlyTool(), auto_render=False)
p = Pather(Library(), tools=PlanLOnlyTool(), render='deferred')
p.pattern.ports['A'] = Port((0, 0), rotation=0, ptype='wire')
with pytest.raises((BuildError, NotImplementedError)):

View file

@ -62,7 +62,7 @@ def test_builder_tool_symbol_exports() -> None:
def test_pather_pending_render_steps_are_private() -> None:
p = Pather(Library(), tools=PathTool(layer=(1, 0), width=1), auto_render=False)
p = Pather(Library(), tools=PathTool(layer=(1, 0), width=1))
assert not hasattr(p, 'paths')
assert hasattr(p, '_paths')
@ -81,7 +81,7 @@ def test_pather_accepts_and_reuses_planner_instance() -> None:
p = Pather(
Library(),
tools=PathTool(layer=(1, 0), width=1),
auto_render=False,
render='deferred',
planner=planner,
)
p.ports['A'] = Port((0, 0), rotation=0, ptype='wire')
@ -148,7 +148,7 @@ def test_pather_dead_ports() -> None:
def test_pather_trace_basic() -> None:
lib = Library()
tool = PathTool(layer='M1', width=1000)
p = Pather(lib, tools=tool, auto_render=False)
p = Pather(lib, tools=tool)
# Routing extends opposite the port's inward-facing rotation.
p.pattern.ports['A'] = Port((0, 0), rotation=0)
@ -164,7 +164,7 @@ def test_pather_trace_basic() -> None:
def test_pather_trace_to() -> None:
lib = Library()
tool = PathTool(layer='M1', width=1000)
p = Pather(lib, tools=tool, auto_render=False)
p = Pather(lib, tools=tool)
p.pattern.ports['A'] = Port((0, 0), rotation=0)
@ -177,7 +177,7 @@ def test_pather_trace_to() -> None:
def test_pather_bundle_trace() -> None:
lib = Library()
tool = PathTool(layer='M1', width=1000)
p = Pather(lib, tools=tool, auto_render=False)
p = Pather(lib, tools=tool)
p.pattern.ports['A'] = Port((0, 0), rotation=0)
p.pattern.ports['B'] = Port((0, 2000), rotation=0)
@ -194,13 +194,13 @@ def test_pather_bundle_trace() -> None:
def test_portpather_default_spacing_matches_explicit_spacing() -> None:
lib_default = Library()
tool_default = PathTool(layer='M1', width=1000)
p_default = Pather(lib_default, tools=tool_default, auto_render=False)
p_default = Pather(lib_default, tools=tool_default)
p_default.pattern.ports['A'] = Port((0, 0), rotation=0)
p_default.pattern.ports['B'] = Port((0, 2000), rotation=0)
lib_explicit = Library()
tool_explicit = PathTool(layer='M1', width=1000)
p_explicit = Pather(lib_explicit, tools=tool_explicit, auto_render=False)
p_explicit = Pather(lib_explicit, tools=tool_explicit)
p_explicit.pattern.ports['A'] = Port((0, 0), rotation=0)
p_explicit.pattern.ports['B'] = Port((0, 2000), rotation=0)
@ -211,11 +211,11 @@ def test_portpather_default_spacing_matches_explicit_spacing() -> None:
assert_allclose(p_default.pattern.ports['B'].offset, p_explicit.pattern.ports['B'].offset)
def test_portpather_default_spacing_reused_and_overridden() -> None:
p_default = Pather(Library(), tools=PathTool(layer='M1', width=1000), auto_render=False)
p_default = Pather(Library(), tools=PathTool(layer='M1', width=1000))
p_default.pattern.ports['A'] = Port((0, 0), rotation=0)
p_default.pattern.ports['B'] = Port((0, 2000), rotation=0)
p_explicit = Pather(Library(), tools=PathTool(layer='M1', width=1000), auto_render=False)
p_explicit = Pather(Library(), tools=PathTool(layer='M1', width=1000))
p_explicit.pattern.ports['A'] = Port((0, 0), rotation=0)
p_explicit.pattern.ports['B'] = Port((0, 2000), rotation=0)
@ -227,11 +227,11 @@ def test_portpather_default_spacing_reused_and_overridden() -> None:
assert_allclose(p_default.pattern.ports['A'].offset, p_explicit.pattern.ports['A'].offset)
assert_allclose(p_default.pattern.ports['B'].offset, p_explicit.pattern.ports['B'].offset)
p_override = Pather(Library(), tools=PathTool(layer='M1', width=1000), auto_render=False)
p_override = Pather(Library(), tools=PathTool(layer='M1', width=1000))
p_override.pattern.ports['A'] = Port((0, 0), rotation=0)
p_override.pattern.ports['B'] = Port((0, 2000), rotation=0)
p_override_explicit = Pather(Library(), tools=PathTool(layer='M1', width=1000), auto_render=False)
p_override_explicit = Pather(Library(), tools=PathTool(layer='M1', width=1000))
p_override_explicit.pattern.ports['A'] = Port((0, 0), rotation=0)
p_override_explicit.pattern.ports['B'] = Port((0, 2000), rotation=0)
@ -244,7 +244,7 @@ def test_portpather_default_spacing_reused_and_overridden() -> None:
def test_portpather_default_spacing_not_injected_for_straight_bundle() -> None:
lib = Library()
tool = PathTool(layer='M1', width=1000)
p = Pather(lib, tools=tool, auto_render=False)
p = Pather(lib, tools=tool)
p.pattern.ports['A'] = Port((0, 0), rotation=0)
p.pattern.ports['B'] = Port((0, 2000), rotation=0)
@ -256,7 +256,7 @@ def test_portpather_default_spacing_not_injected_for_straight_bundle() -> None:
def test_portpather_default_spacing_vector_revalidated_after_selection_change() -> None:
lib = Library()
tool = PathTool(layer='M1', width=1000)
p = Pather(lib, tools=tool, auto_render=False)
p = Pather(lib, tools=tool)
p.pattern.ports['A'] = Port((0, 0), rotation=0)
p.pattern.ports['B'] = Port((0, 2000), rotation=0)
p.pattern.ports['C'] = Port((0, 4000), rotation=0)
@ -270,7 +270,7 @@ def test_portpather_default_spacing_vector_revalidated_after_selection_change()
def test_pather_each_bound() -> None:
lib = Library()
tool = PathTool(layer='M1', width=1000)
p = Pather(lib, tools=tool, auto_render=False)
p = Pather(lib, tools=tool)
p.pattern.ports['A'] = Port((0, 0), rotation=0)
p.pattern.ports['B'] = Port((-1000, 2000), rotation=0)
@ -390,7 +390,7 @@ def test_rename() -> None:
def test_pather_dead_fallback_preserves_out_ptype() -> None:
lib = Library()
tool = PathTool(layer='M1', width=1000, ptype='wire')
p = Pather(lib, tools=tool, auto_render=False)
p = Pather(lib, tools=tool)
p.pattern.ports['A'] = Port((0, 0), rotation=0, ptype='wire')
p.set_dead()
@ -424,7 +424,7 @@ def test_pather_dead_fallback_does_not_hide_fatal_route_errors() -> None:
commit_planner=lambda length: {'length': length},
),)
p = Pather(Library(), tools=MismatchedEndpointTool(layer='M1', width=1000, ptype='wire'), auto_render=False)
p = Pather(Library(), tools=MismatchedEndpointTool(layer='M1', width=1000, ptype='wire'))
p.pattern.ports['A'] = Port((0, 0), rotation=0, ptype='wire')
p.set_dead()
@ -472,7 +472,7 @@ def test_pather_valid_candidate_does_not_hide_fatal_route_errors() -> None:
),
)
p = Pather(Library(), tools=PartlyMismatchedEndpointTool(layer='M1', width=1000, ptype='wire'), auto_render=False)
p = Pather(Library(), tools=PartlyMismatchedEndpointTool(layer='M1', width=1000, ptype='wire'))
p.pattern.ports['A'] = Port((0, 0), rotation=0, ptype='wire')
with pytest.raises(BuildError, match='does not match declared offer out_ptype'):

View file

@ -10,7 +10,7 @@ from masque.error import PortError, PatternError
def test_pather_place_treeview_resolves_once() -> None:
lib = Library()
tool = PathTool(layer='M1', width=1000)
p = Pather(lib, tools=tool)
p = Pather(lib, tools=tool, render='deferred')
tree = {'child': Pattern(ports={'B': Port((1, 0), pi)})}
@ -24,7 +24,7 @@ def test_pather_place_treeview_resolves_once() -> None:
def test_pather_plug_treeview_resolves_once() -> None:
lib = Library()
tool = PathTool(layer='M1', width=1000)
p = Pather(lib, tools=tool)
p = Pather(lib, tools=tool, render='deferred')
p.pattern.ports['A'] = Port((0, 0), rotation=0)
tree = {'child': Pattern(ports={'B': Port((0, 0), pi)})}
@ -39,7 +39,7 @@ def test_pather_plug_treeview_resolves_once() -> None:
def test_pather_failed_plug_does_not_add_break_marker() -> None:
lib = Library()
tool = PathTool(layer='M1', width=1000)
p = Pather(lib, tools=tool, auto_render=False)
p = Pather(lib, tools=tool, render='deferred')
p.pattern.annotations = {'k': [1]}
p.pattern.ports['A'] = Port((0, 0), rotation=0)
@ -60,7 +60,7 @@ def test_pather_failed_plug_does_not_add_break_marker() -> None:
def test_pather_place_reused_deleted_name_keeps_break_marker() -> None:
lib = Library()
tool = PathTool(layer='M1', width=1000)
p = Pather(lib, tools=tool, auto_render=False)
p = Pather(lib, tools=tool, render='deferred')
p.pattern.ports['A'] = Port((0, 0), rotation=0)
p.at('A').straight(5000)
@ -80,7 +80,7 @@ def test_pather_place_reused_deleted_name_keeps_break_marker() -> None:
def test_pather_plug_reused_deleted_name_keeps_break_marker() -> None:
lib = Library()
tool = PathTool(layer='M1', width=1000)
p = Pather(lib, tools=tool, auto_render=False)
p = Pather(lib, tools=tool, render='deferred')
p.pattern.ports['A'] = Port((0, 0), rotation=0)
p.pattern.ports['B'] = Port((0, 0), rotation=0)
@ -107,7 +107,7 @@ def test_pather_plug_reused_deleted_name_keeps_break_marker() -> None:
def test_pather_failed_plugged_does_not_add_break_marker() -> None:
lib = Library()
tool = PathTool(layer='M1', width=1000)
p = Pather(lib, tools=tool, auto_render=False)
p = Pather(lib, tools=tool, render='deferred')
p.pattern.ports['A'] = Port((0, 0), rotation=0)
p.at('A').straight(5000)

View file

@ -360,7 +360,7 @@ def test_pather_treats_notimplemented_offer_query_as_no_offers() -> None:
_ = kind, in_ptype, out_ptype, kwargs
raise NotImplementedError
p = Pather(Library(), tools=NoOfferTool(), auto_render=False)
p = Pather(Library(), tools=NoOfferTool())
p.ports['A'] = Port((0, 0), rotation=0, ptype='wire')
with pytest.raises(BuildError, match='No legal primitive offer for trace'):
@ -381,7 +381,7 @@ def test_pather_propagates_offer_query_errors(err_type: type[Exception]) -> None
_ = kind, in_ptype, out_ptype, kwargs
raise err_type('offer query failed')
p = Pather(Library(), tools=BrokenTool(), auto_render=False)
p = Pather(Library(), tools=BrokenTool())
p.ports['A'] = Port((0, 0), rotation=0, ptype='wire')
with pytest.raises(err_type):
@ -413,7 +413,7 @@ def test_pather_selects_lowest_cost_offer() -> None:
StraightOffer(in_ptype=in_ptype, out_ptype=out_ptype, priority_bias=0, **offer_callbacks(low)),
)
p = Pather(Library(), tools=MultiOfferTool(), auto_render=False)
p = Pather(Library(), tools=MultiOfferTool(), render='deferred')
p.ports['A'] = Port((0, 0), rotation=0, ptype='wire')
p.straight('A', 7)
@ -422,6 +422,53 @@ def test_pather_selects_lowest_cost_offer() -> None:
assert numpy.allclose(p.ports['A'].offset, (-7, 0))
def test_solver_rejects_rotation_impossible_candidates_before_parameter_solving() -> None:
invalid_parameters: list[float] = []
class RotationOfferTool(PlanningOnlyTool):
def primitive_offers(
self,
kind: Literal['straight', 'bend', 's', 'u'],
*,
in_ptype: str | None = None,
out_ptype: str | None = None,
**kwargs: Any,
) -> tuple[PrimitiveOffer, ...]:
_ = kwargs
if kind != 'straight':
return ()
def invalid_endpoint(length: float) -> Port:
invalid_parameters.append(length)
return Port((length, 0), rotation=0, ptype=out_ptype or in_ptype)
def valid_endpoint(length: float) -> Port:
return Port((length, 0), rotation=pi, ptype=out_ptype or in_ptype)
return (
StraightOffer(
in_ptype=in_ptype,
out_ptype=out_ptype,
endpoint_planner=invalid_endpoint,
commit_planner=lambda length: {'kind': 'invalid', 'length': length},
),
StraightOffer(
in_ptype=in_ptype,
out_ptype=out_ptype,
endpoint_planner=valid_endpoint,
commit_planner=lambda length: {'kind': 'valid', 'length': length},
),
)
p = Pather(Library(), tools=RotationOfferTool(), render='deferred')
p.ports['A'] = Port((0, 0), rotation=0, ptype='wire')
p.straight('A', 7)
assert p._paths['A'][0].data == {'kind': 'valid', 'length': 7}
assert invalid_parameters == [0.0, 0.0]
def test_pather_commits_only_selected_offer() -> None:
committed: list[str] = []
@ -456,7 +503,7 @@ def test_pather_commits_only_selected_offer() -> None:
return (make('expensive', 100), make('cheap', 0))
p = Pather(Library(), tools=RecordingTool(), auto_render=False)
p = Pather(Library(), tools=RecordingTool(), render='deferred')
p.ports['A'] = Port((0, 0), rotation=0, ptype='wire')
p.straight('A', 5)
@ -506,7 +553,7 @@ def test_pather_routes_with_offer_only_s_and_u_tool() -> None:
),)
return ()
p = Pather(Library(), tools=OfferOnlyTool(), auto_render=False)
p = Pather(Library(), tools=OfferOnlyTool(), render='deferred')
p.ports['A'] = Port((0, 0), rotation=0, ptype='wire')
p.jog('A', 3)
p.uturn('A', 4)

View file

@ -1,4 +1,5 @@
from typing import TYPE_CHECKING, cast
import logging
import pytest
import numpy
@ -20,10 +21,27 @@ if TYPE_CHECKING:
def deferred_render_setup() -> tuple[Pather, PathTool, Library]:
lib = Library()
tool = PathTool(layer=(1, 0), width=2, ptype="wire")
rp = Pather(lib, tools=tool, auto_render=False)
rp = Pather(lib, tools=tool, render='deferred')
rp.ports["start"] = Port((0, 0), pi / 2, ptype="wire")
return rp, tool, lib
def add_pending_straight(pather: Pather) -> None:
pather.ports["start"] = Port((0, 0), pi / 2, ptype="wire")
pather.straight("start", 10)
def route_in_context(pather: Pather) -> None:
with pather:
add_pending_straight(pather)
def fail_in_context(pather: Pather) -> None:
with pather:
add_pending_straight(pather)
raise RuntimeError('body failed')
def test_deferred_render_stores_pending_paths_until_render(deferred_render_setup: tuple[Pather, PathTool, Library]) -> None:
rp, tool, lib = deferred_render_setup
rp.at("start").straight(10).straight(10)
@ -101,7 +119,7 @@ def test_portpather_translate_only_affects_future_steps(deferred_render_setup: t
def test_deferred_render_dead_ports() -> None:
lib = Library()
tool = PathTool(layer=(1, 0), width=1)
rp = Pather(lib, ports={"in": Port((0, 0), 0)}, tools=tool, auto_render=False)
rp = Pather(lib, ports={"in": Port((0, 0), 0)}, tools=tool, render='deferred')
rp.set_dead()
rp.straight("in", -10)
@ -113,6 +131,91 @@ def test_deferred_render_dead_ports() -> None:
rp.render()
assert not rp.pattern.has_shapes()
def test_pather_default_auto_policy_renders_immediately_outside_context() -> None:
lib = Library()
tool = PathTool(layer=(1, 0), width=2, ptype="wire")
p = Pather(lib, tools=tool)
p.ports["start"] = Port((0, 0), pi / 2, ptype="wire")
p.straight("start", 10)
assert p.pattern.has_shapes()
assert not any(p._paths.values())
def test_pather_default_auto_policy_defers_until_clean_context_exit() -> None:
lib = Library()
tool = PathTool(layer=(1, 0), width=2, ptype="wire")
with Pather(lib, tools=tool) as p:
add_pending_straight(p)
assert not p.pattern.has_shapes()
assert len(p._paths["start"]) == 1
assert p.pattern.has_shapes()
assert not any(p._paths.values())
def test_pather_context_warn_policy_logs_pending_paths(caplog: pytest.LogCaptureFixture) -> None:
lib = Library()
tool = PathTool(layer=(1, 0), width=2, ptype="wire")
with caplog.at_level(logging.WARNING, logger="masque.builder.pather"), Pather(lib, tools=tool, render='warn') as p:
add_pending_straight(p)
records = [
record
for record in caplog.records
if 'Pather context exited with 1 pending render step on 1 port' in record.getMessage()
]
assert len(records) == 1
assert records[0].stack_info is None
assert len(p._paths["start"]) == 1
assert not p.pattern.has_shapes()
def test_pather_context_error_policy_rejects_pending_paths() -> None:
lib = Library()
tool = PathTool(layer=(1, 0), width=2, ptype="wire")
p = Pather(lib, tools=tool, render='error')
with pytest.raises(BuildError, match='1 pending render step on 1 port'):
route_in_context(p)
assert len(p._paths["start"]) == 1
assert not p.pattern.has_shapes()
def test_pather_context_ignore_policy_leaves_pending_paths_silent(caplog: pytest.LogCaptureFixture) -> None:
lib = Library()
tool = PathTool(layer=(1, 0), width=2, ptype="wire")
with caplog.at_level(logging.WARNING, logger="masque.builder.pather"), Pather(lib, tools=tool, render='ignore') as p:
add_pending_straight(p)
assert not caplog.records
assert len(p._paths["start"]) == 1
assert not p.pattern.has_shapes()
def test_pather_context_policy_does_not_mask_body_exception() -> None:
lib = Library()
tool = PathTool(layer=(1, 0), width=2, ptype="wire")
p = Pather(lib, tools=tool, render='error')
with pytest.raises(RuntimeError, match='body failed'):
fail_in_context(p)
assert len(p._paths["start"]) == 1
assert not p.pattern.has_shapes()
def test_pather_rejects_invalid_render_policy() -> None:
with pytest.raises(BuildError, match='Invalid render policy'):
Pather(Library(), tools=PathTool(layer=(1, 0), width=2), render='later') # type: ignore[arg-type]
def test_deferred_render_rename_port(deferred_render_setup: tuple[Pather, PathTool, Library]) -> None:
rp, tool, lib = deferred_render_setup
rp.at("start").straight(10)
@ -174,7 +277,7 @@ def test_pathtool_s_offer_render_geometry_matches_ports() -> None:
def test_deferred_render_uturn_fallback() -> None:
lib = Library()
tool = PathTool(layer='M1', width=1000)
rp = Pather(lib, tools=tool, auto_render=False)
rp = Pather(lib, tools=tool, render='deferred')
rp.pattern.ports['A'] = Port((0, 0), rotation=0)
rp.at('A').uturn(offset=10000, length=5000)
@ -213,7 +316,7 @@ def test_pather_render_auto_renames_single_use_tool_children() -> None:
return tree
lib = Library()
p = Pather(lib, tools=FullTreeTool(), auto_render=False)
p = Pather(lib, tools=FullTreeTool())
p.pattern.ports['A'] = Port((0, 0), rotation=0, ptype='wire')
p.straight('A', 10)
@ -258,7 +361,7 @@ def test_custom_tool_render_preserves_segment_subtrees() -> None:
return batch[0].data
lib = Library()
p = Pather(lib, tools=TraceTreeTool(), auto_render=False)
p = Pather(lib, tools=TraceTreeTool())
p.pattern.ports['A'] = Port((0, 0), rotation=0, ptype='wire')
p.straight('A', 10)
@ -293,7 +396,7 @@ def test_pather_render_rejects_missing_single_use_tool_refs() -> None:
lib = Library()
lib['_seg'] = Pattern(annotations={'stale': [1]})
p = Pather(lib, tools=MissingSingleUseTool(), auto_render=False)
p = Pather(lib, tools=MissingSingleUseTool(), render='deferred')
p.pattern.ports['A'] = Port((0, 0), rotation=0, ptype='wire')
p.straight('A', 10)
@ -329,7 +432,7 @@ def test_pather_render_allows_missing_non_single_use_tool_refs() -> None:
lib = Library()
lib['shared'] = Pattern(annotations={'shared': [1]})
p = Pather(lib, tools=SharedRefTool(), auto_render=False)
p = Pather(lib, tools=SharedRefTool())
p.pattern.ports['A'] = Port((0, 0), rotation=0, ptype='wire')
p.straight('A', 10)
@ -362,7 +465,7 @@ def test_pather_render_rejects_output_port_that_misses_planned_endpoint(append:
return tree
lib = Library()
p = Pather(lib, tools=WrongEndpointTool(), auto_render=False)
p = Pather(lib, tools=WrongEndpointTool(), render='deferred')
p.pattern.ports['A'] = Port((0, 0), rotation=0, ptype='wire')
p.straight('A', 10)
@ -396,7 +499,7 @@ def test_pather_render_rejects_output_port_with_wrong_ptype(append: bool) -> Non
return tree
lib = Library()
p = Pather(lib, tools=WrongPtypeTool(), auto_render=False)
p = Pather(lib, tools=WrongPtypeTool(), render='deferred')
p.pattern.ports['A'] = Port((0, 0), rotation=0, ptype='wire')
p.straight('A', 10)
@ -409,7 +512,7 @@ def test_pather_render_rejects_output_port_with_wrong_ptype(append: bool) -> Non
def test_deferred_render_rename_to_none_keeps_pending_geometry_without_port() -> None:
lib = Library()
tool = PathTool(layer='M1', width=1000)
rp = Pather(lib, tools=tool, auto_render=False)
rp = Pather(lib, tools=tool, render='deferred')
rp.pattern.ports['A'] = Port((0, 0), rotation=0)
rp.at('A').straight(5000)

View file

@ -6,6 +6,8 @@ from numpy import pi
from numpy.testing import assert_equal
from masque import Library, PathTool, Port, Pather
from masque.builder.planner import PreparedRouteResult, RoutePlanningError, RoutePortContext, RoutingPlanner
from masque.builder.planner.planner import Candidate, RouteRequest
from masque.builder.tools import BendOffer, PrimitiveOffer, StraightOffer, Tool
from masque.error import BuildError, PortError
@ -14,7 +16,7 @@ from masque.error import BuildError, PortError
def trace_into_setup() -> tuple[Pather, PathTool, Library]:
lib = Library()
tool = PathTool(layer=(1, 0), width=2, ptype="wire")
p = Pather(lib, tools=tool, auto_render=True, auto_render_append=False)
p = Pather(lib, tools=tool, render='immediate', render_append=False)
return p, tool, lib
@ -69,7 +71,7 @@ def test_path_into_thru(trace_into_setup: tuple[Pather, PathTool, Library]) -> N
def test_pather_trace_into_shapes() -> None:
lib = Library()
tool = PathTool(layer='M1', width=1000)
p = Pather(lib, tools=tool, auto_render=False)
p = Pather(lib, tools=tool)
p.pattern.ports['A'] = Port((0, 0), rotation=0)
p.pattern.ports['B'] = Port((-10000, 0), rotation=pi)
@ -173,7 +175,7 @@ def test_pather_trace_into_refines_output_adapter_route_before_plug() -> None:
pat.add_port_pair(names=port_names, ptype=batch[-1].end_port.ptype if batch else 'm1wire')
return tree
p = Pather(Library(), tools=TransitionTool(), auto_render=False)
p = Pather(Library(), tools=TransitionTool(), render='deferred')
p.pattern.ports['src'] = Port((-65000, -11500), rotation=pi / 2, ptype='m1wire')
p.pattern.ports['dst'] = Port((-100000, -100000), rotation=pi, ptype='m2wire')
@ -186,7 +188,7 @@ def test_pather_trace_into_refines_output_adapter_route_before_plug() -> None:
def test_pather_trace_into_dead_updates_ports_without_geometry() -> None:
lib = Library()
tool = PathTool(layer='M1', width=1000, ptype='wire')
p = Pather(lib, tools=tool, auto_render=False)
p = Pather(lib, tools=tool)
p.pattern.ports['A'] = Port((0, 0), rotation=0, ptype='wire')
p.pattern.ports['B'] = Port((-10000, 0), rotation=pi, ptype='wire')
p.set_dead()
@ -260,3 +262,90 @@ def test_pather_trace_into_rejects_reserved_route_kwargs(
assert p.pattern.ports['B'].rotation is not None
assert numpy.isclose(p.pattern.ports['B'].rotation, dst.rotation)
assert len(p._paths['A']) == 0
class TraceIntoBudgetSolver:
def __init__(self, successes: set[tuple[int, int]], fatal_at: set[tuple[int, int]] | None = None) -> None:
self.successes = successes
self.fatal_at = set() if fatal_at is None else fatal_at
self.attempts: list[tuple[int, int]] = []
def solve(
self,
*,
min_bends: int = 0,
max_bends: int | None = None,
) -> Candidate:
assert max_bends is not None
band = (min_bends, max_bends)
self.attempts.append(band)
if band in self.fatal_at:
raise RoutePlanningError('fatal', fatal=True)
if band not in self.successes:
raise BuildError('try next budget')
return Candidate((), Port((0, 0), rotation=0, ptype='wire'), 0.0, 0, 0.0)
class TraceIntoBudgetPlanner(RoutingPlanner):
def __init__(self, successes: set[tuple[int, int]], fatal_at: set[tuple[int, int]] | None = None) -> None:
self.solver = TraceIntoBudgetSolver(successes, fatal_at=fatal_at)
self.solver_requests = 0
def solver_for_request(self, request: RouteRequest) -> Any:
_ = request
self.solver_requests += 1
return self.solver
def prepared_result_from_legs(
self,
legs: Any,
*,
renames: tuple[tuple[str, str], ...] = (),
) -> PreparedRouteResult:
_ = legs, renames
return PreparedRouteResult(())
@pytest.mark.parametrize(
('dst', 'successes', 'attempts'),
[
(Port((-10, 0), rotation=pi, ptype='wire'), {(0, 2)}, [(0, 2)]),
(Port((-10, 0), rotation=pi, ptype='wire'), {(4, 4)}, [(0, 2), (4, 4)]),
(Port((-10, -10), rotation=3 * pi / 2, ptype='wire'), {(1, 1)}, [(1, 1)]),
(Port((-10, -10), rotation=3 * pi / 2, ptype='wire'), {(3, 3)}, [(1, 1), (3, 3)]),
],
)
def test_trace_into_reuses_solver_across_staged_bend_bands(
dst: Port,
successes: set[tuple[int, int]],
attempts: list[tuple[int, int]],
) -> None:
planner = TraceIntoBudgetPlanner(successes)
context = RoutePortContext('src', Port((0, 0), rotation=0, ptype='wire'), PathTool(layer='M1', width=1, ptype='wire'))
planner.plan_trace_into(context, 'dst', dst, out_ptype=None, plug_destination=True, thru=None)
assert planner.solver.attempts == attempts
assert planner.solver_requests == 1
def test_trace_into_staged_bend_budget_stops_on_fatal_error() -> None:
planner = TraceIntoBudgetPlanner({(4, 4)}, fatal_at={(0, 2)})
context = RoutePortContext('src', Port((0, 0), rotation=0, ptype='wire'), PathTool(layer='M1', width=1, ptype='wire'))
with pytest.raises(RoutePlanningError, match='fatal'):
planner.plan_trace_into(context, 'dst', Port((-10, 0), rotation=pi, ptype='wire'), out_ptype=None, plug_destination=True, thru=None)
assert planner.solver.attempts == [(0, 2)]
assert planner.solver_requests == 1
def test_trace_into_bend_bands_respect_max_bends() -> None:
class OneBendPlanner(RoutingPlanner):
TRACE_INTO_MAX_BENDS = 1
planner = OneBendPlanner()
assert planner.trace_into_bend_bands('straight') == ((0, 0),)
assert planner.trace_into_bend_bands('s') == ((0, 0),)
assert planner.trace_into_bend_bands('bend') == ((1, 1),)