How to serve a FastCS controller over sharedbox¶

When a session and its service run on the same computer, they can exchange values through shared memory instead of a network protocol. The sharedbox transport serves a fastcs controller as boxes, and the session builds an ophyd-async device from them without naming a single field. This page shows the three pieces: the service, the device and the declaration. How to write a service covers what every launched service does, and this page builds on it.

Prerequisites¶

Install redsun with the sharedbox extra, which brings sharedbox and fastcs:

uv add "redsun[sharedbox]"

sharedbox runs on Windows x64 and Linux x86_64.

The service and the session have to run on the same computer, because a box lives in that computer's memory. A service other machines must reach needs one of the network transports.

Write the service¶

Describe the hardware as a fastcs controller. This one has an exposure the session can set, a state it can read and a command it can run:

class MyController(Controller):
    def __init__(self) -> None:
        super().__init__()
        self.exposure = AttrRW(Float(units="ms", min=0.0))
        self.state = AttrR(String(length=16), initial_value="idle")

    @command()
    async def start(self) -> None:
        await self.state.update("running")

Then serve it with serve_controller and a BoxTransport:

if __name__ == "__main__":
    configure_logging()
    prefix = None if identity() else "MYAPP:DET"
    try:
        asyncio.run(serve_controller(MyController(), BoxTransport(), prefix))
    except KeyboardInterrupt:
        pass

• serve_controller serves the controller under the prefix the session gave the process, and stops when the session asks. Run alone, the service has no session, so it passes MYAPP:DET instead.
• Once the boxes exist, it prints the ready text the declaration gives, so the session never connects a device before the service serves.
• configure_logging sends the records of fastcs to the session, as on the other transports.

BoxTransport gives the controller a box, and each of its sub-controllers a box of its own. It then publishes one more box under the prefix, the index, which lists every box and the attributes and commands in it. The session learns the layout from the index, so a controller whose attributes are read from the hardware in initialise works without any change.

Point a device at it¶

Declare the signals and commands the session needs, and give the device a BoxDeviceConnector with the prefix:

class MyDetector(Device):
    exposure: SignalRW[float]
    state: SignalR[str]
    start: TriggerableCommand

    def __init__(self, prefix: str, name: str = "") -> None:
        self.counts_stream = stream_name(prefix, "counts")
        super().__init__(name=name, connector=BoxDeviceConnector(prefix))

When the device connects, the connector reads the index and fills in every attribute, command and sub-controller the service serves, declared or not. It then checks the ones you declared, so connect fails with the name of a declared child the service doesn't serve, or one whose type differs from what the service serves. A fastcs ControllerVector, a numbered list of sub-controllers, becomes an ophyd-async DeviceVector, its numbered list of sub-devices. A fastcs command becomes a TriggerableCommand.

The values of a declared enum must be the names of the members of the fastcs enum, because the index lists only those names.

One service can also back several devices, each taking a part of it, such as a flyer that runs a scan and a detector that records it. Give each of them BoxDeviceConnector(prefix, declared_only=True), and its connector fills only the signals, commands and children the device declares. A declared child follows the same rule, while a declared DeviceVector takes every entry the service serves. A declared name the service lacks still fails connect, and two devices that declare the same attribute both read and write it.

Declare it¶

Name the transport in the session's configuration, and declare the service beside the device:

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

    detector_service: Annotated[
        AsService, Launch("box_service", ready="detector ready", prefix="MYAPP:DET")
    ]
    detector: Annotated[AsDevice[MyDetector], Declare(service="detector_service")]


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

In a session file the transport goes in the services section, as for the other transports:

services:
  transport: sharedbox
  detector_service:
    module: box_service
    ready: "detector ready"
    prefix: "MYAPP:DET"

A trailing colon makes no difference: MYAPP:DET: and MYAPP:DET name the same boxes. The prefix holds letters, digits, _ and -, in parts separated by colons.

Handle a refused write¶

A set returns once the service has put the value, and raises BoxPutError when the service refuses it, for example a value below the attribute's min. The error carries the attribute's name and the service's message:

async def try_exposure(detector: MyDetector, value: float) -> bool:
    try:
        await detector.exposure.set(value)
    except BoxPutError as error:
        print(f"{error.attribute} refused {value}: {error}")
        return False
    return True

A command that raises in the service raises BoxPutError in the session too.

Name a stream of your own¶

A service can send data the controller doesn't hold, such as counts, on a sharedbox stream it makes itself. The device gets the same prefix as the service, so it builds the stream's name with stream_name:

self.counts_stream = stream_name(prefix, "counts")

Under the prefix MYAPP:DET, that names MYAPP:DET:counts. Agree on the last part between the service and the device, which usually come from the same package. stream_name refuses the part fastcs, since the transport's own boxes use it.

Reconnect after a restart¶

Each start of the service gives its boxes new names, and the index tells a device what they are. A device connected to a service that has restarted keeps the old boxes, so connect it again:

await detector.connect(force_reconnect=True)

Signals you observe keep their subscribers across the reconnect.

What the transport refuses¶

BoxTransport refuses some controllers when it starts, and the service then exits with the reason:

• an attribute whose datatype is Table;
• an attribute or command whose name contains __, which the transport uses for its own fields;
• a controller with more attributes and commands than one box can hold, or a controller tree whose description is larger than the index can hold;
• more than one controller given to one fastcs.control_system.FastCS, the object that serves controllers over transports.

The example in full¶

"""The service of the guide "How to serve a FastCS controller over sharedbox"."""

from __future__ import annotations

import asyncio

from fastcs.attributes import AttrR, AttrRW
from fastcs.controllers import Controller
from fastcs.datatypes import Float, String
from fastcs.methods import command

from redsun.services import configure_logging, identity, serve_controller
from redsun.services.sharedbox import BoxTransport


class MyController(Controller):
    def __init__(self) -> None:
        super().__init__()
        self.exposure = AttrRW(Float(units="ms", min=0.0))
        self.state = AttrR(String(length=16), initial_value="idle")

    @command()
    async def start(self) -> None:
        await self.state.update("running")


if __name__ == "__main__":
    configure_logging()
    prefix = None if identity() else "MYAPP:DET"
    try:
        asyncio.run(serve_controller(MyController(), BoxTransport(), prefix))
    except KeyboardInterrupt:
        pass

"""The session of the guide "How to serve a FastCS controller over sharedbox"."""

from __future__ import annotations

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

from ophyd_async.core import Device, SignalR, SignalRW, TriggerableCommand

from redsun import AsDevice, AsService, Declare, Launch
from redsun.device.sharedbox import BoxDeviceConnector, BoxPutError, stream_name
from redsun.qt import QtSession


class MyDetector(Device):
    exposure: SignalRW[float]
    state: SignalR[str]
    start: TriggerableCommand

    def __init__(self, prefix: str, name: str = "") -> None:
        self.counts_stream = stream_name(prefix, "counts")
        super().__init__(name=name, connector=BoxDeviceConnector(prefix))


async def try_exposure(detector: MyDetector, value: float) -> bool:
    try:
        await detector.exposure.set(value)
    except BoxPutError as error:
        print(f"{error.attribute} refused {value}: {error}")
        return False
    return True


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

    detector_service: Annotated[
        AsService, Launch("box_service", ready="detector ready", prefix="MYAPP:DET")
    ]
    detector: Annotated[AsDevice[MyDetector], Declare(service="detector_service")]


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