How to wire components together¶

Components don't connect themselves. A presenter declares signals, a view declares signals and slots, and the session says which signal reaches which slot.

You can write the connections in the session class or in the session file, and each example below shows both. Picking a tab switches every tab on the site to the same form.

Mark a method as a slot¶

Decorate a method with slot to let the session connect to it. Only marked methods can be connected. Once a method is marked, other code relies on its name and arguments, so treat them as public.

from typing import Any

from bluesky.protocols import Reading
from qtpy.QtWidgets import QWidget

from redsun import slot


class ImageView(QWidget):
    def __init__(self, name: str, parent: QWidget) -> None:
        super().__init__(parent)
        self.name = name
        self.images: dict[str, Any] = {}

    @slot
    def update_layers(self, readings: dict[str, Reading[Any]]) -> None:
        for key, reading in readings.items():
            self.images[key] = reading["value"]

slot takes three options:

    @slot(name="frames", thread="current")
    def update_layers(self, readings: dict[str, Reading[Any]]) -> None: ...

• name is the port name a session file uses. It defaults to the method name without leading underscores, so you can rename the method without breaking a file.
• thread chooses the thread the slot runs on: "main" for the main thread, "current" for the thread that made the connection, which is the one the session was built on, or a threading.Thread. Left out, the session chooses, as Choose the thread a slot runs on describes.
• signal names the signals a pairing connects to the slot.

Signals need no marker: every public Signal attribute is a port:

from psygnal import Signal


class DetectorPresenter:
    sig_new_data = Signal(dict)

    def __init__(self, name: str) -> None:
        self.name = name

    @slot
    def configure(self, settings: dict[str, Any]) -> None: ...

The examples below make two links between this presenter, the view above and a DetectorView that sends sig_property_changed:

Declare the connections¶

from collections.abc import Iterator

from redsun import AsPresenter, AsView, Link
from redsun.qt import QtSession


class MyApp(QtSession):
    det_ctrl: AsPresenter[DetectorPresenter]
    img_widget: AsView[ImageView]
    det_widget: AsView[DetectorView]

    def wire(self) -> Iterator[Link]:
        yield self.det_ctrl.sig_new_data, self.img_widget.update_layers
        yield self.det_widget.sig_property_changed, self.det_ctrl.configure

session: my-lab
frontend: qt

presenters:
  det_ctrl:
    plugin_name: my-plugin
    plugin_id: detector

views:
  img_widget:
    plugin_name: my-plugin
    plugin_id: image
  det_widget:
    plugin_name: my-plugin
    plugin_id: detector-view

wiring:
  det_ctrl.sig_new_data: img_widget.update_layers
  det_widget.sig_property_changed: det_ctrl.configure

A session may use both, and wire runs first, then the wiring section. You can wire the path provider too, as path_provider.

class StageView(QWidget):
    sig_nudge = Signal()

    def setup(self, ctrl: StagePresenter) -> None:
        self.ctrl = ctrl

    def on_click(self) -> None:
        self.ctrl.nudge()
        self.sig_nudge.emit()


class MyApp(QtSession):
    stage_ctrl: AsPresenter[StagePresenter]
    stage_view: AsView[StageView]

    def wire(self) -> Iterator[Link]:
        yield self.stage_view.sig_nudge, self.stage_ctrl.nudge

'stage_view' holds 'stage_ctrl' and is also connected to it; a bundle reaches a component one way, by calling it or by a signal

Connect a coroutine¶

An async def method is a slot like any other:

class MotorPresenter:
    @slot
    async def move(self, motor: str, position: float) -> None:
        await self.devices[motor].set(position)

The session delivers it differently from a plain method. Step through the two cases:

If you need the emitter to wait, connect a plain method that calls run_coro(...) from redsun.aio instead. To stop a task such a slot started, for instance from a stop button, use cancel_task(task) from redsun.aio. Called from another thread, it waits for the task to pause rather than cancelling it while it runs.

Every session installs the async backend psygnal needs for coroutine slots when it is built, and removes it at shutdown.

Address a signal group¶

A component keeping its signals in a SignalGroup offers each member as a port, under the member's name:

class FrameSignals(SignalGroup, strict=True):
    median = Signal(object)
    filtered = Signal(object)


class MedianPresenter:
    def __init__(self, name: str) -> None:
        self.name = name
        self.frames = FrameSignals(instance=self)

def wire(self) -> Iterator[Link]:
    yield self.median_ctrl.frames.median, self.img_widget.update_layers

wiring:
  median_ctrl.median: img_widget.update_layers

Pass instance=self when making the group, or the session cannot tell which component owns the signal.

Choose the thread a slot runs on¶

A slot runs on the thread that emitted, unless something says otherwise. The session asks three places in turn and takes the first answer. Point at one to read what it covers:

The frontend answers through Frontend.thread_of. A slot that is a coroutine function runs on the session's event loop whatever the frontend.

Observe a device signal¶

Device signals come from ophyd-async, not psygnal. wire tells the two apart by the signal's type, so a link to a device signal is yielded the same way as one to a psygnal signal:

class MyApp(QtSession):
    detector: AsDevice[MyDetector]
    temperature_widget: AsView[TemperatureView]

    def wire(self) -> Iterator[Link]:
        yield self.detector.temperature, self.temperature_widget.update_temperature

The slot receives each reading. The subscription is released at shutdown.

Inspect what is connected¶

connections lists every link the session made, with the thread a slot runs on when it isn't the one that emitted:

for link in app.connections:
    print(link)

det_ctrl.sig_new_data -> img_widget.update_layers  [thread=main]
det_widget.sig_property_changed -> det_ctrl.configure
detector.temperature -> temperature_widget.update_temperature  [thread=main]

A link from a device signal is listed with the others, under the device and the name of the signal within it.

A component may hold an object with signals or slots of its own as an attribute. The session records that object under the name of the component. It records a link yielded as self.det_widget.sig_action_request, self.det_ctrl.actions.request as:

det_widget.sig_action_request -> det_ctrl.request

ports lists what one component offers:

>>> ports(view).slots
{'update_layers': <bound method ImageView.update_layers ...>}

Find what is not connected¶

A wrong port name fails the build, but a forgotten connection fails nowhere, because the signal just reaches nothing. unconnected finds it:

print(app.unconnected)

det_ctrl.sig_error -> nothing
nothing -> img_widget.clear

An entry is not always a mistake: a component may offer more than a session uses.

Read a failure¶

A wrong connection stops the build and names both ends. The exception is a connection to a component that failed to build: the session logs it, skips it and still makes the other connections.

Not connecting mover.sig_moved -> panel.on_moved: component 'panel' was not built

Pair two components¶

Two components whose slots name the signals of the other connect with one line, pairs: - [a, b] in the file or yield from links_between(self.a, self.b) in wire. The pairs section runs after wiring. Offer a pairing shows how a component offers one.