Skip to content

Putting a device behind a service

In this tutorial you add a third stage, which lives in a program of its own. The session starts that program, and stops it when you close the window. It continues from Arranging the window.

So far your stages have kept their position in memory. Real hardware is reached through a service, a separate program that owns the hardware and offers its values over the network. Apart from one new link, nothing you wrote before changes. The new stage gets a row in the view of the stages, and an entry in the plan widgets of walk and scan.

Devices without a service

A device can also reach its hardware on its own, but for now a service is the preferred way, and Services explains why. This tutorial uses one.

Before you start

What you need

caproto, to write the service, and ophyd-async[ca], so the device can talk to it. Neither comes with redsun:

uv add caproto "ophyd-async[ca]"

Open first_session.py, and add these imports below the ones it has:

from typing import Annotated

from ophyd_async.epics.core import EpicsDevice, PvSuffix

from redsun import AsService, Declare, Launch

1. Write the service

Make a second file in the project folder, beside the first, called stage_ioc.py. Start it with these imports:

from caproto.server import PVGroup, ioc_arg_parser, pvproperty, run

from redsun.services import identity, stop_on_request

Then add the stage:

class Stage(PVGroup):
    position = pvproperty(value=0.0, name="Position")

That's the whole stage: one value, named Position. caproto serves it over Channel Access, one of the two protocols of EPICS. A program of this kind is called an IOC, and a value it serves is called a process variable.

This stage only keeps a number. The service of a real stage would talk to its controller in the same place, since caproto can call a function each time Position is set.

A service that a session starts has to do two more things: stop when the session asks, and listen on this machine only. Add both below the stage:

if __name__ == "__main__":
    me = identity()
    prefix = me.prefix if me else "STAGE:"
    options, run_options = ioc_arg_parser(default_prefix=prefix, desc="stage")
    stop_on_request()
    run(Stage(**options).pvdb, **{**run_options, "interfaces": ["127.0.0.1"]})

identity returns the name and prefix the session gave the service. The prefix is the start of the name of each process variable. When the service runs alone, identity returns None, and the service falls back on STAGE:. stop_on_request stops the service as Ctrl+C would once the session asks, and does nothing when the service runs alone.

Try the service on its own:

uv run stage_ioc.py

It prints Server startup complete. and waits. Stop it with Ctrl+C, and don't leave it running, because the session starts its own. The service listens on this machine only, so to read it with caget from another terminal, set EPICS_CA_ADDR_LIST=127.0.0.1 there first.

On Windows, the service may print a few lines that end with OSError: [WinError 995] as it stops. It has stopped all the same.

2. Write the device

In first_session.py, add a device below FastStage. Its position is the process variable of the service:

class RemoteStage(EpicsDevice, StandardMovable[float]):
    position: Annotated[SignalRW[float], PvSuffix("Position")]

    @cached_property
    def movable_logic(self) -> MovableLogic[float]:
        return MovableLogic(setpoint=self.position, readback=self.position)

The device names only the end of the process variable, Position. The beginning, the prefix, comes from the service you declare it with. Like the other stages, it's a StandardMovable, and its one process variable is both its setpoint and its readback.

3. Declare the service

Add the highlighted lines to the session. They declare the service and the stage, and add the link that sends the stage's position to the view:

class FirstSession(QtSession):
    config = "session.yaml"
    stage_ioc: Annotated[
        AsService,
        Launch("stage_ioc", ready="Server startup complete.", prefix="STAGE:"),
    ]
    stage: AsDevice[MyStage]
    fast_stage: AsDevice[FastStage]
    remote_stage: Annotated[AsDevice[RemoteStage], Declare(service="stage_ioc")]
    camera: AsDevice[SimBlobDetector]
    stage_ctrl: AsPresenter[StagePresenter]
    stage_plans: AsPresenter[StagePlans]
    plan_ctrl: AsPresenter[PlanPresenter]
    camera_ctrl: AsPresenter[CameraPresenter]
    scan_plans: AsPresenter[ScanPlans]
    stage_view: AsView[StageView]
    plan_view: AsView[PlanView]
    image_view: AsView[ImageView]

    def wire(self) -> Iterator[Link]:
        yield self.stage_view.sig_nudge, self.stage_ctrl.nudge
        yield self.stage.position, self.stage_view.show_reading
        yield self.fast_stage.position, self.stage_view.show_reading
        yield self.remote_stage.position, self.stage_view.show_reading
        yield self.plan_view.sig_run, self.plan_ctrl.run
        yield self.plan_ctrl.sig_finished, self.plan_view.on_finished
        yield self.plan_ctrl.sig_started, self.path_provider.set_plan
        yield self.plan_ctrl.sig_finished, self.path_provider.reset_plan
        yield self.plan_ctrl.sig_finished, self.camera_ctrl.show_last
        yield self.camera_ctrl.sig_frame, self.image_view.show_frame


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

AsService declares a service, under the name on the left of its line. Launch says how to start it: the module to run, which is the name of the file, then the line the service prints when it's ready, and its prefix. Declare ties the stage to the service, by the name the session gave the service.

You write the prefix only once, here, and the session hands it to both the service and the device.

4. Run it

uv run first_session.py

The session starts the service, waits for it to be ready, and then builds the rest, as the terminal shows:

Service 'stage_ioc' started
Services started: 1/1
Session built: 4/4 devices, 5/5 presenters, 3/3 views

The window is the one from the last tutorial, with one more stage: the view of the stages has a row for remote_stage, and the plan widgets of walk and scan list it. Press its button, or choose it in a list of stages and press Run. Its position counts up, kept by a different program from the one drawing the window.

The window of the last tutorial, with a third row in the view of the stages, for remote_stage

Two messages you can ignore

Channel Access may print one or both of these messages, which look like errors:

Failed to start executable - "caRepeater".

CA.Client.Exception...
    Warning: "Virtual circuit disconnect"

The first says that a helper program of EPICS is missing. The helper shares the announcements of servers between the programs of one machine, and without it a program takes longer to notice that a server has started again. The second says that the connection to the service closed while the session was still using it, because the service went away. Closing the window doesn't print it, since the session closes its connections before it stops the service.

One message you should not ignore

CA.Client.Exception...
    Warning: "Identical process variable names on multiple servers"

This message means another program the machine can reach also serves STAGE:Position, and the stage may be talking to that one instead. It could be another IOC on the network or, when your own EPICS_CA_ADDR_LIST names 127.0.0.1, a stage_ioc.py left running in another terminal. On a network shared with others, choose a prefix nobody else uses.

5. Stop it

Close the window. The session stops the service it started:

Service 'stage_ioc' stopped with exit code 0
The whole script

The service, stage_ioc.py:

"""The service launched in the "Putting a device behind a service" tutorial."""

from __future__ import annotations

from caproto.server import PVGroup, ioc_arg_parser, pvproperty, run

from redsun.services import identity, stop_on_request



class Stage(PVGroup):
    position = pvproperty(value=0.0, name="Position")


if __name__ == "__main__":
    me = identity()
    prefix = me.prefix if me else "STAGE:"
    options, run_options = ioc_arg_parser(default_prefix=prefix, desc="stage")
    stop_on_request()
    run(Stage(**options).pvdb, **{**run_options, "interfaces": ["127.0.0.1"]})

The session, first_session.py:

"""The session built in the "Putting a device behind a service" tutorial."""

from __future__ import annotations

from collections.abc import Iterator, Mapping  # noqa: TC003
from functools import cached_property
from typing import Annotated, Any, Protocol, runtime_checkable
from urllib.parse import urlsplit
from urllib.request import url2pathname

import bluesky.plan_stubs as bps
import bluesky.plans as bp
import h5py
import numpy as np
from bluesky.protocols import Readable, Reading, Triggerable
from bluesky.utils import MsgGenerator  # noqa: TC002
from event_model import DocumentRouter, StreamResource
from ophyd_async.core import (
    MovableLogic,
    SignalRW,
    StandardMovable,
    StandardReadable,
    soft_signal_rw,
)
from ophyd_async.epics.core import EpicsDevice, PvSuffix
from ophyd_async.sim import SimBlobDetector  # noqa: TC002
from psygnal import Signal
from qtpy.QtGui import QImage, QPixmap
from qtpy.QtWidgets import (
    QComboBox,
    QFormLayout,
    QLabel,
    QPushButton,
    QStackedWidget,
    QVBoxLayout,
    QWidget,
)

from redsun import (
    AsDevice,
    AsPresenter,
    AsService,
    AsView,
    CallbackType,
    Declare,
    DeviceMapping,
    DevicesOf,
    HasPlans,
    Launch,
    Link,
    Placement,
    PlanEntry,
    slot,
)
from redsun.engine import RunEngine
from redsun.presenter.plan_spec import (
    PlanSpec,
    collect_arguments,
    create_plan_spec,
    resolve_arguments,
)
from redsun.qt import Central, Dock, QtSession
from redsun.view.qt.utils import PlanWidget, create_plan_widget


class MyStage(StandardReadable, StandardMovable[float]):
    def __init__(self, name: str = "", *, units: str = "mm") -> None:
        with self.add_children_as_readables():
            self.position = soft_signal_rw(float, units=units)
        super().__init__(name=name)

    @cached_property
    def movable_logic(self) -> MovableLogic[float]:
        return MovableLogic(setpoint=self.position, readback=self.position)


class FastStage(StandardReadable, StandardMovable[float]):
    def __init__(self, name: str = "", *, units: str = "mm") -> None:
        with self.add_children_as_readables():
            self.position = soft_signal_rw(float, initial_value=5.0, units=units)
            self.speed = soft_signal_rw(float, initial_value=10.0)
        super().__init__(name=name)

    @cached_property
    def movable_logic(self) -> MovableLogic[float]:
        return MovableLogic(setpoint=self.position, readback=self.position)


class RemoteStage(EpicsDevice, StandardMovable[float]):
    position: Annotated[SignalRW[float], PvSuffix("Position")]

    @cached_property
    def movable_logic(self) -> MovableLogic[float]:
        return MovableLogic(setpoint=self.position, readback=self.position)




@runtime_checkable
class HasPosition(Protocol):
    position: SignalRW[float]


@runtime_checkable
class Camera(Readable[Any], Triggerable, Protocol): ...


class StagePresenter:
    def __init__(
        self, name: str, *, stages: DevicesOf[HasPosition], step: float = 1.0
    ) -> None:
        self.name = name
        self.stages = stages
        self.step = step

    @slot
    async def nudge(self, stage: str) -> None:
        position = await self.stages[stage].position.get_value()
        await self.stages[stage].position.set(position + self.step)


class StageView(QWidget):
    placement: Placement = Dock("bottom")
    sig_nudge = Signal(str)

    def __init__(self, name: str, parent: QWidget) -> None:
        super().__init__(parent)
        self.name = name
        self.rows = QFormLayout(self)
        self.labels: dict[str, QLabel] = {}

    def add_row(self, stage: str) -> None:
        button = QPushButton(f"Nudge {stage}")
        button.clicked.connect(lambda: self.sig_nudge.emit(stage))
        self.labels[stage] = QLabel()
        self.rows.addRow(button, self.labels[stage])

    @slot
    def show_reading(self, reading: dict[str, Reading[float]]) -> None:
        for stage, entry in reading.items():
            if stage not in self.labels:
                self.add_row(stage)
            self.labels[stage].setText(f"position: {entry['value']}")


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

    def walk(
        self, stage: HasPosition, steps: int = 5, size: float = 1.0
    ) -> MsgGenerator[None]:
        for _ in range(steps):
            position = yield from bps.rd(stage.position)
            yield from bps.mv(stage.position, position + size)

    def plan_map(self) -> Mapping[str, PlanEntry]:
        return {"walk": {"plan": self.walk}}


class PlanPresenter:
    sig_started = Signal(str)
    sig_finished = Signal()

    def __init__(self, name: str, *, devices: DeviceMapping) -> None:
        self.name = name
        self.devices = devices
        self.engine = RunEngine()
        self.plans: dict[str, PlanEntry] = {}
        self.specs: dict[str, PlanSpec] = {}

    def setup(
        self,
        plan_sources: Mapping[str, HasPlans],
        callbacks: Mapping[str, CallbackType],
    ) -> None:
        for component in plan_sources.values():
            self.plans.update(component.plan_map())
        for plan, entry in self.plans.items():
            self.specs[plan] = create_plan_spec(entry["plan"], self.devices)
        for callback in callbacks.values():
            self.engine.subscribe(callback)

    @slot
    def run(self, plan: str, values: dict[str, Any]) -> None:
        resolved = resolve_arguments(self.specs[plan], values, self.devices)
        args, kwargs = collect_arguments(self.specs[plan], resolved)
        self.sig_started.emit(plan)
        future = self.engine(self.plans[plan]["plan"](*args, **kwargs))
        future.add_done_callback(lambda _: self.sig_finished.emit())


class PlanView(QWidget):
    placement: Placement = Dock("right")
    sig_run = Signal(str, dict)

    def __init__(self, name: str, parent: QWidget) -> None:
        super().__init__(parent)
        self.name = name
        self.chooser = QComboBox()
        self.pages = QStackedWidget()
        self.chooser.currentIndexChanged.connect(self.pages.setCurrentIndex)
        layout = QVBoxLayout(self)
        layout.addWidget(self.chooser)
        layout.addWidget(self.pages)
        self.widgets: dict[str, PlanWidget] = {}

    def setup(
        self, plan_sources: Mapping[str, HasPlans], devices: DeviceMapping
    ) -> None:
        for component in plan_sources.values():
            for entry in component.plan_map().values():
                self.add_plan(create_plan_spec(entry["plan"], devices))

    def add_plan(self, spec: PlanSpec) -> None:
        widget = create_plan_widget(
            spec, run_callback=lambda: self.ask_to_run(spec.name)
        )
        self.widgets[spec.name] = widget
        self.chooser.addItem(spec.name)
        self.pages.addWidget(widget.group_box)

    def ask_to_run(self, plan: str) -> None:
        self.setEnabled(False)
        self.sig_run.emit(plan, self.widgets[plan].parameters)

    @slot
    def on_finished(self) -> None:
        self.setEnabled(True)


class CameraPresenter(DocumentRouter):
    sig_frame = Signal(object)

    def __init__(self, name: str, *, cameras: DevicesOf[Camera]) -> None:
        super().__init__()
        self.name = name
        self.cameras = cameras
        self.written: tuple[str, str] | None = None

    def snap(self, camera: Camera, frames: int = 3) -> MsgGenerator[Any]:
        return (yield from bp.count([camera], num=frames))

    def plan_map(self) -> Mapping[str, PlanEntry]:
        return {"snap": {"plan": self.snap}}

    def stream_resource(self, doc: StreamResource) -> StreamResource:
        if doc["data_key"] in self.cameras:
            self.written = (doc["uri"], doc["parameters"]["dataset"])
        return doc

    @slot
    def show_last(self) -> None:
        if self.written is not None:
            uri, dataset = self.written
            with h5py.File(url2pathname(urlsplit(uri).path), "r") as file:
                self.sig_frame.emit(file[dataset][-1])
            self.written = None


class ImageView(QWidget):
    placement: Placement = Central()

    def __init__(self, name: str, parent: QWidget) -> None:
        super().__init__(parent)
        self.name = name
        self.image = QLabel("No image yet")
        QVBoxLayout(self).addWidget(self.image)

    @slot
    def show_frame(self, frame: object) -> None:
        values = np.asarray(frame, dtype=float)
        low, high = values.min(), values.max()
        grey = (255 * (values - low) / max(high - low, 1.0)).astype(np.uint8)
        height, width = grey.shape
        image = QImage(
            grey.tobytes(), width, height, width, QImage.Format.Format_Grayscale8
        )
        self.image.setPixmap(QPixmap.fromImage(image.copy()))


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

    def scan(
        self,
        stage: HasPosition,
        camera: Camera,
        start: float = 0.0,
        stop: float = 5.0,
        points: int = 6,
    ) -> MsgGenerator[Any]:
        return (yield from bp.scan([camera], stage.position, start, stop, points))

    def plan_map(self) -> Mapping[str, PlanEntry]:
        return {"scan": {"plan": self.scan}}


class FirstSession(QtSession):
    config = "session.yaml"
    stage_ioc: Annotated[
        AsService,
        Launch("stage_ioc", ready="Server startup complete.", prefix="STAGE:"),
    ]
    stage: AsDevice[MyStage]
    fast_stage: AsDevice[FastStage]
    remote_stage: Annotated[AsDevice[RemoteStage], Declare(service="stage_ioc")]
    camera: AsDevice[SimBlobDetector]
    stage_ctrl: AsPresenter[StagePresenter]
    stage_plans: AsPresenter[StagePlans]
    plan_ctrl: AsPresenter[PlanPresenter]
    camera_ctrl: AsPresenter[CameraPresenter]
    scan_plans: AsPresenter[ScanPlans]
    stage_view: AsView[StageView]
    plan_view: AsView[PlanView]
    image_view: AsView[ImageView]

    def wire(self) -> Iterator[Link]:
        yield self.stage_view.sig_nudge, self.stage_ctrl.nudge
        yield self.stage.position, self.stage_view.show_reading
        yield self.fast_stage.position, self.stage_view.show_reading
        yield self.remote_stage.position, self.stage_view.show_reading
        yield self.plan_view.sig_run, self.plan_ctrl.run
        yield self.plan_ctrl.sig_finished, self.plan_view.on_finished
        yield self.plan_ctrl.sig_started, self.path_provider.set_plan
        yield self.plan_ctrl.sig_finished, self.path_provider.reset_plan
        yield self.plan_ctrl.sig_finished, self.camera_ctrl.show_last
        yield self.camera_ctrl.sig_frame, self.image_view.show_frame


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

What you built

You wrote a service that serves a stage, and a session that starts it, talks to it and stops it. The application now has three stages, a camera, five presenters and three views.

Next steps