Source code for jitx.run.layout

"""The physical design (layout) of a design held by the runtime.

Reached through ``RuntimeDesign.layout``. Requests change the design in the
runtime, as if made in the UI. Placements are applied to the local design tree
from the reply; routes, vias, control points, and area regions appear in it
when the runtime reports them. ``monitor`` registers callbacks on changes
(see :mod:`jitx.run.monitor`), and ``emit`` sends user events.
"""

from __future__ import annotations

from collections.abc import Awaitable, Callable, Collection, Iterable, Sequence
from types import TracebackType
from enum import StrEnum
from typing import cast, overload

import jitxcore._proto.enums_pb2 as epb2
import jitxcore._proto.physical_design_pb2 as pdpb2
import jitxcore._proto.route_config_pb2 as rcpb2
import jitxcore._proto.shapes_pb2 as spb2
import jitxcore._proto.stable_id_pb2 as sipb2
import shapely

from jitx._instantiation import instantiation
from jitx._structural import Proxied, Proxy, Structurable, dispose
from jitx.circuit import Route
from jitx.controlpoint import ControlPoint
from jitx.component import Component
from jitx.copper import Pour
from jitx.feature import (
    Courtyard,
    Custom,
    Cutout,
    Feature,
    Finish,
    Glue,
    KeepOut,
    Paste,
    Silkscreen,
    Soldermask,
    SurfaceFeature,
)
from jitx.inspect import extract
from jitx.landpattern import Pad, PadMapping
from jitx.layerindex import LayerSet, Side
from jitx.net import DiffPair, Net, Port
from jitx.placement import Placement
from jitx.proxy import origin, typeof
from jitx.shapes import Shape
from jitx.shapes.hull import ConvexHull
from jitx.shapes.shapely import ShapelyGeometry
from jitx.transform import IDENTITY, Point
from jitx.via import Via

from .._translate.idmap import (
    CompositeKey,
    ReverseMapper,
    composite_id,
    composite_key,
    stable_id_key,
    stable_id_proto,
)
from .._translate.reverse_flow.placement import (
    placement_from_pose_proto,
    placement_from_pose_side_proto,
)
from .._translate.reverse_flow.resolve import resolve_key
from .monitor import LayoutMonitor, UserEvent, _Monitors, user_event_proto


[docs] class RequestRejected(RuntimeError): """The runtime refused a request. Args: kind: The runtime's error kind. message: The runtime's reason. operation: For a batch, the index of the operation the runtime refused, counted in the order the batch sent them; None if the refusal is not about one operation. """ def __init__(self, kind: str, message: str, operation: int | None = None): super().__init__(f"{message} ({kind})" if kind else message) self.kind = kind self.message = message self.operation = operation
[docs] class RouteSetting(StrEnum): """A route setting, as named by the ``reset`` argument of a modification. >>> rd.layout.modify(route, reset=[RouteSetting.Width]) """ Width = "width" """The trace width.""" Space = "space" """The trace clearance.""" PairSpace = "pair_space" """The differential pair spacing.""" EnvelopeOverhead = "envelope_overhead" """The extra width of the length matching region."""
_RESET_FIELDS: dict[RouteSetting, pdpb2.RouteConfigField] = { RouteSetting.Width: pdpb2.ROUTE_CONFIG_TRACE_WIDTH, RouteSetting.Space: pdpb2.ROUTE_CONFIG_TRACE_SPACE, RouteSetting.PairSpace: pdpb2.ROUTE_CONFIG_PAIR_SPACE, RouteSetting.EnvelopeOverhead: pdpb2.ROUTE_CONFIG_SQUIGGLE_ROOM, } class _Pending: # A handle resolved from its operation's result once the batch is sent. def _resolve(self, result: pdpb2.PhysicalDesignOperationResult) -> None: raise NotImplementedError class _Done(_Pending): # An operation without a handle (delete, modify). def _resolve(self, result: pdpb2.PhysicalDesignOperationResult) -> None: pass
[docs] class PendingPlacement: """The resulting placement of a component placed in a batch, available once the batch has been sent.""" def __init__(self, target: Component): self.target = target self.__placement: Placement | None = None @property def done(self) -> bool: """Whether the batch was sent and the placement is available.""" return self.__placement is not None @property def placement(self) -> Placement: """The board-absolute placement the runtime reported. Raises: RuntimeError: if the batch has not been sent. """ if self.__placement is None: raise RuntimeError("placement is pending until the batch is sent") return self.__placement def _resolve(self, placement: Placement) -> None: self.__placement = placement
class _Placements(_Pending): # The placements of one PlaceComponents operation, by component. def __init__(self): self.pending: dict[CompositeKey, PendingPlacement] = {} def _resolve(self, result: pdpb2.PhysicalDesignOperationResult) -> None: placements = { composite_key(stable_id_key(sid) for sid in ip.instance.ids): ip for ip in result.place_components.placements } missing = [ str(p.target) for key, p in self.pending.items() if key not in placements ] if missing: raise RuntimeError( f"runtime reported no placement for {', '.join(missing)}" ) for key, p in self.pending.items(): ip = placements[key] p._resolve(placement_from_pose_side_proto(ip.pose, ip.side))
[docs] class PendingSync(_Pending): """The outcome of a sync in a batch, available once the batch has been sent.""" def __init__(self): self.__settled: bool | None = None @property def done(self) -> bool: """Whether the batch was sent and the outcome is available.""" return self.__settled is not None @property def settled(self) -> bool: """Whether background tasks had finished; False if the sync timed out. Raises: RuntimeError: if the batch has not been sent. """ if self.__settled is None: raise RuntimeError("sync is pending until the batch is sent") return self.__settled def _resolve(self, result: pdpb2.PhysicalDesignOperationResult) -> None: self.__settled = result.sync.settled
[docs] class PendingRoute(_Pending): """A route created by a request. Its runtime id is available once the batch has been sent; the :class:`~jitx.circuit.Route` object once the runtime reports it routed or blocked (follow the request with a sync to wait for that), or after a capture.""" def __init__(self, rmap: Callable[[], ReverseMapper]): self.__rmap = rmap self.__key: CompositeKey | None = None @property def done(self) -> bool: """Whether the batch was sent and the route id is available.""" return self.__key is not None @property def key(self) -> CompositeKey: """The runtime id of the route. Raises: RuntimeError: if the batch has not been sent. """ if self.__key is None: raise RuntimeError("route is pending until the batch is sent") return self.__key @property def route(self) -> Route | None: """The route in the current design tree, or None if the runtime has not reported it yet. Its ``status`` tells whether it is routed or blocked.""" if self.__key is None: return None obj = resolve_key(self.__key, self.__rmap(), False, False, expect=Route) return obj if isinstance(obj, Route) else None def _set(self, key: CompositeKey) -> None: self.__key = key def _resolve(self, result: pdpb2.PhysicalDesignOperationResult) -> None: self.__key = (stable_id_key(result.create_route.route),)
[docs] class PendingVia(_Pending): """A via created or updated by a request. Its runtime id and position are available once the batch has been sent; the :class:`~jitx.via.Via` object once the runtime reports it (follow the request with a sync to wait for that), or after a capture.""" def __init__( self, rmap: Callable[[], ReverseMapper], key: CompositeKey | None = None ): self.__rmap = rmap self.__key = key self.__position: Point | None = None self.__route = PendingRoute(rmap) @property def done(self) -> bool: """Whether the batch was sent and the via id and position are available.""" return self.__position is not None @property def key(self) -> CompositeKey: """The runtime id of the via. Raises: RuntimeError: if the batch has not been sent. """ if self.__key is None: raise RuntimeError("via is pending until the batch is sent") return self.__key @property def position(self) -> Point: """The board-absolute position of the via, after the runtime moved it to a free spot, if it did. Raises: RuntimeError: if the batch has not been sent. """ if self.__position is None: raise RuntimeError("via is pending until the batch is sent") return self.__position @property def route(self) -> PendingRoute | None: """The route from the via's origin to the via; None if they coincide or the via was updated rather than created.""" return self.__route if self.__route.done else None @property def via(self) -> Via | None: """The via in the current design tree, or None if the runtime has not reported it yet.""" if self.__key is None: return None obj = resolve_key(self.__key, self.__rmap(), False, True, expect=Via) return obj if isinstance(obj, Via) else None def _set(self, key: CompositeKey, position: spb2.Point) -> None: self.__key = key self.__position = (position.x, position.y) def _set_route(self, route: CompositeKey) -> None: self.__route._set(route)
class _CreatedVias(_Pending): def __init__(self, vias: Sequence[PendingVia]): self.vias = vias def _resolve(self, result: pdpb2.PhysicalDesignOperationResult) -> None: created = result.create_vias.vias if len(created) != len(self.vias): raise RuntimeError( f"runtime reported {len(created)} vias for {len(self.vias)}" ) for pending, cv in zip(self.vias, created, strict=True): pending._set( composite_key(stable_id_key(s) for s in cv.via.ids), cv.position ) if cv.HasField("route"): pending._set_route((stable_id_key(cv.route),)) class _UpdatedVias(_Pending): def __init__(self, vias: Sequence[PendingVia]): self.vias = vias def _resolve(self, result: pdpb2.PhysicalDesignOperationResult) -> None: updated = result.update_vias.vias if len(updated) != len(self.vias): raise RuntimeError( f"runtime reported {len(updated)} vias for {len(self.vias)}" ) for pending, uv in zip(self.vias, updated, strict=True): pending._set( composite_key(stable_id_key(s) for s in uv.via.ids), uv.position )
[docs] class PendingControlPoint(_Pending): """A control point created or updated by a request. Its runtime id, and the routes made to it, are available once the batch has been sent; the control point object once the runtime reports it (follow the request with a sync to wait for that), or after a capture.""" def __init__( self, rmap: Callable[[], ReverseMapper], key: CompositeKey | None = None ): self.__rmap = rmap self.__key = key self.__routes: list[PendingRoute] = [] self.__placement: Placement | None = None self.__done = False @property def done(self) -> bool: """Whether the batch was sent and the id is available.""" return self.__done @property def key(self) -> CompositeKey: """The runtime id of the control point. Raises: RuntimeError: if the batch has not been sent. """ if self.__key is None: raise RuntimeError("control point is pending until the batch is sent") return self.__key @property def routes(self) -> Sequence[PendingRoute]: """The routes made from the origin(s) to the control point when it was created; empty for an update.""" return self.__routes @property def placement(self) -> Placement | None: """For an update, the resulting board-absolute placement.""" return self.__placement @property def control_point(self) -> ControlPoint | None: """The control point in the current design tree, or None if the runtime has not reported it yet.""" if self.__key is None: return None obj = resolve_key(self.__key, self.__rmap(), False, True, expect=ControlPoint) return obj if isinstance(obj, ControlPoint) else None def _set(self, key: CompositeKey, routes: Iterable[CompositeKey]) -> None: self.__key = key for route in routes: pending = PendingRoute(self.__rmap) pending._set(route) self.__routes.append(pending) self.__done = True def _set_placement(self, key: CompositeKey, placement: Placement) -> None: self.__key = key self.__placement = placement self.__done = True
class _CreatedControls(_Pending): def __init__(self, controls: Sequence[PendingControlPoint], field: str): self.controls = controls self.field = field def _resolve(self, result: pdpb2.PhysicalDesignOperationResult) -> None: created = getattr(result, self.field).controls if len(created) != len(self.controls): raise RuntimeError( f"runtime reported {len(created)} control points for {len(self.controls)}" ) for pending, cc in zip(self.controls, created, strict=True): pending._set( composite_key(stable_id_key(s) for s in cc.control.ids), [(stable_id_key(r),) for r in cc.routes], ) class _UpdatedControls(_Pending): def __init__(self, controls: Sequence[PendingControlPoint]): self.controls = controls def _resolve(self, result: pdpb2.PhysicalDesignOperationResult) -> None: updated = result.update_control_points.controls if len(updated) != len(self.controls): raise RuntimeError( f"runtime reported {len(updated)} control points for {len(self.controls)}" ) for pending, uc in zip(self.controls, updated, strict=True): pending._set_placement( composite_key(stable_id_key(s) for s in uc.control.ids), placement_from_pose_proto(uc.pose), )
[docs] class PendingRegion(_Pending): """An area region created by a request. Its runtime id is available once the batch has been sent; the model object (the region's feature, or for a copper pour its :class:`~jitx.copper.Pour`) once the runtime reports it (follow the request with a sync to wait for that), or after a capture.""" def __init__(self, rmap: Callable[[], ReverseMapper]): self.__rmap = rmap self.__key: CompositeKey | None = None @property def done(self) -> bool: """Whether the batch was sent and the region id is available.""" return self.__key is not None @property def key(self) -> CompositeKey: """The runtime id of the region. Raises: RuntimeError: if the batch has not been sent. """ if self.__key is None: raise RuntimeError("region is pending until the batch is sent") return self.__key @property def region(self) -> Feature | Pour | None: """The region in the current design tree, or None if the runtime has not reported it yet. A convex hull region's feature has a :class:`~jitx.shapes.hull.ConvexHull` shape holding its anchors and the evaluated polygon; a copper pour's shape is its poured copper.""" if self.__key is None: return None obj = self.__rmap().new_objects.get(self.__key) return obj if isinstance(obj, Feature | Pour) else None def _set(self, key: CompositeKey) -> None: self.__key = key
class _CreatedRegions(_Pending): def __init__(self, regions: Sequence[PendingRegion]): self.regions = regions def _resolve(self, result: pdpb2.PhysicalDesignOperationResult) -> None: created = result.create_area_regions.regions if len(created) != len(self.regions): raise RuntimeError( f"runtime reported {len(created)} regions for {len(self.regions)}" ) for pending, sid in zip(self.regions, created, strict=True): pending._set((stable_id_key(sid),)) type RegionFeature = ( KeepOut | Soldermask | Silkscreen | Paste | Glue | Finish | Custom | Cutout | Pour ) """What an area region can be: a keepout, a cutout, a surface feature, or a single-layer copper pour.""" type RegionElement = ( Pad | Port | Via | PendingVia | ControlPoint | PendingControlPoint | Component | object ) """A board element a region refers to (an origin, a hull anchor).""" type RouteEndpoint = Port | Pad | Via | PendingVia | PendingControlPoint | object """What a route can start or end at: a pad, a port mapped to a single pad, a via (or a :class:`PendingVia` from an earlier request), or a control point (or a :class:`PendingControlPoint`) or one of its endpoints.""" type DiffPairPads = ( DiffPair | tuple[Port | Pad | Via | PendingVia, Port | Pad | Via | PendingVia] ) """The two sides of a differential pair: a :class:`~jitx.net.DiffPair` port whose ``p`` and ``n`` each map to a single pad, or ``(p, n)`` as pads, vias, or ports mapped to a single pad.""" type Deletable = ( Route | Via | ControlPoint | Feature | Pour | PendingRoute | PendingVia | PendingControlPoint | PendingRegion ) """What :meth:`delete` accepts.""" class _Batch: def __init__(self, rmap: Callable[[], ReverseMapper]): self.__rmap = rmap self.__operations: list[pdpb2.PhysicalDesignOperation] = [] self.__pending: list[_Pending] = [] def __add( self, operation: pdpb2.PhysicalDesignOperation, pending: _Pending ) -> None: self.__operations.append(operation) self.__pending.append(pending) def sync(self, timeout: float | None = None) -> PendingSync: """Add a sync: wait until the runtime has no background tasks running (routing, checks, ...), after everything before it is applied. Only allowed at the start or end of a batch. At the start it is a precondition: if it times out, the batch is refused. At the end, the handle reports whether everything settled. Args: timeout: Seconds to wait at most; None for the runtime's default. Returns: A handle holding the outcome once the batch is sent. """ sync = pdpb2.SyncOperation() if timeout is not None: sync.timeout_ms = max(0, round(timeout * 1000)) pending = PendingSync() self.__add(pdpb2.PhysicalDesignOperation(sync=sync), pending) return pending @overload def place(self, target: Component, placement: Placement, /) -> PendingPlacement: ... @overload def place( self, target: Component, /, position: Point | None = None, rotation: float | None = None, *, on: Side | None = None, ) -> PendingPlacement: ... def place( self, target: Component, /, position: Placement | Point | None = None, rotation: float | None = None, *, on: Side | None = None, ) -> PendingPlacement: """Add a component placement to the batch. Args: target: The component to place. position: A board-absolute :class:`~jitx.placement.Placement`, setting position, rotation, and side together, or the board-absolute position. Omitted aspects keep their current value in the runtime. rotation: Rotation in degrees. on: The board side. Returns: A handle holding the resulting placement once the batch is sent. Raises: ValueError: if the runtime does not know the component, or nothing would change. """ return self._place(target, position, rotation, on) def _place( self, target: Component, position: Placement | Point | None, rotation: float | None, on: Side | None, ) -> PendingPlacement: key, update = _placement_update(self.__rmap(), target, position, rotation, on) # Consecutive placements share an operation. A component placed again # starts a new one, so it applies to the earlier placement's result. current = self.__pending[-1] if self.__pending else None if not isinstance(current, _Placements) or key in current.pending: current = _Placements() self.__add( pdpb2.PhysicalDesignOperation(place_components=pdpb2.PlaceComponents()), current, ) self.__operations[-1].place_components.placements.append(update) pending = PendingPlacement(target) current.pending[key] = pending return pending def route( self, start: RouteEndpoint, end: RouteEndpoint, layer: int, *, turns: Iterable[Point] | None = None, width: float | None = None, space: float | None = None, pair_space: float | None = None, envelope_overhead: float | None = None, pour: bool | None = None, ) -> PendingRoute: """Add a route between two endpoints, as if routed in the UI. Without turns the route is autorouted from ``start`` to ``end``, like a UI autoroute; with turns (an empty sequence for a straight sketch) it follows them where it can, like a UI guided route, and is autorouted otherwise. The route is realized in the background; a sync after it waits for that. A route the router can't find a path for is kept, blocked, and may route later. The net follows from the endpoints. Args: start: Where the route starts: a pad, a port mapped to a single pad, a via, or a control point or one of its endpoints. end: Where the route ends, as ``start``. layer: The copper layer, counted from the top; negative counts from the bottom. turns: Board-absolute turning points for the route to pass, in order from ``start`` to ``end``. Only the points between the endpoints: unlike a :class:`~jitx.circuit.Route` sketch, the list does not begin and end with the endpoints, which are always ``start`` and ``end``. None to autoroute. width: Trace width; None for the design's constraints. space: Trace clearance; None for the design's constraints. pair_space: Differential pair spacing; None for the design's constraints. envelope_overhead: Extra width, beyond the route's copper, of the region the trace can meander within for length matching; 0 disables length matching for the route. None for the design's constraints. pour: Pour the route: its copper is grown into a pour filling the free space around it. None for no pour. Returns: A handle holding the route id once the batch is sent. Raises: ValueError: if an endpoint is unknown to the runtime, or a port maps to several pads. """ rmap = self.__rmap() op = pdpb2.CreateRoute( start=composite_id(_endpoint_key(rmap, start)), end=composite_id(_endpoint_key(rmap, end)), layer=_layer(rmap, layer), ) if turns is not None: op.sketch.CopyFrom(_turns(turns)) config = _route_config( width, space, pair_space, envelope_overhead, (), pour=pour ) if config is not None: op.config.CopyFrom(config) pending = PendingRoute(self.__rmap) self.__add(pdpb2.PhysicalDesignOperation(create_route=op), pending) return pending def via( self, definition: Via | type[Via] | str, origin: Port | Pad, position: Point, layer: int, *, turns: Iterable[Point] | None = None, force: bool = False, width: float | None = None, space: float | None = None, ) -> PendingVia: """Add a routing via, connected to its origin by a route, as if placed in the UI. Args: definition: The via definition used in the design: a :class:`~jitx.via.Via` instance, a subclass standing for its no-argument instance, or its name. origin: The pad (or port mapped to a single pad) the via connects to. position: Board-absolute position of the via. layer: The copper layer of the route from the origin to the via. turns: Board-absolute turning points for that route to pass, in order from ``origin`` to the via; only the points between them, as for :meth:`route`. None to autoroute. force: Place the via exactly at ``position``; otherwise it is moved to the nearest free spot. width: Trace width of the route; None for the constraints. space: Trace clearance of the route; None for the constraints. Returns: A handle holding the via id, final position, and route once the batch is sent. Raises: ValueError: if the definition or origin is unknown to the runtime, or a port maps to several pads. """ rmap = self.__rmap() x, y = position entry = pdpb2.CreateVia( definition=_via_name(rmap, definition), origin=composite_id(_endpoint_key(rmap, origin)), layer=_layer(rmap, layer), position=spb2.Point(x=x, y=y), ) if turns is not None: entry.sketch.CopyFrom(_turns(turns)) op = pdpb2.CreateVias(vias=[entry], force_place=force) config = _route_config(width, space, None, None, ()) if config is not None: op.config.CopyFrom(config) pending = PendingVia(self.__rmap) self.__add( pdpb2.PhysicalDesignOperation(create_vias=op), _CreatedVias([pending]) ) return pending def update_via( self, via: Via | PendingVia, *, position: Point | None = None, definition: Via | type[Via] | str | None = None, force: bool = False, ) -> PendingVia: """Move a via, change its definition, or both, as in the UI. Routes attached to it follow. Args: via: The via, created by the runtime (not placed in code). position: New board-absolute position; None keeps it. definition: New via definition, as for :meth:`via`; None keeps it. force: Place the via exactly at ``position``; otherwise it is moved to the nearest free spot. Returns: A handle holding the via's resulting position once the batch is sent. Raises: ValueError: if the via or definition is unknown to the runtime, or nothing would change. """ if position is None and definition is None: raise ValueError("update_via needs a position or a definition") rmap = self.__rmap() key = _object_key(rmap, via) entry = pdpb2.UpdateVia(via=composite_id(key)) if position is not None: x, y = position entry.position.CopyFrom(spb2.Point(x=x, y=y)) if definition is not None: entry.definition = _via_name(rmap, definition) op = pdpb2.UpdateVias(vias=[entry], force_place=force) pending = PendingVia(self.__rmap, key) self.__add( pdpb2.PhysicalDesignOperation(update_vias=op), _UpdatedVias([pending]) ) return pending def modify( self, route: Route | PendingRoute, *, turns: Iterable[Point] | None = None, width: float | None = None, space: float | None = None, pair_space: float | None = None, envelope_overhead: float | None = None, pour: bool | None = None, reset: Collection[RouteSetting] = (), ) -> None: """Change a route's sketch or settings, as in the UI; the route is rerouted as needed. A route created in code can be modified except for what code sets on it. Args: route: The route. turns: A new sketch: board-absolute turning points for the route to pass, in order from its start to its end. Only the points between the endpoints: unlike a :class:`~jitx.circuit.Route` sketch, the list does not begin and end with the endpoints, which stay the route's own. Empty for a straight sketch; None keeps the sketch. width: New trace width; None keeps it. space: New trace clearance; None keeps it. pair_space: New differential pair spacing; None keeps it. envelope_overhead: New extra width of the length matching region (see :meth:`route`); None keeps it. pour: Whether the route is poured (see :meth:`route`); None keeps it. reset: Settings to return to the design's constraints. Raises: ValueError: if the route is unknown to the runtime, or nothing would change. """ entry = pdpb2.ModifyRoute( route=stable_id_proto(_object_key(self.__rmap(), route)[-1]) ) config = _route_config( width, space, pair_space, envelope_overhead, reset, pour=pour ) if config is not None: entry.config.CopyFrom(config) if turns is not None: entry.sketch.CopyFrom(_turns(turns)) if config is None and turns is None: raise ValueError("modify needs turns, a setting, or a reset") op = pdpb2.ModifyRoutes(routes=[entry]) self.__add(pdpb2.PhysicalDesignOperation(modify_routes=op), _Done()) def route_point( self, origin: RouteEndpoint, position: Point, layer: int, *, radius: float | None = None, size: tuple[float, float] | None = None, turns: Iterable[Point] | None = None, width: float | None = None, space: float | None = None, ) -> PendingControlPoint: """Add a route point (a single-ended control point) routed from ``origin``, as if placed in the UI. Routes can continue from it. Args: origin: What the route to the point starts at: a pad, a port mapped to a single pad, a via, or a control point. position: Board-absolute position. layer: The copper layer, counted from the top; negative counts from the bottom. radius: A circular point of this radius, which must be positive; None (and no ``size``) for an automatic shape. size: A rectangular point of this width and height, both positive. turns: Board-absolute turning points for the route from ``origin``; only the points between the endpoints, as for :meth:`route`. None to autoroute. width: Trace width of the route; None for the constraints. space: Trace clearance of the route; None for the constraints. Returns: A handle holding the control point's id and route once the batch is sent. Raises: ValueError: if ``radius`` or a side of ``size`` isn't positive. """ rmap = self.__rmap() entry = pdpb2.CreateControlPoint( type=epb2.SINGLE, origin=composite_id(_endpoint_key(rmap, origin)), layer=_layer(rmap, layer), pose=_pose(position, 0), ) if turns is not None: entry.sketch.CopyFrom(_turns(turns)) shape = _control_shape(radius, size) if shape is not None: entry.shape.CopyFrom(shape) return self.__create_control(entry, _route_config(width, space, None, None, ())) def pair_insertion( self, pair: DiffPairPads, position: Point, rotation: float, layer: int, *, invert: bool = False, width: float | None = None, space: float | None = None, pair_space: float | None = None, ) -> PendingControlPoint: """Add a differential pair insertion point, where the p and n traces from their pads merge into a coupled pair, as if placed in the UI. It is routed from both pads; a pair route continues from it. Use this when where the point should go is known; see :meth:`drop_pair_insertion` to have it placed next to the pads. Args: pair: The p and n pads (or vias) the point is routed from. position: Board-absolute position of the point. rotation: Rotation in degrees; at 0 the coupled side points along +x. layer: The copper layer, counted from the top; negative counts from the bottom. invert: Swap the p and n sides, as for a :class:`~jitx.controlpoint.PairInsertion`, when the traces would otherwise cross. width: Trace width of the routes; None for the constraints. space: Trace clearance of the routes; None for the constraints. pair_space: Differential pair spacing; None for the constraints. Returns: A handle holding the control point's id and routes once the batch is sent. Raises: ValueError: if a pad is unknown to the runtime, or a port maps to several pads. """ rmap = self.__rmap() p, n = _pair_keys(rmap, pair) if invert: p, n = n, p entry = pdpb2.CreateControlPoint( type=epb2.MERGE, origin=composite_id(p), dual_origin=composite_id(n), layer=_layer(rmap, layer), pose=_pose(position, rotation), ) return self.__create_control( entry, _route_config(width, space, pair_space, None, ()) ) def pair_point( self, origin: ControlPoint | PendingControlPoint, position: Point, rotation: float, layer: int, *, turns: Iterable[Point] | None = None, width: float | None = None, space: float | None = None, pair_space: float | None = None, ) -> PendingControlPoint: """Add a pair point continuing a coupled pair route from ``origin``, a pair insertion or another pair point, as if placed in the UI. Args: origin: The pair insertion or pair point the pair route comes from. position: Board-absolute position. rotation: Rotation in degrees. layer: The copper layer, counted from the top; negative counts from the bottom. turns: Board-absolute turning points for the pair route; only the points between the endpoints, as for :meth:`route`. None to autoroute. width: Trace width; None for the constraints. space: Trace clearance; None for the constraints. pair_space: Differential pair spacing; None for the constraints. Returns: A handle holding the control point's id and route once the batch is sent. """ rmap = self.__rmap() entry = pdpb2.CreateControlPoint( type=epb2.PAIR, origin=composite_id(_object_key(rmap, origin)), layer=_layer(rmap, layer), pose=_pose(position, rotation), ) if turns is not None: entry.sketch.CopyFrom(_turns(turns)) return self.__create_control( entry, _route_config(width, space, pair_space, None, ()) ) def __create_control( self, entry: pdpb2.CreateControlPoint, config: pdpb2.RouteConfig | None ) -> PendingControlPoint: op = pdpb2.CreateControlPoints(controls=[entry]) if config is not None: op.config.CopyFrom(config) pending = PendingControlPoint(self.__rmap) self.__add( pdpb2.PhysicalDesignOperation(create_control_points=op), _CreatedControls([pending], "create_control_points"), ) return pending def drop_pair_insertion( self, pair: DiffPairPads, layer: int, *, pull: bool | None = None, width: float | None = None, space: float | None = None, pair_space: float | None = None, ) -> PendingControlPoint: """Add a differential pair insertion point placed automatically next to the pads, as the UI's drop does: between them, pointing away from the component, at the nearest spot clear of other copper. The sides are assigned so the traces don't cross. For when no precise placement is at hand; :meth:`pair_insertion` does the same job when the location is known. Args: pair: The p and n pads (or vias). layer: The copper layer, counted from the top; negative counts from the bottom. pull: True to point towards the component; None to try away from it first, then towards it. width: Trace width of the routes; None for the constraints. space: Trace clearance of the routes; None for the constraints. pair_space: Differential pair spacing; None for the constraints. Returns: A handle holding the control point's id and routes once the batch is sent. The request is refused if there is no room. """ rmap = self.__rmap() p, n = _pair_keys(rmap, pair) drop = pdpb2.DropPair(p=composite_id(p), n=composite_id(n)) if pull is not None: drop.pull = pull op = pdpb2.DropPairInsertions(layer=_layer(rmap, layer), pairs=[drop]) config = _route_config(width, space, pair_space, None, ()) if config is not None: op.config.CopyFrom(config) pending = PendingControlPoint(self.__rmap) self.__add( pdpb2.PhysicalDesignOperation(drop_pair_insertions=op), _CreatedControls([pending], "drop_pair_insertions"), ) return pending def update_control( self, control: ControlPoint | PendingControlPoint, *, position: Point | None = None, rotation: float | None = None, invert: bool = False, radius: float | None = None, size: tuple[float, float] | None = None, ) -> PendingControlPoint: """Move a control point, change its shape, or invert a pair insertion (swap its p and n sides), as in the UI. Routes attached to it follow. Args: control: The control point, created by the runtime (not placed in code). position: New board-absolute position; None keeps it. rotation: New rotation in degrees; None keeps it. invert: Swap the sides of a pair insertion. radius: A new circular shape of this radius, which must be positive. size: A new rectangular shape of this width and height, both positive. Returns: A handle holding the resulting placement once the batch is sent. Raises: ValueError: if the control point is unknown to the runtime, nothing would change, or ``radius`` or a side of ``size`` isn't positive. """ rmap = self.__rmap() key = _object_key(rmap, control) entry = pdpb2.UpdateControlPoint(control=composite_id(key), invert=invert) if position is not None or rotation is not None: current = _current_placement(control) if position is None: if current is None: raise ValueError( "the control point's position is not known; pass one" ) position = current._translate if rotation is None: rotation = current._rotate if current is not None else 0 entry.pose.CopyFrom(_pose(position, rotation)) shape = _control_shape(radius, size) if shape is not None: entry.shape.CopyFrom(shape) if not (entry.HasField("pose") or shape is not None or invert): raise ValueError( "update_control needs a position, rotation, shape, or invert" ) op = pdpb2.UpdateControlPoints(controls=[entry]) pending = PendingControlPoint(self.__rmap, key) self.__add( pdpb2.PhysicalDesignOperation(update_control_points=op), _UpdatedControls([pending]), ) return pending def region( self, feature: RegionFeature, /, *, origin: RegionElement | None = None, net: Net | Pad | Port | Via | PendingVia | None = None, ) -> PendingRegion: """Add an area region, as if drawn in the UI: a keepout, a cutout, a surface feature (soldermask, silkscreen, paste, glue, finish, custom), or a copper pour. The feature describes the region; its shape is a board-absolute shape or a :class:`~jitx.shapes.hull.ConvexHull` of board elements, which the runtime keeps around them as they move. Layers and side are board absolute. >>> rd.layout.region(Pour(ConvexHull([u1, c3.p1], expand=0.5), 0, rank=5), ... net=gnd) >>> rd.layout.region(KeepOut(rect, LayerSet(0, 1), route=True), origin=u1) Args: feature: The region. A :class:`~jitx.copper.Pour` is single layer and takes no isolation. origin: What the region moves (and flips) with: a component, or a pad, via, or control point of one; None to fix it to the board. net: The copper pour's net: a net (a pad on it is used), or a pad, port, or via whose net it is. Required for a pour, not accepted for other kinds. Returns: A handle holding the region id once the batch is sent. Raises: ValueError: if the feature can't be a region, or an element is unknown to the runtime. """ rmap = self.__rmap() entry = _region_entry(rmap, feature) if isinstance(feature, Structurable): # The feature only describes the region; it isn't added to the # design, and dropping it during instantiation would warn. dispose(feature) if isinstance(feature, Pour): if net is None: raise ValueError("a copper pour region needs a net") entry.connect.CopyFrom(composite_id(_net_key(rmap, net))) elif net is not None: raise ValueError("only a copper pour region takes a net") if origin is not None: entry.origin.CopyFrom(composite_id(_origin_key(rmap, origin))) pending = PendingRegion(self.__rmap) self.__add( pdpb2.PhysicalDesignOperation( create_area_regions=pdpb2.CreateAreaRegions(regions=[entry]) ), _CreatedRegions([pending]), ) return pending def update_region( self, region: Feature | Pour | PendingRegion, /, *, shape: Shape | None = None, layers: LayerSet | int | None = None, side: Side | None = None, rank: int | None = None, pour: bool | None = None, via: bool | None = None, route: bool | None = None, name: str | None = None, origin: RegionElement | None = None, net: Net | Pad | Port | Via | PendingVia | None = None, ) -> None: """Change an area region, as in the UI. Its kind can't change: delete it and create another instead. Args: region: The region, created by the runtime (by a request or the UI); not one made by a routing structure. shape: New board-absolute shape, or a :class:`~jitx.shapes.hull.ConvexHull`. layers: New copper layers (copper pour: a single layer). side: New board side (surface features, custom). rank: New copper pour priority. pour: Whether a keepout keeps out pours. via: Whether a keepout keeps out vias. route: Whether a keepout keeps out routes. name: New custom layer name. origin: New origin, as for :meth:`region`. net: New net of a copper pour, as for :meth:`region`. Raises: ValueError: if the region or an element is unknown to the runtime, or nothing would change. """ rmap = self.__rmap() entry = pdpb2.UpdateAreaRegion(region=_region_id(rmap, region)) if shape is not None: entry.shape.CopyFrom(_region_shape(rmap, shape)) if layers is not None: entry.layers.CopyFrom( pdpb2.AreaRegionLayers( layers=_region_layers( rmap, LayerSet(layers) if isinstance(layers, int) else layers ) ) ) if side is not None: entry.side = epb2.Side.BOTTOM if side == Side.Bottom else epb2.Side.TOP if rank is not None: entry.rank = rank if pour is not None or via is not None or route is not None: current = region.region if isinstance(region, PendingRegion) else region flags = [pour, via, route] if None in flags: if not isinstance(current, KeepOut): raise ValueError( "the keepout's current flags are not known; pass pour, " "via, and route" ) known = [current.pour, current.via, current.route] flags = [ k if f is None else f for f, k in zip(flags, known, strict=True) ] entry.keepout.CopyFrom( epb2.AreaKeepout( pour=bool(flags[0]), via=bool(flags[1]), route=bool(flags[2]) ) ) if name is not None: entry.name = name if origin is not None: entry.origin.CopyFrom(composite_id(_origin_key(rmap, origin))) if net is not None: entry.connect.CopyFrom(composite_id(_net_key(rmap, net))) if len(entry.ListFields()) == 1: raise ValueError("update_region needs something to change") self.__add( pdpb2.PhysicalDesignOperation( update_area_regions=pdpb2.UpdateAreaRegions(regions=[entry]) ), _Done(), ) def delete(self, *objects: Deletable) -> None: """Delete routes, vias, control points, and area regions created by the runtime, as in the UI. Routes attached to a deleted via or control point are removed or joined. Args: objects: What to delete. Raises: ValueError: if an object is unknown to the runtime. """ rmap = self.__rmap() routes: list[CompositeKey] = [] vias: list[CompositeKey] = [] controls: list[CompositeKey] = [] regions: list[sipb2.StableID] = [] for obj in objects: if isinstance(obj, Feature | Pour | PendingRegion): regions.append(_region_id(rmap, obj)) continue key = _object_key(rmap, obj) if isinstance(obj, Route | PendingRoute): routes.append(key) elif isinstance(obj, ControlPoint | PendingControlPoint): controls.append(key) else: vias.append(key) if routes: op = pdpb2.DeleteRoutes(routes=[stable_id_proto(k[-1]) for k in routes]) self.__add(pdpb2.PhysicalDesignOperation(delete_routes=op), _Done()) if vias: op = pdpb2.DeleteVias(vias=[composite_id(k) for k in vias]) self.__add(pdpb2.PhysicalDesignOperation(delete_vias=op), _Done()) if controls: op = pdpb2.DeleteControlPoints(controls=[composite_id(k) for k in controls]) self.__add(pdpb2.PhysicalDesignOperation(delete_control_points=op), _Done()) if regions: op = pdpb2.DeleteAreaRegions(regions=regions) self.__add(pdpb2.PhysicalDesignOperation(delete_area_regions=op), _Done()) def _request(self) -> pdpb2.PhysicalDesignOperations | None: if not self.__operations: return None return pdpb2.PhysicalDesignOperations(operations=self.__operations) def _resolve(self, response: pdpb2.PhysicalDesignOperationsResponse) -> None: results = response.results if len(results) != len(self.__pending): raise RuntimeError( f"runtime reported {len(results)} results for " f"{len(self.__pending)} operations" ) for result, pending in zip(results, self.__pending, strict=True): pending._resolve(result)
[docs] class AsyncBatch(_Batch): """Requests sent together as one request when the ``async with`` block exits without an exception. The runtime applies the batch in order and all-or-nothing, as one undoable step. Placing a component again in the same batch applies to its earlier placement in the batch. >>> async with design.layout.batch() as b: ... p = b.place(design.root.u1, (10, 5), 90) ... b.place(design.root.u2, on=Side.Bottom) >>> p.placement Raises: RequestRejected: from the block exit, if the runtime refused the batch. """ def __init__( self, rmap: Callable[[], ReverseMapper], send: Callable[ [pdpb2.PhysicalDesignOperations], Awaitable[pdpb2.PhysicalDesignOperationsResponse], ], ): super().__init__(rmap) self.__send = send async def __aenter__(self) -> AsyncBatch: return self async def __aexit__( self, exc_type: type[BaseException] | None, exc: BaseException | None, tb: TracebackType | None, ) -> None: if exc_type is not None: return request = self._request() if request is not None: self._resolve(await self.__send(request))
[docs] class SyncBatch(_Batch): """Requests sent together as one request when the ``with`` block exits without an exception. The runtime applies the batch in order and all-or-nothing, as one undoable step. Placing a component again in the same batch applies to its earlier placement in the batch. >>> with design.layout.batch() as b: ... p = b.place(design.root.u1, (10, 5), 90) ... b.place(design.root.u2, on=Side.Bottom) >>> p.placement Raises: RequestRejected: from the block exit, if the runtime refused the batch. """ def __init__( self, rmap: Callable[[], ReverseMapper], send: Callable[ [pdpb2.PhysicalDesignOperations], pdpb2.PhysicalDesignOperationsResponse ], ): super().__init__(rmap) self.__send = send def __enter__(self) -> SyncBatch: return self def __exit__( self, exc_type: type[BaseException] | None, exc: BaseException | None, tb: TracebackType | None, ) -> None: if exc_type is not None: return request = self._request() if request is not None: self._resolve(self.__send(request))
[docs] class AsyncLayout: """The layout of an async runtime design. Each request method is one request; see :meth:`batch` to send several as one.""" def __init__( self, rmap: Callable[[], ReverseMapper], send: Callable[ [pdpb2.PhysicalDesignOperations], Awaitable[pdpb2.PhysicalDesignOperationsResponse], ], *, monitor: LayoutMonitor | None = None, publish: Callable[[pdpb2.PhysicalDesignEvent], Awaitable[None]] | None = None, ): self.__rmap = rmap self.__send = send self.__monitor = monitor or LayoutMonitor(_Monitors()) self.__publish = publish @property def monitor(self) -> LayoutMonitor: """Callbacks on changes to the layout. See :class:`LayoutMonitor`.""" return self.__monitor
[docs] async def emit(self, event: UserEvent) -> None: """Send a user event to the other clients monitoring this design. Raises: TypeError: if the event's data is not JSON-compatible. ValueError: if an object has no runtime ID. """ if self.__publish is None: raise RuntimeError("this layout is not attached to a runtime") user = user_event_proto(event, self.__rmap()) await self.__publish(pdpb2.PhysicalDesignEvent(user=user))
[docs] def batch(self) -> AsyncBatch: """Start a batch of requests, sent as one request. See :class:`AsyncBatch`.""" return AsyncBatch(self.__rmap, self.__send)
[docs] async def sync(self, timeout: float | None = None) -> bool: """Wait until the runtime has no background tasks running (routing, checks, ...), so the design is settled. Args: timeout: Seconds to wait at most; None for the runtime's default. Returns: True if everything settled, False if the wait timed out. """ async with self.batch() as b: pending = b.sync(timeout) return pending.settled
@overload async def place(self, target: Component, placement: Placement, /) -> Placement: ... @overload async def place( self, target: Component, /, position: Point | None = None, rotation: float | None = None, *, on: Side | None = None, ) -> Placement: ...
[docs] async def place( self, target: Component, /, position: Placement | Point | None = None, rotation: float | None = None, *, on: Side | None = None, ) -> Placement: """Place a component on the board, as if dragged in the UI. Args: target: The component to place. position: A board-absolute :class:`~jitx.placement.Placement`, setting position, rotation, and side together, or the board-absolute position. Omitted aspects keep their current value in the runtime. rotation: Rotation in degrees. on: The board side. Returns: The resulting board-absolute placement. Raises: RequestRejected: if the runtime refused the request. """ async with self.batch() as b: pending = b._place(target, position, rotation, on) return pending.placement
[docs] async def route( self, start: RouteEndpoint, end: RouteEndpoint, layer: int, *, turns: Iterable[Point] | None = None, width: float | None = None, space: float | None = None, pair_space: float | None = None, envelope_overhead: float | None = None, pour: bool | None = None, wait: bool = True, ) -> PendingRoute: """Route between two endpoints. See :meth:`AsyncBatch.route`. Args: wait: Wait for background work (the routing) to finish, so the route is routed or blocked when this returns. Raises: RequestRejected: if the runtime refused the request. """ async with self.batch() as b: pending = b.route( start, end, layer, turns=turns, width=width, space=space, pair_space=pair_space, envelope_overhead=envelope_overhead, pour=pour, ) if wait: b.sync() return pending
[docs] async def via( self, definition: Via | type[Via] | str, origin: Port | Pad, position: Point, layer: int, *, turns: Iterable[Point] | None = None, force: bool = False, width: float | None = None, space: float | None = None, wait: bool = True, ) -> PendingVia: """Place a routing via. See :meth:`AsyncBatch.via`. Args: wait: Wait for background work (routing to the via) to finish. Raises: RequestRejected: if the runtime refused the request. """ async with self.batch() as b: pending = b.via( definition, origin, position, layer, turns=turns, force=force, width=width, space=space, ) if wait: b.sync() return pending
[docs] async def update_via( self, via: Via | PendingVia, *, position: Point | None = None, definition: Via | type[Via] | str | None = None, force: bool = False, wait: bool = True, ) -> PendingVia: """Move a via or change its definition. See :meth:`AsyncBatch.update_via`. Args: wait: Wait for background work (rerouting) to finish. Raises: RequestRejected: if the runtime refused the request. """ async with self.batch() as b: pending = b.update_via( via, position=position, definition=definition, force=force ) if wait: b.sync() return pending
[docs] async def modify( self, route: Route | PendingRoute, *, turns: Iterable[Point] | None = None, width: float | None = None, space: float | None = None, pair_space: float | None = None, envelope_overhead: float | None = None, pour: bool | None = None, reset: Collection[RouteSetting] = (), wait: bool = True, ) -> None: """Change a route's sketch or settings. See :meth:`AsyncBatch.modify`. Args: wait: Wait for background work (rerouting) to finish. Raises: RequestRejected: if the runtime refused the request. """ async with self.batch() as b: b.modify( route, turns=turns, width=width, space=space, pair_space=pair_space, envelope_overhead=envelope_overhead, pour=pour, reset=reset, ) if wait: b.sync()
[docs] async def route_point( self, origin: RouteEndpoint, position: Point, layer: int, *, radius: float | None = None, size: tuple[float, float] | None = None, turns: Iterable[Point] | None = None, width: float | None = None, space: float | None = None, wait: bool = True, ) -> PendingControlPoint: """Add a route point. See :meth:`AsyncBatch.route_point`. Args: wait: Wait for background work (routing to the point) to finish. Raises: RequestRejected: if the runtime refused the request. """ async with self.batch() as b: pending = b.route_point( origin, position, layer, radius=radius, size=size, turns=turns, width=width, space=space, ) if wait: b.sync() return pending
[docs] async def pair_insertion( self, pair: DiffPairPads, position: Point, rotation: float, layer: int, *, invert: bool = False, width: float | None = None, space: float | None = None, pair_space: float | None = None, wait: bool = True, ) -> PendingControlPoint: """Add a differential pair insertion point. See :meth:`AsyncBatch.pair_insertion`. Args: wait: Wait for background work (routing to the point) to finish. Raises: RequestRejected: if the runtime refused the request. """ async with self.batch() as b: pending = b.pair_insertion( pair, position, rotation, layer, invert=invert, width=width, space=space, pair_space=pair_space, ) if wait: b.sync() return pending
[docs] async def pair_point( self, origin: ControlPoint | PendingControlPoint, position: Point, rotation: float, layer: int, *, turns: Iterable[Point] | None = None, width: float | None = None, space: float | None = None, pair_space: float | None = None, wait: bool = True, ) -> PendingControlPoint: """Add a pair point. See :meth:`AsyncBatch.pair_point`. Args: wait: Wait for background work (routing to the point) to finish. Raises: RequestRejected: if the runtime refused the request. """ async with self.batch() as b: pending = b.pair_point( origin, position, rotation, layer, turns=turns, width=width, space=space, pair_space=pair_space, ) if wait: b.sync() return pending
[docs] async def drop_pair_insertion( self, pair: DiffPairPads, layer: int, *, pull: bool | None = None, width: float | None = None, space: float | None = None, pair_space: float | None = None, wait: bool = True, ) -> PendingControlPoint: """Add a pair insertion point placed next to its pads. See :meth:`AsyncBatch.drop_pair_insertion`. Args: wait: Wait for background work (routing to the point) to finish. Raises: RequestRejected: if the runtime refused the request. """ async with self.batch() as b: pending = b.drop_pair_insertion( pair, layer, pull=pull, width=width, space=space, pair_space=pair_space ) if wait: b.sync() return pending
[docs] async def update_control( self, control: ControlPoint | PendingControlPoint, *, position: Point | None = None, rotation: float | None = None, invert: bool = False, radius: float | None = None, size: tuple[float, float] | None = None, wait: bool = True, ) -> PendingControlPoint: """Move, reshape, or invert a control point. See :meth:`AsyncBatch.update_control`. Args: wait: Wait for background work (rerouting) to finish. Raises: RequestRejected: if the runtime refused the request. """ async with self.batch() as b: pending = b.update_control( control, position=position, rotation=rotation, invert=invert, radius=radius, size=size, ) if wait: b.sync() return pending
[docs] async def region( self, feature: RegionFeature, /, *, origin: RegionElement | None = None, net: Net | Pad | Port | Via | PendingVia | None = None, wait: bool = True, ) -> PendingRegion: """Add an area region. See :meth:`AsyncBatch.region`. Args: wait: Wait for background work (evaluating, pouring) to finish, so the region is in the model when this returns. Raises: RequestRejected: if the runtime refused the request. """ async with self.batch() as b: pending = b.region(feature, origin=origin, net=net) if wait: b.sync() return pending
[docs] async def update_region( self, region: Feature | Pour | PendingRegion, /, *, shape: Shape | None = None, layers: LayerSet | int | None = None, side: Side | None = None, rank: int | None = None, pour: bool | None = None, via: bool | None = None, route: bool | None = None, name: str | None = None, origin: RegionElement | None = None, net: Net | Pad | Port | Via | PendingVia | None = None, wait: bool = True, ) -> None: """Change an area region. See :meth:`AsyncBatch.update_region`. Args: wait: Wait for background work (evaluating, pouring) to finish. Raises: RequestRejected: if the runtime refused the request. """ async with self.batch() as b: b.update_region( region, shape=shape, layers=layers, side=side, rank=rank, pour=pour, via=via, route=route, name=name, origin=origin, net=net, ) if wait: b.sync()
[docs] async def delete(self, *objects: Deletable) -> None: """Delete routes, vias, control points, and area regions. See :meth:`AsyncBatch.delete`. Raises: RequestRejected: if the runtime refused the request. """ async with self.batch() as b: b.delete(*objects)
[docs] class SyncLayout: """The layout of a sync runtime design. Each request method is one request; see :meth:`batch` to send several as one.""" def __init__( self, rmap: Callable[[], ReverseMapper], send: Callable[ [pdpb2.PhysicalDesignOperations], pdpb2.PhysicalDesignOperationsResponse ], *, monitor: LayoutMonitor | None = None, publish: Callable[[pdpb2.PhysicalDesignEvent], None] | None = None, ): self.__rmap = rmap self.__send = send self.__monitor = monitor or LayoutMonitor(_Monitors()) self.__publish = publish @property def monitor(self) -> LayoutMonitor: """Callbacks on changes to the layout. See :class:`LayoutMonitor`.""" return self.__monitor
[docs] def emit(self, event: UserEvent) -> None: """Send a user event to the other clients monitoring this design. Raises: TypeError: if the event's data is not JSON-compatible. ValueError: if an object has no runtime ID. """ if self.__publish is None: raise RuntimeError("this layout is not attached to a runtime") user = user_event_proto(event, self.__rmap()) self.__publish(pdpb2.PhysicalDesignEvent(user=user))
[docs] def batch(self) -> SyncBatch: """Start a batch of requests, sent as one request. See :class:`SyncBatch`.""" return SyncBatch(self.__rmap, self.__send)
[docs] def sync(self, timeout: float | None = None) -> bool: """Wait until the runtime has no background tasks running (routing, checks, ...), so the design is settled. Args: timeout: Seconds to wait at most; None for the runtime's default. Returns: True if everything settled, False if the wait timed out. """ with self.batch() as b: pending = b.sync(timeout) return pending.settled
@overload def place(self, target: Component, placement: Placement, /) -> Placement: ... @overload def place( self, target: Component, /, position: Point | None = None, rotation: float | None = None, *, on: Side | None = None, ) -> Placement: ...
[docs] def place( self, target: Component, /, position: Placement | Point | None = None, rotation: float | None = None, *, on: Side | None = None, ) -> Placement: """Place a component on the board, as if dragged in the UI. Args: target: The component to place. position: A board-absolute :class:`~jitx.placement.Placement`, setting position, rotation, and side together, or the board-absolute position. Omitted aspects keep their current value in the runtime. rotation: Rotation in degrees. on: The board side. Returns: The resulting board-absolute placement. Raises: RequestRejected: if the runtime refused the request. """ with self.batch() as b: pending = b._place(target, position, rotation, on) return pending.placement
[docs] def route( self, start: RouteEndpoint, end: RouteEndpoint, layer: int, *, turns: Iterable[Point] | None = None, width: float | None = None, space: float | None = None, pair_space: float | None = None, envelope_overhead: float | None = None, pour: bool | None = None, wait: bool = True, ) -> PendingRoute: """Route between two endpoints. See :meth:`SyncBatch.route`. Args: wait: Wait for background work (the routing) to finish, so the route is routed or blocked when this returns. Raises: RequestRejected: if the runtime refused the request. """ with self.batch() as b: pending = b.route( start, end, layer, turns=turns, width=width, space=space, pair_space=pair_space, envelope_overhead=envelope_overhead, pour=pour, ) if wait: b.sync() return pending
[docs] def via( self, definition: Via | type[Via] | str, origin: Port | Pad, position: Point, layer: int, *, turns: Iterable[Point] | None = None, force: bool = False, width: float | None = None, space: float | None = None, wait: bool = True, ) -> PendingVia: """Place a routing via. See :meth:`SyncBatch.via`. Args: wait: Wait for background work (routing to the via) to finish. Raises: RequestRejected: if the runtime refused the request. """ with self.batch() as b: pending = b.via( definition, origin, position, layer, turns=turns, force=force, width=width, space=space, ) if wait: b.sync() return pending
[docs] def update_via( self, via: Via | PendingVia, *, position: Point | None = None, definition: Via | type[Via] | str | None = None, force: bool = False, wait: bool = True, ) -> PendingVia: """Move a via or change its definition. See :meth:`SyncBatch.update_via`. Args: wait: Wait for background work (rerouting) to finish. Raises: RequestRejected: if the runtime refused the request. """ with self.batch() as b: pending = b.update_via( via, position=position, definition=definition, force=force ) if wait: b.sync() return pending
[docs] def modify( self, route: Route | PendingRoute, *, turns: Iterable[Point] | None = None, width: float | None = None, space: float | None = None, pair_space: float | None = None, envelope_overhead: float | None = None, pour: bool | None = None, reset: Collection[RouteSetting] = (), wait: bool = True, ) -> None: """Change a route's sketch or settings. See :meth:`SyncBatch.modify`. Args: wait: Wait for background work (rerouting) to finish. Raises: RequestRejected: if the runtime refused the request. """ with self.batch() as b: b.modify( route, turns=turns, width=width, space=space, pair_space=pair_space, envelope_overhead=envelope_overhead, pour=pour, reset=reset, ) if wait: b.sync()
[docs] def route_point( self, origin: RouteEndpoint, position: Point, layer: int, *, radius: float | None = None, size: tuple[float, float] | None = None, turns: Iterable[Point] | None = None, width: float | None = None, space: float | None = None, wait: bool = True, ) -> PendingControlPoint: """Add a route point. See :meth:`SyncBatch.route_point`. Args: wait: Wait for background work (routing to the point) to finish. Raises: RequestRejected: if the runtime refused the request. """ with self.batch() as b: pending = b.route_point( origin, position, layer, radius=radius, size=size, turns=turns, width=width, space=space, ) if wait: b.sync() return pending
[docs] def pair_insertion( self, pair: DiffPairPads, position: Point, rotation: float, layer: int, *, invert: bool = False, width: float | None = None, space: float | None = None, pair_space: float | None = None, wait: bool = True, ) -> PendingControlPoint: """Add a differential pair insertion point. See :meth:`SyncBatch.pair_insertion`. Args: wait: Wait for background work (routing to the point) to finish. Raises: RequestRejected: if the runtime refused the request. """ with self.batch() as b: pending = b.pair_insertion( pair, position, rotation, layer, invert=invert, width=width, space=space, pair_space=pair_space, ) if wait: b.sync() return pending
[docs] def pair_point( self, origin: ControlPoint | PendingControlPoint, position: Point, rotation: float, layer: int, *, turns: Iterable[Point] | None = None, width: float | None = None, space: float | None = None, pair_space: float | None = None, wait: bool = True, ) -> PendingControlPoint: """Add a pair point. See :meth:`SyncBatch.pair_point`. Args: wait: Wait for background work (routing to the point) to finish. Raises: RequestRejected: if the runtime refused the request. """ with self.batch() as b: pending = b.pair_point( origin, position, rotation, layer, turns=turns, width=width, space=space, pair_space=pair_space, ) if wait: b.sync() return pending
[docs] def drop_pair_insertion( self, pair: DiffPairPads, layer: int, *, pull: bool | None = None, width: float | None = None, space: float | None = None, pair_space: float | None = None, wait: bool = True, ) -> PendingControlPoint: """Add a pair insertion point placed next to its pads. See :meth:`SyncBatch.drop_pair_insertion`. Args: wait: Wait for background work (routing to the point) to finish. Raises: RequestRejected: if the runtime refused the request. """ with self.batch() as b: pending = b.drop_pair_insertion( pair, layer, pull=pull, width=width, space=space, pair_space=pair_space ) if wait: b.sync() return pending
[docs] def update_control( self, control: ControlPoint | PendingControlPoint, *, position: Point | None = None, rotation: float | None = None, invert: bool = False, radius: float | None = None, size: tuple[float, float] | None = None, wait: bool = True, ) -> PendingControlPoint: """Move, reshape, or invert a control point. See :meth:`SyncBatch.update_control`. Args: wait: Wait for background work (rerouting) to finish. Raises: RequestRejected: if the runtime refused the request. """ with self.batch() as b: pending = b.update_control( control, position=position, rotation=rotation, invert=invert, radius=radius, size=size, ) if wait: b.sync() return pending
[docs] def region( self, feature: RegionFeature, /, *, origin: RegionElement | None = None, net: Net | Pad | Port | Via | PendingVia | None = None, wait: bool = True, ) -> PendingRegion: """Add an area region. See :meth:`SyncBatch.region`. Args: wait: Wait for background work (evaluating, pouring) to finish, so the region is in the model when this returns. Raises: RequestRejected: if the runtime refused the request. """ with self.batch() as b: pending = b.region(feature, origin=origin, net=net) if wait: b.sync() return pending
[docs] def update_region( self, region: Feature | Pour | PendingRegion, /, *, shape: Shape | None = None, layers: LayerSet | int | None = None, side: Side | None = None, rank: int | None = None, pour: bool | None = None, via: bool | None = None, route: bool | None = None, name: str | None = None, origin: RegionElement | None = None, net: Net | Pad | Port | Via | PendingVia | None = None, wait: bool = True, ) -> None: """Change an area region. See :meth:`SyncBatch.update_region`. Args: wait: Wait for background work (evaluating, pouring) to finish. Raises: RequestRejected: if the runtime refused the request. """ with self.batch() as b: b.update_region( region, shape=shape, layers=layers, side=side, rank=rank, pour=pour, via=via, route=route, name=name, origin=origin, net=net, ) if wait: b.sync()
[docs] def delete(self, *objects: Deletable) -> None: """Delete routes, vias, control points, and area regions. See :meth:`SyncBatch.delete`. Raises: RequestRejected: if the runtime refused the request. """ with self.batch() as b: b.delete(*objects)
def _placement_update( rmap: ReverseMapper, target: Component, position: Placement | Point | None, rotation: float | None, on: Side | None, ) -> tuple[CompositeKey, pdpb2.PlaceComponent]: if not isinstance(target, Component): raise TypeError(f"only components can be placed, got {type(target).__name__}") if isinstance(position, Placement): if rotation is not None or on is not None: raise TypeError("pass either a Placement, or position, rotation, and on") # The flip is carried by the side, so the scale is not sent. position, rotation, on = position._translate, position._rotate, position.side if position is None and rotation is None and on is None: raise ValueError("place needs a placement, position, rotation, or side") key = rmap.stable_id_of(target) if key is None: raise ValueError(f"the runtime has no ID for component {target}") update = pdpb2.PlaceComponent(component=composite_id(key)) if position is not None: x, y = position update.center.CopyFrom(spb2.Point(x=x, y=y)) if rotation is not None: update.angle = rotation if on is not None: update.side = epb2.Side.BOTTOM if on == Side.Bottom else epb2.Side.TOP return key, update def _object_key(rmap: ReverseMapper, obj: Deletable) -> CompositeKey: if isinstance(obj, PendingRoute | PendingVia | PendingControlPoint): return obj.key key = rmap.stable_id_of(obj) if key is None: raise ValueError(f"the runtime has no ID for {obj}") return key def _endpoint_key(rmap: ReverseMapper, endpoint: RouteEndpoint) -> CompositeKey: if isinstance(endpoint, PendingVia | PendingControlPoint): return endpoint.key if isinstance(endpoint, Port): endpoint = _port_pad(rmap, endpoint) key = rmap.stable_id_of(endpoint) if key is None: raise ValueError(f"the runtime has no ID for route endpoint {endpoint}") return key def _owning_component(ob: object) -> Component | None: # Walks up a capture proxy's parents (port, bundle, ...) to its component. while Proxy.is_proxy(ob): ob = Proxy.parent(cast(Proxied, ob)) if isinstance(ob, Component): return ob return None def _port_pad(rmap: ReverseMapper, port: Port) -> Pad: # Routes connect pads; a port names one only if it maps to a single pad. # Only the port's component is searched; the whole design if the port # isn't reached through the captured tree. port_key = rmap.stable_id_of(port) owner = _owning_component(port) if owner is not None: mappings = extract(owner, PadMapping, through=()) else: mappings = extract(rmap.root, PadMapping) for mapping in mappings: for mapped, pads in mapping.items(): if mapped is port or ( port_key is not None and rmap.stable_id_of(mapped) == port_key ): if isinstance(pads, Pad): return pads if len(pads) == 1: return pads[0] raise ValueError( f"port {port} maps to {len(pads)} pads; route to one of the pads" ) raise ValueError(f"port {port} is not mapped to a pad; route to a pad") def _via_name(rmap: ReverseMapper, definition: Via | type[Via] | str) -> str: if isinstance(definition, str): if definition not in rmap.via_names.values(): raise ValueError(f"no via definition named {definition!r}") return definition if isinstance(definition, type): # A class means its no-argument instance, whose origin is the one # the design memoized. try: with instantiation.require(): definition = definition() except TypeError as e: raise ValueError( f"could not instantiate {definition.__name__} without arguments;" " pass an instance" ) from e canonical = origin(definition) for seq, name in rmap.via_names.items(): if rmap.obj_by_seq.get(seq) is canonical: return name raise ValueError( f"via definition {typeof(definition).__name__} is not used in the design" ) def _layer(rmap: ReverseMapper, layer: int) -> int: if layer < 0: layer += len(rmap.root.substrate.stackup.conductors) return layer def _route_config( width: float | None, space: float | None, pair_space: float | None, envelope_overhead: float | None, reset: Collection[RouteSetting], *, pour: bool | None = None, ) -> pdpb2.RouteConfig | None: for r in reset: if not isinstance(r, RouteSetting): raise TypeError( f"reset takes RouteSetting members, got {typeof(r).__name__}" ) config = pdpb2.RouteConfig(reset=[_RESET_FIELDS[r] for r in reset]) settings = rcpb2.RouteConfiguration() if width is not None: settings.trace_width = width if space is not None: settings.trace_space = space if pair_space is not None: settings.pair_space = pair_space if envelope_overhead is not None: settings.squiggle_room = envelope_overhead if pour is not None: settings.pour_enabled = pour if settings.ListFields(): config.settings.CopyFrom(settings) if not config.ListFields(): return None return config def _pair_keys( rmap: ReverseMapper, pair: DiffPairPads ) -> tuple[CompositeKey, CompositeKey]: p, n = (pair.p, pair.n) if isinstance(pair, DiffPair) else pair return _endpoint_key(rmap, p), _endpoint_key(rmap, n) def _turns(turns: Iterable[Point]) -> pdpb2.RouteTurns: return pdpb2.RouteTurns(turns=[spb2.Point(x=x, y=y) for x, y in turns]) def _pose(position: Point, rotation: float) -> spb2.Pose: x, y = position return spb2.Pose(center=spb2.Point(x=x, y=y), angle=rotation) def _check_shape_length(name: str, value: object) -> None: if isinstance(value, bool) or not isinstance(value, (int, float)): raise TypeError( f"control point {name} must be a number, got {typeof(value).__name__}" ) if value <= 0: raise ValueError(f"control point {name} must be positive, got {value}") def _control_shape( radius: float | None, size: tuple[float, float] | None ) -> pdpb2.ControlShape | None: if radius is not None and size is not None: raise TypeError("pass either a radius or a size") if radius is not None: _check_shape_length("radius", radius) return pdpb2.ControlShape(circle=rcpb2.RouteTargetCircle(radius=radius)) if size is not None: w, h = size _check_shape_length("width", w) _check_shape_length("height", h) return pdpb2.ControlShape( rectangle=rcpb2.RouteTargetRectangle(width=w, height=h) ) return None def _current_placement(control: ControlPoint | PendingControlPoint) -> Placement | None: cp = control.control_point if isinstance(control, PendingControlPoint) else control if cp is None: return None t = cp.transform return t if isinstance(t, Placement) else None def _region_entry( rmap: ReverseMapper, feature: RegionFeature ) -> pdpb2.CreateAreaRegion: # The wire form of a region's kind, shape, layers, and side. T = epb2.AreaRegionType entry = pdpb2.CreateAreaRegion(shape=_region_shape(rmap, feature.shape)) if isinstance(feature, Pour): if feature.isolate: raise ValueError("a copper pour region takes no isolation; use constraints") entry.type = T.AREA_COPPER_POUR entry.layers.append(_layer(rmap, feature.layer)) entry.rank = feature.rank return entry if isinstance(feature, KeepOut): if not (feature.pour or feature.via or feature.route): raise ValueError("a keepout region must keep out pours, vias, or routes") entry.type = T.AREA_KEEPOUT entry.layers.extend(_region_layers(rmap, feature.layers)) entry.keepout.CopyFrom( epb2.AreaKeepout(pour=feature.pour, via=feature.via, route=feature.route) ) return entry if isinstance(feature, Cutout): entry.type = T.AREA_CUTOUT return entry if isinstance(feature, Courtyard) or not isinstance(feature, SurfaceFeature): raise ValueError(f"a {typeof(feature).__name__} can't be an area region") surface: list[tuple[type[SurfaceFeature], int]] = [ (Soldermask, T.AREA_SOLDERMASK), (Silkscreen, T.AREA_SILKSCREEN), (Paste, T.AREA_PASTEMASK), (Glue, T.AREA_GLUE), (Finish, T.AREA_FINISH), (Custom, T.AREA_CUSTOM), ] for cls, kind in surface: if isinstance(feature, cls): entry.type = cast(epb2.AreaRegionType, kind) break else: raise ValueError(f"a {typeof(feature).__name__} can't be an area region") if isinstance(feature, Custom): entry.name = feature.name entry.side = epb2.Side.BOTTOM if feature.side == Side.Bottom else epb2.Side.TOP return entry def _region_shape(rmap: ReverseMapper, shape: Shape) -> spb2.Shape: # A convex hull of board elements, or any shape as a board-absolute # polygon set. if isinstance(shape.geometry, ConvexHull): if shape.transform != IDENTITY: raise ValueError( "a ConvexHull is defined by its anchors; it can't be moved" ) hull = shape.geometry anchors = [k for a in hull.anchors for k in _anchor_keys(rmap, a)] if not anchors: raise ValueError("a ConvexHull needs at least one anchor") out = spb2.ConvexHull(anchors=[composite_id(k) for k in anchors]) if hull.expand is not None: out.expand = hull.expand return spb2.Shape(convex_hull=out) g = ShapelyGeometry.from_shape(shape).g polygons = ( list(g.geoms) if isinstance(g, shapely.MultiPolygon) else [g] if isinstance(g, shapely.Polygon) else [] ) if not polygons or any(p.is_empty for p in polygons): raise ValueError("an area region's shape must be an area") out = spb2.Shape() for p in polygons: component = out.polygon_set.components.add() _ring(component.outer, p.exterior) for hole in p.interiors: _ring(component.inners.add(), hole) return out def _ring(into: spb2.Polygon, ring: shapely.LinearRing) -> None: for x, y in list(ring.coords)[:-1]: point = into.points.add() point.x, point.y = x, y def _region_layers(rmap: ReverseMapper, layers: LayerSet) -> list[int]: out: list[int] = [] for start, end in layers.ranges: lo, hi = sorted((_layer(rmap, start), _layer(rmap, end))) out.extend(range(lo, hi + 1)) return sorted(set(out)) def _region_id( rmap: ReverseMapper, region: Feature | Pour | PendingRegion ) -> sipb2.StableID: # A region's id is a single StableID; a pour's key ends in it. key = region.key if isinstance(region, PendingRegion) else rmap.stable_id_of(region) if key is None: raise ValueError(f"the runtime has no ID for region {region}") return stable_id_proto(key[-1]) def _component_pads(component: Component) -> list[Pad]: return list(extract(component, Pad)) def _port_pads(rmap: ReverseMapper, port: Port) -> list[Pad]: # Every pad a port maps to, searching the port's component. port_key = rmap.stable_id_of(port) owner = _owning_component(port) mappings = ( extract(owner, PadMapping, through=()) if owner is not None else extract(rmap.root, PadMapping) ) for mapping in mappings: for mapped, pads in mapping.items(): if mapped is port or ( port_key is not None and rmap.stable_id_of(mapped) == port_key ): return [pads] if isinstance(pads, Pad) else list(pads) return [] def _anchor_keys(rmap: ReverseMapper, anchor: object) -> list[CompositeKey]: # The feature ids a hull anchor stands for. if isinstance(anchor, PendingVia | PendingControlPoint): return [anchor.key] if isinstance(anchor, Component): pads: list[object] = list(_component_pads(anchor)) elif isinstance(anchor, Port): pads = list(_port_pads(rmap, anchor)) else: pads = [anchor] keys = [] for pad in pads: key = rmap.stable_id_of(pad) if key is None: raise ValueError(f"the runtime has no ID for {pad}") keys.append(key) if not keys: raise ValueError(f"{anchor} has no pads") return keys def _origin_key(rmap: ReverseMapper, origin: object) -> CompositeKey: # The runtime's origin is a feature: a component or port stands for its # first pad. if isinstance(origin, PendingVia | PendingControlPoint): return origin.key if isinstance(origin, Component | Port): return _anchor_keys(rmap, origin)[0] key = rmap.stable_id_of(origin) if key is None: raise ValueError(f"the runtime has no ID for {origin}") return key def _net_key( rmap: ReverseMapper, net: Net | Pad | Port | Via | PendingVia ) -> CompositeKey: # A pour's connection is a feature on its net; for a net, a pad on it. if not isinstance(net, Net): return _origin_key(rmap, net) seen: set[int] = set() def ports(n: Net) -> Iterable[Port]: if id(n) in seen: return seen.add(id(n)) for c in n.connected: if isinstance(c, Net): yield from ports(c) elif isinstance(c, Port): yield c for port in ports(net): for pad in _port_pads(rmap, port): key = rmap.stable_id_of(pad) if key is not None: return key raise ValueError(f"net {net} has no pad the runtime knows; pass a pad")