from __future__ import annotations
import asyncio
from collections.abc import (
AsyncIterable,
Callable,
Coroutine,
Generator,
Iterable,
Mapping,
Sequence,
)
from contextlib import AbstractContextManager, contextmanager
from logging import getLogger
from pathlib import Path
import threading
from types import UnionType
from typing import Any, Protocol, cast, overload
import warnings
from jitx._structural import Proxy, _warn_skips
from jitx._translate.lookup import ComputedNet
from jitx._utils.unionfind import UnionFind
from jitx.copper import Copper
from jitx.feature import SurfaceFeature
from jitx.layerindex import Side
from jitx.net import Net, Port, PortAttachment, TopologyNet
from jitx.circuit import Circuit
from jitx.component import Component
from jitx.landpattern import PadMapping
from jitx.placement import Placement
from jitx.stackup import Stackup
from jitx.via import Via
from google.protobuf.json_format import ParseDict
from jitxcore._proto import (
configuration_pb2,
design_issues_pb2,
design_pb2,
idmap_pb2,
messages_pb2,
physical_design_pb2,
stable_id_pb2,
)
from .._translate.idmap import ReverseMapper, fqnameob
from .._translate.reverse_flow.placement import placement_from_pose_side_proto
from ._layout_events import (
apply_controls,
apply_pours,
apply_regions,
apply_routes,
apply_vias,
)
from .configuration import AsyncConfiguration, SyncConfiguration
from .layout import AsyncLayout, RequestRejected, SyncLayout
from .monitor import Changes, LayoutMonitor, _Monitors
from .._websocket import (
Envelope,
Message,
PersistentWebSocketClient,
Subscription,
design_route,
runtime_uri,
)
from ..design import Design
from ..inspect import Trace, extract, visit
from ..issue import IssueCategory, IssueReport
from ..query import query
from ..transform import IDENTITY, Transform
logger = getLogger(__name__)
def _warn_if_issues(rmap: ReverseMapper) -> None:
if not rmap.unattached_copper_warning:
return
W = 38
def row(text: str = "") -> str:
return "|" + (" " + text).center(W) + "|"
lines = [
"",
"+" + "-" * W + "+",
row("DESIGN COMPLETED WITH WARNINGS"),
row(),
row("The design may not export correctly."),
row("Please review the warning(s) above."),
"+" + "-" * W + "+",
"",
]
logger.warning("\n".join(lines))
[docs]
class Runtime:
class __Active(threading.local):
runtime: list[Runtime | SyncRuntime | None]
instantiation: list[AbstractContextManager[None]]
def __init__(self):
super().__init__()
self.runtime = []
self.instantiation = []
active = __Active()
@overload
def __init__(self): ...
@overload
def __init__(self, *, uri: str): ...
@overload
def __init__(self, *, project: str | Path): ...
@overload
def __init__(self, *, client: PersistentWebSocketClient): ...
def __init__(
self,
*,
uri: str | None = None,
project: str | Path | None = None,
client: PersistentWebSocketClient | None = None,
):
import jitx.run
# if a Runtime object is constructed, it's fairly safe to assume that
# we're not trying to autodetect a design run as '__main__'.
jitx.run.autodetect_design = False
if client:
self.__uri = client.uri
self.__client = client
else:
self.__uri = uri or (
runtime_uri(project=project) if project else runtime_uri()
)
self.__client = PersistentWebSocketClient(uri=self.__uri)
self.__reentered = 0
# Event subscriptions of submitted designs, and the tasks applying
# them; ended on close.
self.__watches: list[tuple[Subscription, asyncio.Task[None]]] = []
@property
def _client(self):
return self.__client
[docs]
def decouple(self):
return self.__class__(uri=self.__uri)
def __enter(self, r: Runtime | SyncRuntime):
from jitx._instantiation import instantiation
# design objects may construct structural objects lazily on access, so
# keep instantiation active while interacting with a runtime.
active = instantiation.require()
active.__enter__()
Runtime.active.instantiation.append(active)
Runtime.active.runtime.append(r)
if not self.__reentered:
self.__client.keepalive()
self.__reentered += 1
return self.__reentered == 1
def __exit(self):
self.__reentered -= 1
Runtime.active.runtime.pop()
Runtime.active.instantiation.pop().__exit__(None, None, None)
return self.__reentered == 0
async def __aenter__(self) -> Runtime:
if self.__enter(self):
await self.open()
return self
async def __aexit__(self, exc_type, exc_val, exc_tb):
if self.__exit():
await self.close()
[docs]
async def open(self):
await self.__client.establish()
[docs]
async def close(self):
watches, self.__watches = self.__watches, []
for subscription, task in watches:
task.cancel()
if not self.__client.is_closed():
try:
await subscription.unbind()
except Exception:
logger.debug("Failed to unbind design events", exc_info=True)
await self.__client.disconnect()
def __enter__(self) -> SyncRuntime:
sr = SyncRuntime(self)
if self.__enter(sr):
sr.open()
return sr
def __exit__(self, exc_type, exc_val, exc_tb):
sr = Runtime.active.runtime[-1]
if self.__exit():
# should always be true.
if isinstance(sr, SyncRuntime):
sr.close()
[docs]
@staticmethod
def require():
if Runtime.active.runtime:
r = Runtime.active.runtime[-1]
if not isinstance(r, Runtime | SyncRuntime):
raise TypeError("Active runtime is not a Runtime")
return r
else:
return Runtime()
def _construct[T: Design](self, design: T | type[T]) -> T:
return _instantiate_design(design)
def _package_design[T: Design](
self, constructed: T, name: str | None = None
) -> tuple[design_pb2.Design, ReverseMapper[T]]:
return _package_design(constructed, name)
[docs]
async def submit[T: Design](
self, design: T | type[T], *, name: str | None = None
) -> AsyncRuntimeDesign[T]:
constructed = self._construct(design)
packaged, rmap = self._package_design(constructed, name)
design_name = name or fqnameob(constructed)
rd = AsyncRuntimeDesign(_DesignState(constructed, rmap), design_name, self)
conversation = await self.__client.root().request(
Message(
"des",
"request",
messages_pb2.Request(
design_name=design_name,
design_manager=messages_pb2.DesignManagerRequest(
load=messages_pb2.LoadDesignTrigger(design=packaged)
),
),
),
)
from jitx.run._prompts import ConsolePrompts
async def answer(question_id: str, confirmed: bool) -> None:
await conversation.send(
Message("des", "answer", {"id": question_id, "confirmed": confirmed})
)
prompts = ConsolePrompts(answer)
async for envelope in conversation:
async with envelope.process(namespace="des") as des:
@des
def response(response: messages_pb2.Response):
if response.HasField("design_manager"):
pass
# trigger = response.design_manager.trigger
# if trigger.message:
# result_message = trigger.message
elif response.HasField("error"):
raise RuntimeError(f"Load failed: {response.error.message}")
@des
def ids(idmap: idmap_pb2.IDMap):
from jitx._instantiation import instantiation
# flattening might generate IDs that generate access to implicit
# elements, which then needs to trigger instantiation.
with instantiation.require(), instantiation.frame():
rd._rmap.populate_stable_id_chains(idmap)
# Asked of any open UI as well, so either may settle it.
@des
def prompt(p: messages_pb2.UserPrompt):
prompts.prompt(p)
@des
def settled(s: messages_pb2.Settled):
prompts.settled(s)
@des
async def stdin(data: Envelope):
# should no longer be used, prompt is used instead.
await conversation.send(data.message.response({"message": "N"}))
@des
def stdout(data: Mapping):
logger.info(data["message"])
@des
def unhandled():
raise RuntimeError(
f"Unhandled response message: {envelope.type}" # noqa: B023
)
# Keeps the design in step with the runtime, for changes from requests,
# the UI, and other clients alike. Before any capture, so changes made
# while capturing are held and applied to the captured tree.
events = await self.__client.route(design_route(design_name)).subscribe(
"phd", "event"
)
self.__watches.append(
(events, asyncio.create_task(_apply_events(events, rd._state)))
)
if constructed._event_context:
from jitx._instantiation import instantiation
with (
instantiation.require(),
instantiation.frame(),
constructed._event_context,
):
event = Design.Submitted(constructed)
event.fire()
if event._captures:
rd = await rd.capture()
for captured in event._captures:
captured(rd)
if event._resubmit:
return await self.submit(constructed, name=name)
return rd
[docs]
async def capture[T: RuntimeDesign](
self, design: T, *, name: str | None = None, wait: float = 5.0
) -> T:
"""Capture the runtime's physical layout back into the design.
Args:
design: The submitted design.
name: The runtime's name for the design; defaults to the design's
fully qualified name.
wait: Seconds to wait for background tasks to finish before the
layout is taken. 0 captures immediately.
"""
from jitx._translate.reverse_flow.linker import Capture
if name is None:
name = design.name
state = design._state
with state.capturing():
conversation = await self.__client.route(design_route(name)).request(
Message(
"des",
"request",
messages_pb2.Request(
design_name=name,
timeout=max(0, round(wait * 1000)),
physical_design=messages_pb2.PhysicalDesignRequest(
get_layout_output=messages_pb2.GetLayoutOutputQuery()
),
),
)
)
async for envelope in conversation:
with envelope.process(namespace="des") as des:
@des
def response(response: messages_pb2.Response):
layout = response.physical_design.layout_output.output
# A fresh tree each capture: the previous one stays as
# it was for anyone holding on to it.
root = Proxy.create(state.design)
rmap = state.rmap.rebase(root)
Capture(layout, rmap, design=root).run()
_warn_if_issues(rmap)
state.replace(root, rmap)
@des
def stdout(data: Mapping):
logger.info(data["message"])
@des
def failure(data: Mapping):
raise RuntimeError(
f"Failed to capture output ({data['kind']}): {data['message']}"
)
@des
def unhandled():
raise RuntimeError(
f"Unhandled response message: {envelope.type}" # noqa: B023
)
return design
async def _configuration(
self,
design_name: str,
operations: configuration_pb2.ConfigurationOperations,
) -> configuration_pb2.ConfigurationOperationsResponse:
"""Send configuration operations to change a submitted design.
Raises:
RequestRejected: if the runtime refused the operations.
"""
conversation = await self.__client.route(design_route(design_name)).request(
Message("des", "configuration", operations)
)
async for envelope in conversation:
if envelope.ns != "des":
continue
if envelope.type == "configuration":
return ParseDict(
dict(envelope.body),
configuration_pb2.ConfigurationOperationsResponse(),
ignore_unknown_fields=True,
)
if envelope.type == "failure":
body = envelope.body
operation = body.get("operation")
raise RequestRejected(
str(body.get("kind", "")),
str(body.get("message", "")),
int(operation) if operation is not None else None,
)
if envelope.type == "stdout":
logger.info(envelope.body["message"])
raise RuntimeError("the runtime sent no reply to the configuration operations")
async def _operations(
self, design_name: str, operations: physical_design_pb2.PhysicalDesignOperations
) -> physical_design_pb2.PhysicalDesignOperationsResponse:
"""Send operations to change a submitted design.
Raises:
RequestRejected: if the runtime refused the operations.
"""
conversation = await self.__client.route(design_route(design_name)).request(
Message("phd", "operations", operations)
)
async for envelope in conversation:
if envelope.ns != "phd":
continue
if envelope.type == "operations":
return ParseDict(
dict(envelope.body),
physical_design_pb2.PhysicalDesignOperationsResponse(),
ignore_unknown_fields=True,
)
if envelope.type == "failure":
body = envelope.body
operation = body.get("operation")
raise RequestRejected(
str(body.get("kind", "")),
str(body.get("message", "")),
int(operation) if operation is not None else None,
)
# The request is the runtime's current UI message, so the UI output
# it causes (events, progress, geometry) arrives here too; events
# are applied through the design's subscription.
raise RuntimeError("the runtime sent no reply to the operations")
async def _publish(
self, design_name: str, event: physical_design_pb2.PhysicalDesignEvent
) -> None:
"""Send an event to the subscribers of a submitted design's
events."""
# Blind: the runtime does not handle events, and no subscriber is not
# an error.
await self.__client.route(design_route(design_name)).send(
Message("phd", "event", event), blind=True
)
[docs]
async def issues(
self,
design: RuntimeDesign,
*,
categories: Iterable[IssueCategory | int] | None = None,
satisfied: bool = False,
wait: float = 5.0,
) -> IssueReport:
"""Query the physical-design issues of a submitted design.
Args:
design: The submitted design.
categories: Categories to report; all if None.
satisfied: Also report groups whose checks passed.
wait: Seconds to wait for background tasks to finish before the
snapshot is taken, like capture does. Categories still in
progress when the wait ends are reported as pending. 0 answers
immediately.
"""
from jitx._translate.reverse_flow.issue import (
parse_response,
report_from_proto,
)
name = design.name
# The stubs type enum fields as the proto enum wrapper, but any int is
# a valid (open) enum value on the wire, including ones this build of
# jitxcore does not name.
query = messages_pb2.GetDesignIssuesQuery(
categories=[
cast("design_issues_pb2.DesignIssueCategory", int(c))
for c in categories or ()
],
include_satisfied=satisfied,
)
conversation = await self.__client.route(f"design/{name}").request(
Message(
"des",
"request",
messages_pb2.Request(
design_name=name,
timeout=max(0, round(wait * 1000)),
physical_design=messages_pb2.PhysicalDesignRequest(
get_design_issues=query
),
),
)
)
report: IssueReport | None = None
async for envelope in conversation:
with envelope.process(namespace="des") as des:
@des
def response(data: Mapping):
nonlocal report
update = parse_response(data).physical_design.design_issues.update
rmap = design._issue_mapper(update.design_generation)
report = report_from_proto(update, rmap)
@des
def failure(data: Mapping):
raise RuntimeError(
f"Failed to query design issues ({data['kind']}): {data['message']}"
)
@des
def stdout(data: Mapping):
logger.info(data["message"])
@des
def unhandled():
raise RuntimeError(
f"Unhandled response message: {envelope.type}" # noqa: B023
)
if report is None:
raise RuntimeError("The runtime sent no design issue report")
return report
class _RunAsync(Protocol):
def __call__[T](self, coro: Coroutine[None, None, T], /) -> T: ...
[docs]
class SyncRuntime:
class _AsyncBridgeThread:
def __init__(self):
self._loop: asyncio.AbstractEventLoop | None = None
self._thread: threading.Thread | None = None
self._ready = threading.Event()
def start(self) -> None:
if self._thread is not None:
return
def _run():
loop = asyncio.new_event_loop()
self._loop = loop
asyncio.set_event_loop(loop)
# link calling and running thread's runtime stack
# it's a blocking call, so it's safe to do so.
self._ready.set()
try:
loop.run_forever()
finally:
pending = asyncio.all_tasks(loop)
for t in pending:
t.cancel()
loop.run_until_complete(
asyncio.gather(*pending, return_exceptions=True)
)
loop.close()
self._thread = threading.Thread(
target=_run, name="jitx-sync-async-bridge", daemon=True
)
self._thread.start()
self._ready.wait()
def run(self, coro):
assert self._loop is not None, "loop thread not started"
return asyncio.run_coroutine_threadsafe(coro, self._loop).result()
def close(self) -> None:
if self._loop is None or self._thread is None:
return
self._loop.call_soon_threadsafe(self._loop.stop)
self._thread.join()
self._loop = None
self._thread = None
self._ready.clear()
def __init__(self, runtime: Runtime):
self.__runtime = runtime
self.__loop_thread = SyncRuntime._AsyncBridgeThread()
[docs]
def runasync[T](self, coro: Coroutine[None, None, T]) -> T:
from jitx._instantiation import instantiation
self.__loop_thread.start()
outer_instantiation = instantiation.data.stack
async def decoupler():
instantiation.data.stack = outer_instantiation
async with self.__runtime.decouple():
return await coro
return self.__loop_thread.run(decoupler())
# Technically we'd be free to run an event loop here if there isn't one,
# however, if we, for example, run websockets open like this, and then
# websockets close the same way, those will be in different event loops,
# and it has callbacks from the first event loop running in the second
# which it does not like.
# return asyncio.run(coro)
# So it turns out it doesn't matter whether there's an event loop on
# this thread or not, we always have to start a new one.
def __enter__(self):
return self.__runtime.__enter__()
def __exit__(self, exc_type, exc_ob, exc_tb):
return self.__runtime.__exit__(exc_type, exc_ob, exc_tb)
# these should logically not be possible to construct from code,
# and thus could just be exceptions, but it seems fine to keep the parity.
# having them defined allows the Runtime|SyncRuntime return type to not
def __aenter__(self):
return self.__runtime.__aenter__()
def __aexit__(self, exc_type, exc_ob, exc_tb):
return self.__runtime.__aexit__(exc_type, exc_ob, exc_tb)
[docs]
def close(self):
self.runasync(self.__runtime.close())
self.__loop_thread.close()
[docs]
def open(self):
self.runasync(self.__runtime.open())
[docs]
def submit[T: Design](self, design: T | type[T], *, name: str | None = None):
ard = self.runasync(self.__runtime.submit(design, name=name))
return SyncRuntimeDesign(ard._state, ard.name, self.__runtime, self.runasync)
[docs]
def capture[T: RuntimeDesign](
self, design: T, *, name: str | None = None, wait: float = 5.0
) -> T:
return self.runasync(self.__runtime.capture(design, name=name, wait=wait))
[docs]
def issues(
self,
design: RuntimeDesign,
*,
categories: Iterable[IssueCategory | int] | None = None,
satisfied: bool = False,
wait: float = 5.0,
) -> IssueReport:
"""Query the physical-design issues of a submitted design. See
:py:meth:`Runtime.issues`."""
return self.runasync(
self.__runtime.issues(
design, categories=categories, satisfied=satisfied, wait=wait
)
)
class _RuntimeDispatch:
@overload
def __call__(self) -> Runtime | SyncRuntime: ...
@overload
def __call__(self, *, uri: str) -> Runtime: ...
@overload
def __call__(self, *, project: str | Path) -> Runtime: ...
def __call__(self, *, uri: str | None = None, project: str | Path | None = None):
if uri:
r = Runtime(uri=uri)
elif project:
r = Runtime(project=project)
else:
r = Runtime.require()
return r
def __enter__(self) -> SyncRuntime:
return self().__enter__()
def __exit__(self, exc_type, exc_ob, exc_tb):
r = Runtime.active.runtime[-1]
assert r is not None
return r.__exit__(exc_type, exc_ob, exc_tb)
async def __aenter__(self) -> Runtime:
return await self().__aenter__()
async def __aexit__(self, exc_type, exc_ob, exc_tb):
r = Runtime.active.runtime[-1]
assert r is not None
return await r.__aexit__(exc_type, exc_ob, exc_tb)
async def submit[T: Design](self, design: T | type[T], *, name: str | None = None):
async with Runtime.require() as r:
return await r.submit(design, name=name)
async def capture[T: RuntimeDesign](
self, design: T, *, name: str | None = None, wait: float = 5.0
) -> T:
async with Runtime.require() as r:
return await r.capture(design, name=name, wait=wait)
async def issues(
self,
design: RuntimeDesign,
*,
categories: Iterable[IssueCategory | int] | None = None,
satisfied: bool = False,
wait: float = 5.0,
) -> IssueReport:
"""Query the physical-design issues of a submitted design. See
:py:meth:`Runtime.issues`."""
async with Runtime.require() as r:
return await r.issues(
design, categories=categories, satisfied=satisfied, wait=wait
)
@staticmethod
def require():
return Runtime.require()
dispatch = _RuntimeDispatch()
[docs]
class RuntimeDesign[T: Design]:
"""A design that has been submitted to the runtime.
``root`` is the instantiated design object. The rest of this class gives
access to what the runtime knows about it: use :py:meth:`~RuntimeDesign.query`
to walk the design tree, :py:attr:`nets` for connectivity, and :py:attr:`layers`
to resolve layer indices and board sides through placements.
"""
def __init__(self, state: _DesignState[T], name: str):
self._state = state
self.__name = name
@property
def root(self) -> T:
"""The design as last captured. A capture replaces it with a new
object; one held on to keeps its captured state."""
return self._state.root
@property
def _rmap(self) -> ReverseMapper[T]:
return self._state.rmap
def _invalidate(self):
self._state.nets = None
def _issue_mapper(self, generation: str) -> ReverseMapper | None:
# The load reply carries no generation, so the first issue report
# records it; a later mismatch means the runtime reloaded the design
# and ids no longer match this process's mapper. Kept in the shared
# state, so every view of the design agrees.
state = self._state
if state.issue_generation is None:
state.issue_generation = generation
if generation == state.issue_generation:
return self._rmap
if not state.warned_generation:
state.warned_generation = True
logger.warning(
"The runtime's design %s has changed since it was first "
"queried; design issue objects are left unresolved.",
self.__name,
)
return None
@property
def name(self):
"""The name the runtime knows this design by, either the ``name``
passed to :py:meth:`Runtime.submit` or the fully qualified name of the
design object."""
return self.__name
[docs]
def query[Q](
self,
target: type[Q] | tuple[type[Q], ...],
/,
*,
through: tuple[type, ...] | UnionType | None = None,
transform: Transform | None = IDENTITY,
opaque: tuple[type, ...] | UnionType | None = None,
refs: bool = False,
filter: Callable[[Q], bool] | None = None,
) -> Generator[tuple[Trace, Q], None, None]:
"""Convenience method to query this design. See
:py:meth:`~jitx.query.TransformQuery.query` for details."""
return query(
self.root,
target,
through=through,
transform=transform,
opaque=opaque,
refs=refs,
filter=filter,
)
@property
def nets(self):
"""The connectivity of this design computed by the runtime, as a :class:`Nets` instance."""
if self._state.nets is not None:
return self._state.nets
uf: UnionFind[Net | TopologyNet] = UnionFind()
for net in extract(self.root, Net | TopologyNet):
uf.add(net)
if cn := ComputedNet.get(net):
if cn.net:
uf.union(net, cn.net)
non_computed_count = 0
for net in extract(self.root, Net | TopologyNet):
if isinstance(net, Net):
def recurse(net):
count = 0
for n in net._connected:
if isinstance(n, Net):
if not uf.connected(net, n):
count += 1
uf.union(net, n)
count += recurse(n)
return count
non_computed_count += recurse(net)
if non_computed_count:
logger.debug(
"Unioned nets without computed net connection: %d", non_computed_count
)
for netgroup in uf.groups().values():
for net in netgroup:
under = net
while Proxy.is_proxy(under):
under = Proxy.of(under)
# casts here because types in python can't be narrowed to Net & Proxied.
uf.union(cast(Net, net), cast(Net, under))
missing_nets: list[tuple[Trace, Component, set[Port]]] = []
for trace, comp in visit(self.root, (Component, PortAttachment)):
if isinstance(comp, Component):
missing_net_on_comp: list[Port] = []
for pm in extract(comp, PadMapping):
for port, pads in pm.items():
cn = ComputedNet.get(port)
if not cn:
missing_net_on_comp.append(port)
continue
if not isinstance(pads, Sequence):
pads = (pads,)
for pad in pads:
cn.assign(pad)
if missing_net_on_comp:
missing_nets.append((trace, comp, set(missing_net_on_comp)))
elif isinstance(comp, PortAttachment):
if isinstance(comp.port, Sequence):
for ap, cp in zip(
extract(comp.attachment, Port, filter=Port.is_single_pin),
comp.port,
strict=False,
):
cn = ComputedNet.get(cp)
if cn:
cn.assign(ap)
else:
cn = ComputedNet.get(comp.port)
if cn:
cn.assign(comp.attachment)
if missing_nets:
for trace, comp, ports in missing_nets:
portnames = []
for t, _pt in visit(
comp, Port, filter=lambda p, ports=ports: p in ports
):
portnames.append(str(t.path))
if len(portnames) > 1:
portnames.insert(0, "") # don't leave one on the same line
logger.warning(
"no computed net for port in component %s: %s",
trace.path,
"\n ".join(portnames),
)
self._state.nets = self.Nets(uf)
return self._state.nets
@property
def layers(self):
"""Layer index and board side arithmetic for this design's stackup, as
a :class:`Layers` object."""
return self.Layers(self.root.substrate.stackup)
[docs]
class Net[N: Port]:
"""A single net of a captured design.
``nets`` holds the :class:`~jitx.net.Net` and
:class:`~jitx.net.TopologyNet` objects that are connected,
and ``name`` is their connected name.
"""
def __init__(self, nets: Sequence[Net[N] | TopologyNet[N]]):
cname = None
for net in nets:
nname = net._connected_name
if nname:
if cname and cname != nname:
logger.warning(
"Duplicate connected names encountered for nets '%s' and '%s'",
cname,
net._connected_name,
)
else:
cname = nname
self.nets = nets
self.name = cname
@overload
def __iter__(
self: RuntimeDesign.Net[Port],
) -> Generator[Port | Copper | Via, None, None]: ...
@overload
def __iter__(self) -> Generator[N, None, None]: ...
def __iter__(self):
"""Yield every port, copper shape, and via on this net."""
# cast to erase the port type, or pyright thinks the overloads
# might be incompatible - they can't really be at this point,
# but hard to prove.
nets = cast(Sequence[Net], self.nets)
for net in nets:
yield from net
[docs]
class Nets:
"""A manager for looking up the net that an element belongs to."""
def __init__(self, uf: UnionFind[Net | TopologyNet]):
self.__uf = uf
self.__nets = {
root: RuntimeDesign.Net(group) for root, group in uf.groups().items()
}
[docs]
def find(self, element) -> RuntimeDesign.Net | None:
"""Look up the net an element belongs to.
Args:
element: A :class:`~jitx.net.Net` or
:class:`~jitx.net.TopologyNet`, or an element the runtime
assigned to a net, such as a :class:`~jitx.copper.Copper`
or a :class:`~jitx.via.Via`.
Returns:
The :class:`Net` the element is on, or None if it is not on any
net of this design.
"""
if isinstance(element, Net | TopologyNet):
net = element
else:
cnet = ComputedNet.get(element)
if cnet and cnet.net:
net = cnet.net
else:
return None
if net not in self.__uf:
return None
return self.__nets.get(self.__uf.find(net))
def __call__(self):
"""Return self.
.. version-deprecated:: 4.6
Use the ``nets`` property directly.
"""
warnings.warn(
"design.nets() is deprecated; nets is a property, use design.nets",
RuntimeWarning,
skip_file_prefixes=_warn_skips,
stacklevel=0,
)
return self
[docs]
class Layers:
"""A manager for layer index and board side arithmetic."""
def __init__(self, stackup: Stackup):
self.__conductor_count = len(stackup.conductors)
[docs]
def normalize(self, layer: int):
"""Return a 0-indexed ``layer`` counted from the top conductor.
Args:
layer: The layer index to normalize (supports negative indexing).
Returns:
A normalized 0-indexed integer.
"""
if layer < 0:
layer += self.__conductor_count
return layer
[docs]
def apply(self, transform: Transform | Placement, layer: int) -> int:
"""Return the conductor index, counted from the top, that an element
on ``layer`` ends up on once ``transform`` is applied.
Args:
transform: The Transform or Placement to apply.
layer: The conductor layer index to transform.
Returns:
The resulting normalized conductor layer index.
"""
if isinstance(transform, Placement) and transform.side:
layer = -layer - 1
return self.normalize(layer)
[docs]
def side(
self, transform: Transform | Placement, feature: SurfaceFeature
) -> Side:
"""Return the board side a :class:`SurfaceFeature` ends up on
with a transform applied.
Args:
transform: The Transform or Placement to apply.
feature: The SurfaceFeature to transform.
Returns:
The resulting :class:`Side`.
"""
placed = transform.side if isinstance(transform, Placement) else Side.Top
return placed * feature.side
def __call__(self):
"""Return self.
.. version-deprecated:: 4.6
Use the ``layers`` property directly.
"""
warnings.warn(
"design.layers() is deprecated; layers is a property, "
"use design.layers",
RuntimeWarning,
skip_file_prefixes=_warn_skips,
stacklevel=0,
)
return self
[docs]
class AsyncRuntimeDesign[T: Design](RuntimeDesign[T]):
def __init__(self, state: _DesignState[T], name: str, runtime: Runtime):
super().__init__(state, name)
self.__runtime = runtime
self.__layout = AsyncLayout(
lambda: state.rmap,
_operations_sender(runtime, state, name),
monitor=LayoutMonitor(state.monitors),
publish=lambda event: runtime._publish(name, event),
)
self.__configuration = AsyncConfiguration(
lambda: state.rmap,
lambda operations: runtime._configuration(name, operations),
)
@property
def layout(self) -> AsyncLayout:
"""The physical design (layout) of this design in the runtime:
requests that change it, e.g. ``await design.layout.place(...)``, and
:py:attr:`~AsyncLayout.monitor` for callbacks on its changes. Changes
are applied to :py:attr:`root` as the runtime reports them."""
return self.__layout
[docs]
async def capture(self, *, wait: float = 5.0):
"""Capture the runtime's physical layout back into this design. See
:py:meth:`Runtime.capture`."""
return await self.__runtime.capture(self, wait=wait)
[docs]
async def issues(
self,
*,
categories: Iterable[IssueCategory | int] | None = None,
satisfied: bool = False,
wait: float = 5.0,
) -> IssueReport:
"""Query the physical-design issues of this design. See
:py:meth:`Runtime.issues`."""
return await self.__runtime.issues(
self, categories=categories, satisfied=satisfied, wait=wait
)
@property
def configuration(self) -> AsyncConfiguration:
"""The configuration of this design in the runtime: requests that
change settings applied on its next load, e.g.
``await design.configuration.reassign_reference_designators(...)``."""
return self.__configuration
def _sync(self, runasync: _RunAsync) -> SyncRuntimeDesign[T]:
# A sync view sharing this design's state, for code running off the
# event loop; runasync runs a coroutine on the loop and waits for it.
return SyncRuntimeDesign(self._state, self.name, self.__runtime, runasync)
[docs]
class SyncRuntimeDesign[T: Design](RuntimeDesign[T]):
def __init__(
self,
state: _DesignState[T],
name: str,
runtime: Runtime,
runasync: _RunAsync,
):
super().__init__(state, name)
self.__runtime = runtime
self.__runasync = runasync = _off_loop(runasync)
send = _operations_sender(runtime, state, name)
self.__layout = SyncLayout(
lambda: state.rmap,
lambda operations: runasync(send(operations)),
monitor=LayoutMonitor(state.monitors),
publish=lambda event: runasync(runtime._publish(name, event)),
)
self.__configuration = SyncConfiguration(
lambda: state.rmap,
lambda operations: runasync(runtime._configuration(name, operations)),
)
@property
def layout(self) -> SyncLayout:
"""The physical design (layout) of this design in the runtime:
requests that change it, e.g. ``design.layout.place(...)``, and
:py:attr:`~SyncLayout.monitor` for callbacks on its changes. Changes
are applied to :py:attr:`root` as the runtime reports them."""
return self.__layout
[docs]
def capture(self, *, wait: float = 5.0):
"""Capture the runtime's physical layout back into this design. See
:py:meth:`Runtime.capture`."""
return self.__runasync(self.__runtime.capture(self, wait=wait))
[docs]
def issues(
self,
*,
categories: Iterable[IssueCategory | int] | None = None,
satisfied: bool = False,
wait: float = 5.0,
) -> IssueReport:
"""Query the physical-design issues of this design. See
:py:meth:`Runtime.issues`."""
return self.__runasync(
self.__runtime.issues(
self, categories=categories, satisfied=satisfied, wait=wait
)
)
@property
def configuration(self) -> SyncConfiguration:
"""The configuration of this design in the runtime: requests that
change settings applied on its next load, e.g.
``design.configuration.reassign_reference_designators(...)``."""
return self.__configuration
def _off_loop(runasync: _RunAsync) -> _RunAsync:
# Waiting on the loop from the loop's own thread (a monitor callback)
# would deadlock; refuse instead.
def run[T](coro: Coroutine[None, None, T], /) -> T:
try:
asyncio.get_running_loop()
except RuntimeError:
return runasync(coro)
coro.close()
raise RuntimeError(
"a sync runtime design cannot wait on the runtime from an event "
"loop, such as in a layout monitor callback"
)
return run
class _DesignState[T: Design]:
"""The mutable state of a submitted design, shared by every view of it."""
# Views (async, sync, request objects) hold this rather than copies, so a
# capture or a placement update is seen by all of them.
def __init__(self, design: T, rmap: ReverseMapper[T]):
self.design = design
self.root: T = rmap.root
self.rmap = rmap
self.nets: RuntimeDesign.Nets | None = None
# The design generation of the first issue report, and whether a
# later mismatch was warned about.
self.issue_generation: str | None = None
self.warned_generation = False
self.monitors = _Monitors()
self.__queued: list[Callable[[], None]] | None = None
def replace(self, root: T, rmap: ReverseMapper[T]) -> None:
"""Make a newly captured tree the current one."""
self.rmap.dispose_links()
self.root = root
self.rmap = rmap
self.nets = None
@contextmanager
def capturing(self) -> Generator[None, None, None]:
"""Hold updates while a capture is running, and apply them to the
captured tree afterwards."""
# Updates arriving between the runtime producing the layout and the
# new tree replacing the old one would otherwise be lost with the old
# tree. They are absolute (placements) or keyed by runtime id (routes,
# vias), so applying them again is harmless.
self.__queued = []
try:
yield
finally:
queued, self.__queued = self.__queued, None
for update in queued:
update()
def __apply(self, update: Callable[[], None]) -> None:
if self.__queued is not None:
self.__queued.append(update)
else:
update()
def apply_placements(
self, placements: Sequence[tuple[stable_id_pb2.CompositeID, Placement]]
) -> None:
"""Apply board-absolute instance placements reported by the runtime."""
placements = list(placements)
def update() -> None:
self.monitors.notify_placements(
_apply_instance_placements(self.root, self.rmap, placements)
)
self.__apply(update)
def apply_routes(self, changed: physical_design_pb2.RoutesChanged) -> None:
"""Apply routes the runtime reports as routed, blocked, or removed."""
def update() -> None:
changes = apply_routes(self.root, self.rmap, changed)
self.nets = None
self.monitors.notify("routes", changes)
self.__apply(update)
def apply_vias(self, changed: physical_design_pb2.ViasChanged) -> None:
"""Apply vias the runtime reports as added, changed, or removed."""
def update() -> None:
changes = apply_vias(self.root, self.rmap, changed)
self.nets = None
self.monitors.notify("vias", changes)
self.__apply(update)
def apply_controls(self, changed: physical_design_pb2.ControlPointsChanged) -> None:
"""Apply control points the runtime reports as added, changed, or
removed."""
def update() -> None:
changes = apply_controls(self.root, self.rmap, changed)
self.nets = None
self.monitors.notify("controls", changes)
self.__apply(update)
def apply_regions(self, changed: physical_design_pb2.AreaRegionsChanged) -> None:
"""Apply area regions the runtime reports as added, changed, or
removed."""
def update() -> None:
changes = apply_regions(self.root, self.rmap, changed)
self.nets = None
self.monitors.notify("regions", changes)
self.__apply(update)
def apply_pours(self, changed: physical_design_pb2.PoursChanged) -> None:
"""Apply pours the runtime reports as computed or removed."""
def update() -> None:
changes = apply_pours(self.root, self.rmap, changed)
self.nets = None
self.monitors.notify("pours", changes)
self.__apply(update)
def apply_event(self, event: physical_design_pb2.PhysicalDesignEvent) -> None:
"""Apply a ``phd``/``event`` from the runtime, then notify the
monitors."""
def update() -> None:
# Placements first: the other kinds are placed in the frames they
# set. Monitors are notified once the whole transition is applied,
# user events last.
moved: list[Component | Circuit] = []
changes: list[tuple[str, Changes[Any]]] = []
if event.HasField("placements"):
moved = _apply_instance_placements(
self.root, self.rmap, _placements(event.placements.placements)
)
if event.HasField("controls"):
c = apply_controls(self.root, self.rmap, event.controls)
changes.append(("controls", c))
if event.HasField("regions"):
changes.append(
("regions", apply_regions(self.root, self.rmap, event.regions))
)
if event.HasField("pours"):
changes.append(
("pours", apply_pours(self.root, self.rmap, event.pours))
)
if event.HasField("routes"):
changes.append(
("routes", apply_routes(self.root, self.rmap, event.routes))
)
if event.HasField("vias"):
changes.append(("vias", apply_vias(self.root, self.rmap, event.vias)))
if changes:
self.nets = None
self.monitors.notify_placements(moved)
for kind, c in changes:
self.monitors.notify(kind, c)
if event.HasField("user"):
self.monitors.notify_user(event.user, self.rmap)
self.__apply(update)
def apply_results(
self, response: physical_design_pb2.PhysicalDesignOperationsResponse
) -> None:
"""Apply the placements an operations reply reports."""
self.apply_placements(
_placements(
ip
for result in response.results
for ip in result.place_components.placements
)
)
async def _apply_events(events: AsyncIterable[Envelope], state: _DesignState) -> None:
# Applies a design's ``phd``/``event`` messages until the subscription
# ends.
async for envelope in events:
try:
event = ParseDict(
dict(envelope.body),
physical_design_pb2.PhysicalDesignEvent(),
ignore_unknown_fields=True,
)
state.apply_event(event)
except Exception:
logger.exception("Failed to apply a physical design event")
def _placements(
placements: Iterable[physical_design_pb2.InstancePlacement],
) -> list[tuple[stable_id_pb2.CompositeID, Placement]]:
# The model's form of reported board-absolute placements.
return [
(ip.instance, placement_from_pose_side_proto(ip.pose, ip.side))
for ip in placements
]
def _operations_sender(
runtime: Runtime, state: _DesignState, name: str
) -> Callable[
[physical_design_pb2.PhysicalDesignOperations],
Coroutine[None, None, physical_design_pb2.PhysicalDesignOperationsResponse],
]:
async def send(
operations: physical_design_pb2.PhysicalDesignOperations,
) -> physical_design_pb2.PhysicalDesignOperationsResponse:
response = await runtime._operations(name, operations)
state.apply_results(response)
return response
return send
def _apply_instance_placements(
root: Design,
rmap: ReverseMapper,
placements: Sequence[tuple[stable_id_pb2.CompositeID, Placement]],
) -> list[Component | Circuit]:
# Returns the instances whose transform changed.
from jitx._translate.reverse_flow.resolve import resolve
targets: dict[int, Placement] = {}
for instance, placement in placements:
obj = resolve(
instance,
rmap,
False,
context="InstancePlacement",
expect=(Component, Circuit),
)
if obj is None:
logger.debug("No instance for placement update %s", instance)
continue
targets[id(obj)] = placement
moved: list[Component | Circuit] = []
if not targets:
return moved
# Top-down, as capture does: a moved circuit gets its transform before its
# contents are made relative to it.
def loop(parent: Circuit, transform: Transform) -> None:
for t, component in visit(parent, Component, through=(), transform=transform):
placement = targets.get(id(component))
if placement is not None and t.transform is not None:
ctx = ~t.transform * placement
if component.transform != ctx:
component.transform = ctx
moved.append(component)
for t, circuit in visit(parent, Circuit, through=(), transform=transform):
if t.transform is None:
continue
placement = targets.get(id(circuit))
if placement is not None:
ctx = ~t.transform * placement
if circuit.transform != ctx:
circuit.transform = ctx
moved.append(circuit)
loop(
circuit,
t.transform * circuit.transform if circuit.transform else t.transform,
)
top = root if isinstance(root, Circuit) else root.circuit
loop(top, IDENTITY)
return moved
def _instantiate_design[T: Design](design: T | type[T]) -> T:
from jitx._instantiation import instantiation
from jitx._structural import Instantiable, Proxy
import gc
if isinstance(design, Instantiable):
cls = design._instantiable_()
if not (isinstance(cls, type) and issubclass(cls, Design)):
raise ValueError(f"Not an instantiable design: {cls}")
with instantiation.require():
with instantiation.frame():
instantiated = design._instantiate_({})
elif isinstance(design, type):
if not issubclass(design, Design):
raise ValueError(f"Not a Design subclass: {cls}")
try:
if hasattr(design, "__signature__"):
design.__signature__.bind()
except TypeError:
raise ValueError(
"Design is parameterized but no parameters were provided"
) from None
with instantiation.require():
with instantiation.frame():
instantiated = design()
else:
instantiated = design
assert isinstance(instantiated, Design), (
f"Instantiation resulted in non-Design object {Proxy.type(instantiated)}"
)
gc.collect()
return cast(T, instantiated)
def _package_design[T: Design](
design: T, name: str | None = None
) -> tuple[design_pb2.Design, ReverseMapper[T]]:
from jitx._structural import Proxy
from jitx._instantiation import instantiation
from jitx._translate.design import package_design
from .._translate.idmap import idmap
if not isinstance(design, Design):
raise TypeError(f"{design} is not a Design class")
idmap.clear()
with instantiation.require():
packaged = package_design(design, name)
rmap = idmap.export(Proxy.create(design))
return packaged, rmap
