How to connect a device on demand¶

You don't have to connect a device when the session builds it. You can build it unconnected and have a component connect it when the user asks. Connecting explains what the build does with the other devices.

Prerequisites¶

You need a session that declares the device, and a component that uses it. The blocks below are parts of one script, and the whole script is at the end.

Declare the device unconnected¶

Give the declaration autoconnect=False:

motor: Annotated[AsDevice[MyMotor], Declare(prefix="MOTOR:", autoconnect=False)]

MyMotor is in the whole script at the end. It names no service, so you give its prefix here.

In a session file, write it like this:

devices:
  motor:
    plugin_name: mylab
    plugin_id: motor
    autoconnect: false

The build creates the device and adds it to devices, but doesn't connect it. Session file lists what autoconnect accepts.

Connect it from a component¶

Ask for the devices and connect in an async slot:

class MyController:
    sig_connected = Signal(str)
    sig_not_connected = Signal(str, str)

    def __init__(
        self, name: str, *, devices: DeviceMapping, config: SessionConfig
    ) -> None:
        self.name = name
        self.devices = devices
        self.mock = config.mock

    @slot
    async def connect_motor(self) -> None:
        motor = self.devices["motor"]
        try:
            await motor.connect(mock=self.mock, timeout=5)
        except NotConnectedError as e:
            self.sig_not_connected.emit(motor.name, str(e))
            return
        self.sig_connected.emit(motor.name)

The controller reads mock from SessionConfig, so in a mocked session it connects the device to a simulated backend, as the build does with the others.

Ask for it from a view¶

Give the view a signal to ask with and slots for the answers:

class MyView(QWidget):
    placement: Placement = Dock("left")
    sig_connect_clicked = Signal()

    def __init__(self, name: str, parent: QWidget) -> None:
        super().__init__(parent)
        self.name = name
        self.button = QPushButton("Connect")
        self.button.clicked.connect(lambda: self.sig_connect_clicked.emit())
        QVBoxLayout(self).addWidget(self.button)

    @slot
    def on_connected(self, device: str) -> None:
        self.button.setEnabled(False)

    @slot
    def on_not_connected(self, device: str, reason: str) -> None:
        self.button.setToolTip(reason)

Link them in the session:

class MyApp(QtSession):
    motor: Annotated[AsDevice[MyMotor], Declare(prefix="MOTOR:", autoconnect=False)]
    ctrl: AsPresenter[MyController]
    panel: AsView[MyView]

    def wire(self) -> Iterator[Link]:
        yield self.panel.sig_connect_clicked, self.ctrl.connect_motor
        yield self.ctrl.sig_connected, self.panel.on_connected
        yield self.ctrl.sig_not_connected, self.panel.on_not_connected


if __name__ == "__main__":
    MyApp().run()

Shut it down unconnected¶

The session calls the device's shutdown when it ends, whether or not a component connected the device.

shutdown can run on a device that never connected

A shutdown that writes to the hardware has to cope with a device that never connected. Make it check that the device connected first.

The example in full¶

The whole script

"""The session of the guide "How to connect a device on demand"."""

from __future__ import annotations

from collections.abc import Iterator  # noqa: TC003
from typing import Annotated

from ophyd_async.core import NotConnectedError, StandardReadable
from ophyd_async.epics.core import epics_signal_rw
from psygnal import Signal
from qtpy.QtWidgets import QPushButton, QVBoxLayout, QWidget

from redsun import (
    AsDevice,
    AsPresenter,
    AsView,
    Declare,
    DeviceMapping,
    Link,
    Placement,
    SessionConfig,
    slot,
)
from redsun.qt import Dock, QtSession


class MyMotor(StandardReadable):
    def __init__(self, prefix: str, name: str = "") -> None:
        with self.add_children_as_readables():
            self.position = epics_signal_rw(float, f"{prefix}Position")
        super().__init__(name=name)


class MyController:
    sig_connected = Signal(str)
    sig_not_connected = Signal(str, str)

    def __init__(
        self, name: str, *, devices: DeviceMapping, config: SessionConfig
    ) -> None:
        self.name = name
        self.devices = devices
        self.mock = config.mock

    @slot
    async def connect_motor(self) -> None:
        motor = self.devices["motor"]
        try:
            await motor.connect(mock=self.mock, timeout=5)
        except NotConnectedError as e:
            self.sig_not_connected.emit(motor.name, str(e))
            return
        self.sig_connected.emit(motor.name)


class MyView(QWidget):
    placement: Placement = Dock("left")
    sig_connect_clicked = Signal()

    def __init__(self, name: str, parent: QWidget) -> None:
        super().__init__(parent)
        self.name = name
        self.button = QPushButton("Connect")
        self.button.clicked.connect(lambda: self.sig_connect_clicked.emit())
        QVBoxLayout(self).addWidget(self.button)

    @slot
    def on_connected(self, device: str) -> None:
        self.button.setEnabled(False)

    @slot
    def on_not_connected(self, device: str, reason: str) -> None:
        self.button.setToolTip(reason)


class MyApp(QtSession):
    motor: Annotated[AsDevice[MyMotor], Declare(prefix="MOTOR:", autoconnect=False)]
    ctrl: AsPresenter[MyController]
    panel: AsView[MyView]

    def wire(self) -> Iterator[Link]:
        yield self.panel.sig_connect_clicked, self.ctrl.connect_motor
        yield self.ctrl.sig_connected, self.panel.on_connected
        yield self.ctrl.sig_not_connected, self.panel.on_not_connected


if __name__ == "__main__":
    MyApp().run()