Wire components together¶

Components do not connect themselves. A presenter declares signals, a view declares signals and connectable methods, and the application says which signal reaches which method.

An application is declared in one of two ways, and every connection below is shown in both. Picking a tab switches every other tab on this page, and on any other page of this documentation, to the same form.

Mark a method as connectable¶

Decorate it with slot. Only a marked method can be connected, so marking one makes it public API: its name and its signature are what other components are connected against.

from redsun.view import View
from redsun.virtual import slot


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

Name a slot as you would any public method. slot accepts two options:

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

• name is the port name used in a configuration file. It defaults to the method name without leading underscores, and exists so the method can be renamed without breaking a configuration.
• thread overrides the thread the slot is delivered on.

Signals need no marker. Every public Signal attribute is already a port.

Declare the connections¶

class MyApp(QtAppContainer, config="session.yaml"):
    det_ctrl = declare_presenter(DetectorPresenter)
    img_widget = declare_view(ImageView)
    det_widget = declare_view(DetectorView)

    def wire(self) -> None:
        self.connect(self.det_ctrl.sig_new_data, self.img_widget.update_layers)
        self.connect(self.det_widget.sig_property_changed, self.det_ctrl.configure)

schema_version: 1.0
frontend: pyqt
session: my-session

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

wiring:
  - from: det_ctrl.sig_new_data
    to: img_widget.update_layers
  - from: det_widget.sig_property_changed
    to: det_ctrl.configure

Fan-in is another line, not another concept: a second producer of frames reaches the same viewer by adding one connection.

Both forms end in the same call, so a container may use both. wire runs first, then the wiring section of its configuration file.

Connect a coroutine¶

An async def method is a slot like any other. Mark it and connect it; nothing else changes.

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

    def wire(self) -> None:
        self.connect(self.motor_widget.sig_motor_move, self.motor_ctrl.move)

wiring:
  - from: motor_widget.sig_motor_move
    to: motor_ctrl.move

Dispatch differs from a plain method in three ways:

• the coroutine runs on redsun's shared event loop, not on the thread that emitted;
• the emitter does not wait for it. The emission returns as soon as the coroutine is scheduled;
• an exception inside it is logged on the redsun logger instead of propagating back to the emitter, and later emissions keep being delivered.

If the last two matter, keep a sync method that owns the call and connect that instead:

    @slot
    def move(self, motor: str, position: float) -> None:
        run_coro(self.move_async(motor, position))

Address a signal group¶

A component whose signals live in a SignalGroup exposes each member as a port, under the member name. The group attribute itself is not a port.

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


class MedianPresenter(Presenter):
    def __init__(self, name: str, devices: Mapping[str, Device], /) -> None:
        super().__init__(name, devices)
        self.frames = FrameSignals(instance=self)

def wire(self) -> None:
    self.connect(self.median_ctrl.frames.median, self.img_widget.update_layers)
    self.connect(self.median_ctrl.frames.filtered, self.img_widget.update_layers)

wiring:
  - from: median_ctrl.median
    to: img_widget.update_layers
  - from: median_ctrl.filtered
    to: img_widget.update_layers

Group members and plain signals share one port namespace, so a member named after an existing public signal on the same class raises WiringError when the ports are read.

Choose the thread a slot runs on¶

Thread affinity belongs to the component, not to the connection. A class declares it once:

from typing import ClassVar

from redsun.virtual import SlotThread


class MyView(View):
    __redsun_slot_thread__: ClassVar[SlotThread] = "main"

Every slot on that class is then delivered on the main thread. @slot(thread=...) overrides it for one method, and connect(..., thread=...) overrides both.

QtView already declares "main", so a Qt widget's slots need nothing.

Observe a device signal¶

Device signals are ophyd-async, not psygnal, so connect does not take them. subscribe does, and gives the same guarantees:

class StorageView(QtView):
    @slot
    def update_base_dir(self, reading: dict[str, Reading[str]]) -> None:
        self._edit.setText(next(iter(reading.values()))["value"])


class MyApp(QtAppContainer):
    def wire(self) -> None:
        provider = self.virtual_container.require(PATH_PROVIDER)
        self.virtual_container.subscribe(
            provider.signals.base_dir, self.storage_widget.update_base_dir
        )

The slot still has to be marked, the thread affinity still comes from the component, and the subscription is released at shutdown. That last one matters more here than for signals: ophyd-async releases a subscription by identity, so whoever subscribed has to keep the exact callback object to undo it.

Inspect what is connected¶

for link in app.virtual_container.connections:
    print(link)
for record in app.virtual_container.subscriptions:
    print(record)

det_ctrl.sig_new_data -> img_widget.update_layers  [thread=main]
det_widget.sig_property_changed -> det_ctrl.configure
base_dir ~> storage_widget.update_base_dir  [thread=main]

-> is a signal connection, ~> a device subscription.

ports answers the other half, what a component offers:

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

Both are recorded, so AppContainer.shutdown releases them.

Find what is not connected¶

A wrong port name fails at build. A connection you forgot to write fails nowhere: a signal simply emits into nothing. unconnected is what finds it, by subtracting the recorded links from what every built component offers.

report = app.virtual_container.unconnected
if report:
    print(report)

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

It is a plain Unconnected record, so a script can assert on it instead of reading it:

assert not app.virtual_container.unconnected.slots

Not every entry is a defect. A component may legitimately offer more than one application uses; the report says what is unused, not what is wrong.

Read a failure¶

Every way of getting a connection wrong fails at build, naming both ends.