"""Callbacks on changes to a runtime design's layout.
Reached through ``RuntimeDesign.layout.monitor``. Changes are reported however
they were made: by requests, in the UI, or by other clients. User events,
subclasses of :class:`UserEvent`, are application-defined events sent with
``RuntimeDesign.layout.emit`` or by other clients (such as UI hooks).
"""
from __future__ import annotations
import asyncio
from collections.abc import Callable, Mapping, Sequence
from dataclasses import dataclass
import inspect
from logging import getLogger
import math
from types import TracebackType
from typing import Any, cast
import jitxcore._proto.physical_design_pb2 as pdpb2
from google.protobuf.json_format import MessageToDict
from jitx.circuit import Circuit, Route
from jitx.component import Component
from jitx.controlpoint import ControlPoint
from jitx.copper import Pour
from jitx.feature import Feature
from jitx.via import Via
from .._translate.idmap import ReverseMapper, composite_id, fqname
from .._translate.reverse_flow.resolve import resolve
logger = getLogger(__name__)
type JSON = str | int | float | bool | None | list[JSON] | dict[str, JSON]
"""A JSON-compatible value."""
[docs]
class UserEvent:
"""Base for application-defined layout events.
Subclass it to define a kind of event; the subclass's fully qualified name
identifies the kind on the wire, so events of different applications do
not collide. An event carries JSON-compatible ``data`` and a sequence of
layout ``objects`` (components, ports, routes, vias, ...), which ``data``
may refer to by index. A subclass may parse both into its own fields, and
may override the :attr:`data` and :attr:`objects` properties, which are
what is sent.
A received event is constructed as ``cls(data, objects)``, so a subclass
must accept those two arguments positionally, and call this ``__init__``.
>>> class Highlight(UserEvent):
... def __init__(self, data: Mapping[str, JSON], objects: Sequence[object | None]):
... super().__init__(data, objects)
... self.color = data["color"]
>>> await design.layout.emit(Highlight({"color": "red"}, [c.u1]))
Args:
data: JSON-compatible values. Numbers that are whole are received as
``int``, so they can index :attr:`objects`.
objects: Layout objects. A received object the design does not have
is None; see :attr:`resolved`.
"""
def __init__(
self,
data: Mapping[str, JSON] | None = None,
objects: Sequence[object | None] = (),
):
self.__data = dict(data or {})
self.__objects = tuple(objects)
self._resolved = True
@property
def data(self) -> Mapping[str, JSON]:
"""The event's JSON-compatible values."""
return self.__data
@property
def objects(self) -> Sequence[object | None]:
"""The event's layout objects."""
return self.__objects
@property
def resolved(self) -> bool:
"""Whether every object of a received event was found in the design.
Objects not found are None in :attr:`objects`."""
return self._resolved
[docs]
@dataclass(frozen=True)
class Changes[T]:
"""Objects of one kind added, changed, or removed by one change to the
layout. Removed objects are no longer in the design tree. An object the
runtime replaces when it changes (a via, a control point, an area region's
feature) is reported as removed, and its replacement as added."""
added: Sequence[T] = ()
changed: Sequence[T] = ()
removed: Sequence[T] = ()
def __bool__(self) -> bool:
return bool(self.added or self.changed or self.removed)
class _ChangeLog[T]:
# Collects the changes to one kind of object while a change is applied.
def __init__(self):
self.__added: dict[int, T] = {}
self.__changed: dict[int, T] = {}
self.__removed: dict[int, T] = {}
def add(self, obj: T) -> None:
self.__added[id(obj)] = obj
def change(self, obj: T) -> None:
if id(obj) not in self.__added:
self.__changed[id(obj)] = obj
def remove(self, obj: T) -> None:
self.__changed.pop(id(obj), None)
# Made and gone within the change: nothing to report.
if self.__added.pop(id(obj), None) is None:
self.__removed[id(obj)] = obj
def changes(self) -> Changes[T]:
return Changes(
tuple(self.__added.values()),
tuple(self.__changed.values()),
tuple(self.__removed.values()),
)
[docs]
class Monitor:
"""A registered callback. Unregister it with :meth:`close`, or by using
it as a context manager.
>>> with design.layout.monitor.placements(moved):
... ...
"""
def __init__(self, remove: Callable[[], None]):
self.__remove: Callable[[], None] | None = remove
[docs]
def close(self) -> None:
"""Unregister the callback. Calling it again is a no-op."""
remove, self.__remove = self.__remove, None
if remove is not None:
remove()
def __enter__(self) -> Monitor:
return self
def __exit__(
self,
exc_type: type[BaseException] | None,
exc: BaseException | None,
tb: TracebackType | None,
) -> None:
self.close()
[docs]
class LayoutMonitor:
"""Registers callbacks run on changes to a runtime design's layout.
Callbacks run on the runtime's event loop as changes arrive, after the
whole change has been applied to ``design.root``, and receive objects of
the current tree (a capture replaces it). A callback returning an
awaitable has it run as a task. A callback must not block: blocking calls
on a sync design (requests, capture) raise when made from one. Exceptions
raised by a callback are logged.
"""
def __init__(self, monitors: _Monitors):
self.__monitors = monitors
[docs]
def placements(
self, callback: Callable[[Sequence[Component | Circuit]], object]
) -> Monitor:
"""Run ``callback`` with the instances whose placement changed.
Each moved instance is reported, including those anchored to a moved
one; circuits are reported when they move as a whole.
Returns:
The registration; close it to stop.
"""
entries = self.__monitors.placements
entry = (callback,)
entries.append(entry)
return Monitor(lambda: entries.remove(entry))
[docs]
def routes(self, callback: Callable[[Changes[Route]], object]) -> Monitor:
"""Run ``callback`` with the routes the runtime made, rerouted
(routed or blocked), or removed.
Returns:
The registration; close it to stop.
"""
return self.__monitors.register("routes", callback)
[docs]
def vias(self, callback: Callable[[Changes[Via]], object]) -> Monitor:
"""Run ``callback`` with the vias the runtime added, moved, or
removed. A via placed in code is not reported.
Returns:
The registration; close it to stop.
"""
return self.__monitors.register("vias", callback)
[docs]
def controls(self, callback: Callable[[Changes[ControlPoint]], object]) -> Monitor:
"""Run ``callback`` with the control points the runtime added,
changed, or removed. A control point placed in code is not reported.
Returns:
The registration; close it to stop.
"""
return self.__monitors.register("controls", callback)
[docs]
def regions(self, callback: Callable[[Changes[Feature | Pour]], object]) -> Monitor:
"""Run ``callback`` with the area regions the runtime added, changed,
or removed: a feature per region, and for a copper pour region a
:class:`~jitx.copper.Pour` per layer (changed when its rank changes;
see :meth:`pours` for its copper).
Returns:
The registration; close it to stop.
"""
return self.__monitors.register("regions", callback)
[docs]
def pours(self, callback: Callable[[Changes[Pour]], object]) -> Monitor:
"""Run ``callback`` with the pours whose copper the runtime computed
(changed, or added for a pour the design did not have), or removed. A
pour removed with its area region is reported by :meth:`regions`.
Returns:
The registration; close it to stop.
"""
return self.__monitors.register("pours", callback)
[docs]
def user[E: UserEvent](
self, event: type[E], callback: Callable[[E], object]
) -> Monitor:
"""Run ``callback`` with each received user event of type ``event``.
Args:
event: The event class. Only events sent as exactly this class
are received.
callback: Receives the event, constructed as ``event(data,
objects)``.
Returns:
The registration; close it to stop.
"""
entries = self.__monitors.user.setdefault(user_event_name(event), [])
entry = (event, cast(Callable[[UserEvent], object], callback))
entries.append(entry)
return Monitor(lambda: entries.remove(entry))
[docs]
def user_event_name(event: type[UserEvent]) -> str:
"""The wire name of a user event class."""
return fqname(event)
class _Monitors:
# The callbacks registered on a design, shared by its views so they
# survive captures.
def __init__(self):
self.placements: list[
tuple[Callable[[Sequence[Component | Circuit]], object]]
] = []
self.user: dict[
str, list[tuple[type[UserEvent], Callable[[UserEvent], object]]]
] = {}
self.kinds: dict[str, list[tuple[Callable[[Any], object]]]] = {}
# Keeps tasks of async callbacks alive until they finish.
self.__tasks: set[asyncio.Future[Any]] = set()
def notify_placements(self, instances: Sequence[Component | Circuit]) -> None:
if not instances:
return
# A copy: a callback may close its own registration.
for (callback,) in list(self.placements):
self.__call(callback, instances)
def register(self, kind: str, callback: Callable[[Any], object]) -> Monitor:
entries = self.kinds.setdefault(kind, [])
entry = (callback,)
entries.append(entry)
return Monitor(lambda: entries.remove(entry))
def notify(self, kind: str, changes: Changes[Any]) -> None:
if not changes:
return
for (callback,) in list(self.kinds.get(kind, ())):
self.__call(callback, changes)
def notify_user(self, event: pdpb2.UserEvent, rmap: ReverseMapper) -> None:
entries = self.user.get(event.name)
if not entries:
logger.debug("No monitor for user event %s", event.name)
return
data = _json_value(MessageToDict(event.data)) if event.HasField("data") else {}
objects = [
resolve(cid, rmap, False, context=f"UserEvent {event.name}")
for cid in event.objects
]
resolved = all(o is not None for o in objects)
constructed: dict[type[UserEvent], UserEvent | None] = {}
for cls, callback in list(entries):
if cls not in constructed:
try:
instance = cls(cast(dict[str, JSON], data), objects)
instance._resolved = resolved
constructed[cls] = instance
except Exception:
logger.exception("Failed to construct user event %s", event.name)
constructed[cls] = None
instance = constructed[cls]
if instance is not None:
self.__call(callback, instance)
def __call[A](self, callback: Callable[[A], object], arg: A) -> None:
try:
result = callback(arg)
except Exception:
logger.exception("Layout monitor callback %r failed", callback)
return
if inspect.isawaitable(result):
try:
task = asyncio.ensure_future(result)
except RuntimeError:
# No running loop to run it on.
if inspect.iscoroutine(result):
result.close()
logger.error(
"Layout monitor callback %r returned an awaitable "
"with no event loop to run it",
callback,
)
return
self.__tasks.add(task)
task.add_done_callback(self.__done)
def __done(self, task: asyncio.Future[Any]) -> None:
self.__tasks.discard(task)
if not task.cancelled() and task.exception() is not None:
logger.error("Layout monitor callback failed", exc_info=task.exception())
[docs]
def user_event_proto(event: UserEvent, rmap: ReverseMapper) -> pdpb2.UserEvent:
"""The wire form of ``event``.
Raises:
TypeError: if ``data`` is not JSON-compatible.
ValueError: if an object has no runtime ID.
"""
data = event.data
_check_json(dict(data), "data")
message = pdpb2.UserEvent(name=user_event_name(type(event)))
message.data.update(data)
for i, obj in enumerate(event.objects):
key = rmap.stable_id_of(obj) if obj is not None else None
if key is None:
raise ValueError(f"the runtime has no ID for objects[{i}]: {obj!r}")
message.objects.append(composite_id(key))
return message
def _check_json(value: object, path: str) -> None:
if value is None or isinstance(value, str | bool | int):
return
if isinstance(value, float):
if not math.isfinite(value):
raise TypeError(f"{path} is not a finite number: {value!r}")
return
if isinstance(value, list | tuple):
for i, v in enumerate(cast(Sequence[object], value)):
_check_json(v, f"{path}[{i}]")
return
if isinstance(value, dict):
for k, v in cast(dict[object, object], value).items():
if not isinstance(k, str):
raise TypeError(f"{path} has a non-string key: {k!r}")
_check_json(v, f"{path}[{k!r}]")
return
raise TypeError(f"{path} is not JSON-compatible: {type(value).__name__}")
def _json_value(value: Any) -> JSON:
# Struct numbers are doubles; whole ones are made ints again, so they can
# index the objects (the UI's JS does not tell 1 from 1.0 anyway).
if isinstance(value, float) and value.is_integer():
return int(value)
if isinstance(value, list):
return [_json_value(v) for v in cast(list[Any], value)]
if isinstance(value, dict):
return {k: _json_value(v) for k, v in cast(dict[str, Any], value).items()}
return value
