How to write a service¶

To have a session launch a service of your own, you write the service, declare it and point a device at it. The tabs show a caproto IOC served over Channel Access and a fastcs controller served over PVAccess, and the rest of the page holds for both. Services explains what a service is and how a session handles it.

Prerequisites¶

redsun depends on ophyd-async but not on any control-system protocol, so you install the packages your service needs, plus the ophyd-async extra for the device side:

uv add caproto "ophyd-async[ca]"

uv add "fastcs[epicspva]" "ophyd-async[pva]"

Write the service¶

A service that redsun launches does three things besides serving its process variables: it prints a line when it is ready, stops when the session asks, and listens only on the local machine. Step through what the session and the service tell each other:

# mylab/iocs/camera.py
from caproto.server import PVGroup, ioc_arg_parser, pvproperty, run

from redsun.services import stop_on_request


class Camera(PVGroup):
    exposure = pvproperty(value=0.1, name="Exposure")


if __name__ == "__main__":
    options, run_options = ioc_arg_parser(default_prefix="CAM:", desc="camera")
    stop_on_request()
    run(Camera(**options).pvdb, **{**run_options, "interfaces": ["127.0.0.1"]})

class Stage(Controller):
    def __init__(self) -> None:
        super().__init__()
        self.position = AttrRW(Float(units="mm"))

async def serve(prefix: str) -> None:
    controller = Stage()
    controller.set_path([prefix])
    served = FastCS(controller, [EpicsPVATransport()])
    serving = asyncio.ensure_future(served.serve(interactive=False))
    announcing = asyncio.ensure_future(ready_when_reachable(f"{prefix}:PVI"))
    stopping = asyncio.ensure_future(wait_for_stop())
    await asyncio.wait({serving, stopping}, return_when=asyncio.FIRST_COMPLETED)
    for task in (serving, announcing, stopping):
        task.cancel()
    await asyncio.gather(serving, announcing, stopping, return_exceptions=True)
    if not serving.cancelled():
        serving.result()


if __name__ == "__main__":
    configure_logging()
    me = identity()
    prefix = me.prefix if me else "STAGE:"
    try:
        asyncio.run(serve(prefix.rstrip(":")))
    except KeyboardInterrupt:
        pass

Since the session asks a service to stop by closing its standard input, a service that blocks in a call of its own, as caproto's run does, calls stop_on_request. A service built on asyncio awaits wait_for_stop instead.

Under channel-access, a launched service listens on a port the session chooses when its process starts, and no other program is told which one. A Channel Access client run from another terminal, such as caget, doesn't find it. Under pv-access, another program on the machine reaches it with EPICS_PVA_ADDR_LIST=127.0.0.1. If other machines must reach the service, run it on its own, on the ports you set, and let the session attach to it.

Point a device at it¶

from ophyd_async.core import StandardReadable
from ophyd_async.epics.core import epics_signal_rw


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

class MyStage(Device):
    position: SignalRW[float]

    def __init__(self, prefix: str, name: str = "") -> None:
        super().__init__(name=name, connector=fastcs_connector(prefix, self))

Declare it¶

Declare the service in the session class, beside the device that talks to it:

from typing import Annotated

from redsun import AsDevice, AsService, Attach, Declare, Launch
from redsun.qt import QtSession


class MyApp(QtSession):
    camera_ioc: Annotated[
        AsService,
        Launch(
            "mylab.iocs.camera",
            ready="Server startup complete.",
            prefix="CAM:",
            args={"prefix": "CAM:"},
            stop_timeout=30,
        ),
    ]
    beamline: Annotated[AsService, Attach("BL01:", address="10.0.0.5")]
    camera: Annotated[AsDevice[MyCamera], Declare(service="camera_ioc")]

class MyApp(QtSession):
    config: ClassVar[Mapping[str, Any]] = {"services": {"transport": "pv-access"}}

    stage_service: Annotated[
        AsService, Launch("stage_fastcs", ready="stage ready", prefix="STAGE:")
    ]
    stage: Annotated[AsDevice[MyStage], Declare(service="stage_service")]

prefix is what the devices of the service receive. args is what the process is started with, which is how the caproto IOC learns the same prefix. The fastcs service needs no prefix in args, because it reads the prefix with identity.

args is a list, or a mapping of options. In a mapping, -- goes before each name, True passes the option alone, False leaves it out (true and false in a session file), and a list passes each item after the option. For example, args={"prefix": "CAM:", "simulate": True, "debug": False} starts the module with --prefix CAM: --simulate. The session file reference has the table.

ready is text the session waits for in the output of the service. The first line containing it marks the service ready, so pick text an error message would not contain. The session waits 15 seconds for it, so a service that takes longer to start gives startup_timeout in seconds.

A service that prints no line of its own once it serves calls ready, which prints the text the declaration gives, so you write the text in one place. A PVAccess service can call ready_when_reachable with one of its own process variables instead, and it prints the text once that variable answers.

Launch declares a service the session starts and stops. Attach declares one already running elsewhere: the session starts and stops nothing, and its devices only get the service's prefix. They find it by searching the network and through the address list of your environment. When the search doesn't reach it, on another subnet for instance, give its address, and the session adds it to the address list of its transport before any device connects. See Environment variables.

The device gets the service's prefix as its prefix argument, so the session refuses a device that also gives a prefix of its own. A device whose service didn't start is left out.

Check that it starts¶

Run the session from the folder that holds the module of the service, since that is where the session looks for it. For the fastcs example, the session logs:

Service 'stage_service' started
Services started: 1/1

Serve several devices from one service¶

Every device naming a service gets the same prefix, so the devices tell their process variables apart by the rest of the name. A constructor keyword can carry it:

from ophyd_async.core import StandardReadable
from ophyd_async.epics.core import epics_signal_rw


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


class MyApp(QtSession):
    stage_ioc: Annotated[
        AsService,
        Launch("mylab.iocs.stage", ready="Server startup complete.", prefix="ST:"),
    ]
    x: Annotated[AsDevice[Axis], Declare(service="stage_ioc", axis="X")]
    y: Annotated[AsDevice[Axis], Declare(service="stage_ioc", axis="Y")]

x reads ST:X:Position, and y reads ST:Y:Position.

A package can define the declaration once, as a type alias, for every session to reuse:

CameraIoc: TypeAlias = Annotated[
    AsService, Launch("mylab.iocs.camera", ready="Server startup complete.")
]


class MyApp(QtSession):
    camera_ioc: CameraIoc

A plugin can also list the service in its manifest, which lets a session file name it without Python:

# mylab/redsun.yaml
services:
  camera-ioc:
    module: mylab.iocs.camera
    ready: "Server startup complete."

# session.yaml
services:
  camera_ioc:
    plugin_name: mylab
    plugin_id: camera-ioc
    prefix: "CAM:"
  beamline:
    prefix: "BL01:"

devices:
  camera:
    plugin_name: mylab
    plugin_id: camera
    service: camera_ioc
  microscope:
    plugin_name: mylab
    plugin_id: microscope
    services:
      stage_prefix: beamline
      camera_prefix: camera_ioc

beamline names no module, so the session attaches to it. microscope takes two prefixes, so services names the service for each of its arguments.

Name the transport¶

All the services of a session speak one transport, channel-access unless the configuration says otherwise:

services:
  transport: pv-access

In a session class, put it in config:

class MyApp(QtSession):
    config: ClassVar[dict[str, Any]] = {"services": {"transport": "pv-access"}}

An unknown transport, or two sources naming different ones, raises when the configuration is read. The transport holds for every device of the session, so a device can't speak both. Services says what a session does for each transport.

identity gives a launched process the name and the prefix its session declared for it. A module that several sessions launch, each with its own prefix, reads them there and needs no args to pass them. identity returns None when no session launched the process:

from redsun.services import identity

me = identity()
prefix = me.prefix if me else ""

React when it exits¶

A launched service that exits by itself sends sig_exited with its name and exit code. The session holds each service under its name, so wire can connect the signal to a slot:

from collections.abc import Iterator

from redsun import AsPresenter, Link, slot
from redsun.log import Loggable


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

    @slot(thread="main")
    def on_service_exited(self, name: str, code: int) -> None:
        self.logger.warning(f"{name} exited with code {code}")


class MyApp(QtSession):
    camera_ioc: CameraIoc
    presenter: AsPresenter[CameraPresenter]

    def wire(self) -> Iterator[Link]:
        yield self.camera_ioc.sig_exited, self.presenter.on_service_exited

Without thread="main", the slot runs on the thread that reads the service's output.

Log from it¶

The service's output is logged under redsun.service.<name>, and written to a log file of its own. A service calling configure_logging at startup logs at the level the session records at, and each record keeps its level instead of arriving as a DEBUG line; see Log from a service.

Quiet the type checker for fastcs¶

fastcs ships no py.typed file, so a type checker can't see its types. mypy reports the import:

Skipping analyzing "fastcs.controllers": module is installed, but missing
library stubs or py.typed marker  [import-untyped]

and, in strict mode, the class written on top of it:

Class cannot subclass "Controller" (has type "Any")  [misc]

To quiet both, add this to the mypy configuration in pyproject.toml, with the name of your module in the second entry:

[[tool.mypy.overrides]]
module = ["fastcs.*"]
ignore_missing_imports = true

[[tool.mypy.overrides]]
module = ["stage_fastcs"]
disable_error_code = ["misc"]

The fastcs example in full¶

"""The `fastcs` service of the guide "How to write a service"."""

from __future__ import annotations

import asyncio

from fastcs.attributes import AttrRW
from fastcs.control_system import FastCS
from fastcs.controllers import Controller
from fastcs.datatypes import Float
from fastcs.transports.epics.pva.transport import EpicsPVATransport

from redsun.services import (
    configure_logging,
    identity,
    ready_when_reachable,
    wait_for_stop,
)


class Stage(Controller):
    def __init__(self) -> None:
        super().__init__()
        self.position = AttrRW(Float(units="mm"))


async def serve(prefix: str) -> None:
    controller = Stage()
    controller.set_path([prefix])
    served = FastCS(controller, [EpicsPVATransport()])
    serving = asyncio.ensure_future(served.serve(interactive=False))
    announcing = asyncio.ensure_future(ready_when_reachable(f"{prefix}:PVI"))
    stopping = asyncio.ensure_future(wait_for_stop())
    await asyncio.wait({serving, stopping}, return_when=asyncio.FIRST_COMPLETED)
    for task in (serving, announcing, stopping):
        task.cancel()
    await asyncio.gather(serving, announcing, stopping, return_exceptions=True)
    if not serving.cancelled():
        serving.result()


if __name__ == "__main__":
    configure_logging()
    me = identity()
    prefix = me.prefix if me else "STAGE:"
    try:
        asyncio.run(serve(prefix.rstrip(":")))
    except KeyboardInterrupt:
        pass

"""The session of the `fastcs` example in the guide "How to write a service"."""

from __future__ import annotations

from collections.abc import Iterator, Mapping  # noqa: TC003
from typing import Annotated, Any, ClassVar, Protocol

from bluesky.protocols import Reading  # noqa: TC002
from ophyd_async.core import Device, SignalRW
from ophyd_async.fastcs.core import fastcs_connector
from psygnal import Signal
from qtpy.QtWidgets import QFormLayout, QLabel, QPushButton, QWidget

from redsun import (
    AsDevice,
    AsPresenter,
    AsService,
    AsView,
    Declare,
    DevicesOf,
    Launch,
    Link,
    Placement,
    slot,
)
from redsun.qt import Dock, QtSession


class MyStage(Device):
    position: SignalRW[float]

    def __init__(self, prefix: str, name: str = "") -> None:
        super().__init__(name=name, connector=fastcs_connector(prefix, self))


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


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("left")
    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 signal, entry in reading.items():
            stage = signal.removesuffix("-position")
            if stage not in self.labels:
                self.add_row(stage)
            self.labels[stage].setText(f"position: {entry['value']}")


class MyApp(QtSession):
    config: ClassVar[Mapping[str, Any]] = {"services": {"transport": "pv-access"}}

    stage_service: Annotated[
        AsService, Launch("stage_fastcs", ready="stage ready", prefix="STAGE:")
    ]
    stage: Annotated[AsDevice[MyStage], Declare(service="stage_service")]
    stage_ctrl: AsPresenter[StagePresenter]
    stage_view: AsView[StageView]

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


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