Creating an Odin Detector Driver

Introduction

This tutorial walks through creating a FastCS driver for a detector controlled by Odin. The fastcs-odin package builds on FastCS to provide an OdinController that introspects an Odin server and creates sub controllers and attributes for each adapter it finds.

This tutorial will walk through the creation of a driver that can:

• Introspect an Odin deployment and expose all parameters as PVs

• Configure and run detector acquisitions from a single top-level API

• Display a live view of captured frames

• Use logging and tracing to debug the data path

Set Up

Odin Deployment

The odin-data-example deployment container should be started:

docker run --rm -it --security-opt label=disable \
    -v /dev/shm:/dev/shm -v /tmp:/tmp --net=host \
    ghcr.io/odin-detector/odin-data-example-runtime:0.2.3

All applications should start without errors.

Python Environment

Clone fastcs-odin and open it in VS Code. Reopen in the dev container and install the dependencies:

pip install 'fastcs[epics]' pillow aioca

An example.py file should be created in the project root.

Phoebus

A Phoebus container should be started. A settings file will likely be needed to configure name servers for both PVA and CA.

A Minimal Controller

The core of a FastCS device driver is the Controller. An ExampleOdinController should be created that inherits from Controller with a single read-write integer attribute, and launched with FastCS.

Code 1

from fastcs.attributes import AttrRW
from fastcs.control_system import FastCS
from fastcs.controllers import Controller
from fastcs.datatypes import Int


class ExampleOdinController(Controller):
    foo = AttrRW(Int())


fastcs = FastCS(ExampleOdinController(), [])

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

The application will start and drop into an interactive shell. The attribute can be read and written from the shell:

In [1]: controller.foo.get()
Out[1]: 0

In [2]: await controller.foo.put(1)

In [3]: controller.foo.get()
Out[3]: 1

Note

There is also a helper if there are errors about running on the wrong event loop:

run(controller.foo.put(1))

Adding an EPICS Transport

An EPICS CA transport can be added to expose the controller’s attributes as PVs. The PV prefix should be unique.

Code 2

from fastcs.attributes import AttrRW
from fastcs.control_system import FastCS
from fastcs.controllers import Controller
from fastcs.datatypes import Int
from fastcs.transports.epics import EpicsIOCOptions
from fastcs.transports.epics.ca.transport import EpicsCATransport


class ExampleOdinController(Controller):
    foo = AttrRW(Int())


fastcs = FastCS(
    ExampleOdinController(),
    [EpicsCATransport(EpicsIOCOptions(pv_prefix="EXAMPLE"))],
)

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

The IOC will now be serving PVs. They can be listed in the interactive shell and interacted with from a terminal:

In [1]: dbl()
EXAMPLE:Foo_RBV
EXAMPLE:Foo
EXAMPLE:PVI_PV

❯ caget EXAMPLE:Foo_RBV
EXAMPLE:Foo_RBV            1
❯ caput EXAMPLE:Foo 5
Old : EXAMPLE:Foo                1
New : EXAMPLE:Foo                5
❯ caget EXAMPLE:Foo_RBV
EXAMPLE:Foo_RBV            5

Generating a Phoebus UI

FastCS can be configured to generate a Phoebus .bob file for a UI.

Code 3

from pathlib import Path

from fastcs.attributes import AttrRW
from fastcs.control_system import FastCS
from fastcs.controllers import Controller
from fastcs.datatypes import Int
from fastcs.transports.epics import EpicsGUIOptions, EpicsIOCOptions
from fastcs.transports.epics.ca.transport import EpicsCATransport


class ExampleOdinController(Controller):
    foo = AttrRW(Int())


fastcs = FastCS(
    ExampleOdinController(),
    [
        EpicsCATransport(
            EpicsIOCOptions(pv_prefix="EXAMPLE"),
            gui=EpicsGUIOptions(
                output_path=Path.cwd() / "opis" / "example.bob",
                title="Odin Example Detector",
            ),
        )
    ],
)

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

opis/example.bob should appear and can be opened in Phoebus with File > Open > example.bob. Values can be set from the UI to verify it works.

Connecting to Odin

The controller should be updated to inherit from OdinController instead of Controller. This controller connects to an Odin server, introspects all of its adapters, and creates sub controllers and attributes for each one automatically.

Code 4

from pathlib import Path

from fastcs.attributes import AttrRW
from fastcs.connections import IPConnectionSettings
from fastcs.control_system import FastCS
from fastcs.datatypes import Int
from fastcs.transports.epics import EpicsGUIOptions, EpicsIOCOptions
from fastcs.transports.epics.ca.transport import EpicsCATransport

from fastcs_odin.controllers import OdinController


class ExampleOdinController(OdinController):
    foo = AttrRW(Int())


fastcs = FastCS(
    ExampleOdinController(IPConnectionSettings("127.0.0.1", 8888)),
    [
        EpicsCATransport(
            EpicsIOCOptions(pv_prefix="EXAMPLE"),
            gui=EpicsGUIOptions(
                output_path=Path.cwd() / "opis" / "example.bob",
                title="Odin Example Detector",
            ),
        )
    ],
)

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

Note

Warnings about parameters failing to be read are expected for some adapter parameters.

The PVs can be listed in the interactive shell to see what has been created:

In [1]: dbl()
EXAMPLE:DETECTOR:ConfigExposureTime_RBV
...
EXAMPLE:FP:0:HDF:FileUseNumbers_RBV
...
EXAMPLE:FR:2:DecoderEnablePacketLogging_RBV
...
EXAMPLE:DETECTOR:Start

The controller now has sub controllers for each adapter:

In [2]: controller.sub_controllers
Out[2]:
{'DETECTOR': OdinAdapterController(path=DETECTOR, sub_controllers=None),
 'FR': FrameReceiverAdapterController(path=FR, sub_controllers=['0', '1', '2', '3']),
 'FP': FrameProcessorAdapterController(path=FP, sub_controllers=['0', '1', '2', '3']),
 'LIVE': OdinAdapterController(path=LIVE, sub_controllers=None),
 'SYSTEM': OdinAdapterController(path=SYSTEM, sub_controllers=None)}

Sub controllers for adapters can be mapped to specific classes, or the fallback OdinAdapterController, which introspects the parameter tree and adds no additional logic.

Phoebus UI

The display can be reloaded in Phoebus (right-click > Re-load display). There should now be buttons for each sub controller.

• Open the DETECTOR screen. Pressing Start should cause the frame counter to tick up.

• The FP screen has top-level attributes that read/write each individual FP instance. Things that must differ per instance (like CtrlEndpoint) are excluded. These can be seen in the screen for the specific instance, e.g. FP0.

• The FP screen also has PVs for the example dataset. These are detector-specific and defined in the Odin config file.

Running an Acquisition

An acquisition can now be run using the sub controller screens:

1. Set FP.FilePath = /tmp

2. Set FP.FilePrefix = test

3. Set FP.Frames = 10

4. Press FP.StartWriting

5. Check that FP.Writing is set

6. Set DETECTOR.Frames = 10

7. Press DETECTOR.Start

8. Watch FP.FramesWritten count up to 10 and then FP.Writing unset

The interactive shell will show the parameters that are being set.

Improving the API

Navigating between sub screens is fiddly. Top-level attributes can be added that fan values out to the relevant sub controllers. The foo attribute should be removed and an initialise method added to create new attributes after the Odin introspection has completed.

Code 5

from pathlib import Path

from fastcs.attributes import AttrRW
from fastcs.connections import IPConnectionSettings
from fastcs.control_system import FastCS
from fastcs.datatypes import Int, String
from fastcs.transports.epics import EpicsGUIOptions, EpicsIOCOptions
from fastcs.transports.epics.ca.transport import EpicsCATransport

from fastcs_odin.controllers import OdinController
from fastcs_odin.io.config_fan_sender_attribute_io import ConfigFanAttributeIORef


class ExampleOdinController(OdinController):
    async def initialise(self):
        await super().initialise()

        self.file_path = AttrRW(
            String(),
            io_ref=ConfigFanAttributeIORef([self.FP.file_path]),
        )
        self.file_prefix = AttrRW(
            String(),
            io_ref=ConfigFanAttributeIORef([self.FP.file_prefix]),
        )
        self.frames = AttrRW(
            Int(),
            io_ref=ConfigFanAttributeIORef(
                [self.FP.frames, self.DETECTOR.config_frames]
            ),
        )


fastcs = FastCS(
    ExampleOdinController(IPConnectionSettings("127.0.0.1", 8888)),
    [
        EpicsCATransport(
            EpicsIOCOptions(pv_prefix="EXAMPLE"),
            gui=EpicsGUIOptions(
                output_path=Path.cwd() / "opis" / "example.bob",
                title="Odin Example Detector",
            ),
        )
    ],
)

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

After running and reloading the display, the new PVs will appear on the top screen.

Note

FP and DETECTOR are accessed as attributes of self, but they are unknown to static type checkers because they are only created at runtime during Odin introspection. There will be no autocompletion for them yet.

Type Hints for Sub Controllers

FastCS validates type-hinted attributes, methods and sub controllers during initialisation to fail early if something is wrong. Typed sub controller classes can be created and type hints added to ExampleOdinController.

An ExampleFrameProcessorAdapterController inheriting from FrameProcessorAdapterController with a frames attribute hint, and an ExampleDetectorAdapterController inheriting from OdinAdapterController with a config_frames attribute hint should be created.

Note

frames has since been added to the parent FrameProcessorAdapterController, but is kept here as an example of the pattern that can be applied to any other attribute on the parent controller.

Type hints for FP and DETECTOR should be added on ExampleOdinController, and _create_adapter_controller overridden so that the correct controller types are instantiated.

Code 6

from pathlib import Path

from fastcs.attributes import AttrRW
from fastcs.connections import IPConnectionSettings
from fastcs.control_system import FastCS
from fastcs.controllers import BaseController
from fastcs.datatypes import Int, String
from fastcs.transports.epics import EpicsGUIOptions, EpicsIOCOptions
from fastcs.transports.epics.ca.transport import EpicsCATransport

from fastcs_odin.controllers import OdinController
from fastcs_odin.controllers.odin_adapter_controller import OdinAdapterController
from fastcs_odin.controllers.odin_data.frame_processor import (
    FrameProcessorAdapterController,
)
from fastcs_odin.http_connection import HTTPConnection
from fastcs_odin.io.config_fan_sender_attribute_io import ConfigFanAttributeIORef
from fastcs_odin.util import OdinParameter


class ExampleFrameProcessorAdapterController(FrameProcessorAdapterController):
    frames: AttrRW[int]


class ExampleDetectorAdapterController(OdinAdapterController):
    config_frames: AttrRW[int]


class ExampleOdinController(OdinController):
    FP: ExampleFrameProcessorAdapterController
    DETECTOR: ExampleDetectorAdapterController

    async def initialise(self):
        await super().initialise()

        self.file_path = AttrRW(
            String(),
            io_ref=ConfigFanAttributeIORef([self.FP.file_path]),
        )
        self.file_prefix = AttrRW(
            String(),
            io_ref=ConfigFanAttributeIORef([self.FP.file_prefix]),
        )
        self.frames = AttrRW(
            Int(),
            io_ref=ConfigFanAttributeIORef(
                [self.FP.frames, self.DETECTOR.config_frames]
            ),
        )

    def _create_adapter_controller(
        self,
        connection: HTTPConnection,
        parameters: list[OdinParameter],
        adapter: str,
        module: str,
    ) -> BaseController:
        match module:
            case "ExampleDetectorAdapter":
                return ExampleDetectorAdapterController(
                    connection, parameters, f"{self.API_PREFIX}/{adapter}", self._ios
                )
            case "FrameProcessorAdapter":
                return ExampleFrameProcessorAdapterController(
                    connection, parameters, f"{self.API_PREFIX}/{adapter}", self._ios
                )
            case _:
                return super()._create_adapter_controller(
                    connection, parameters, adapter, module
                )


fastcs = FastCS(
    ExampleOdinController(IPConnectionSettings("127.0.0.1", 8888)),
    [
        EpicsCATransport(
            EpicsIOCOptions(pv_prefix="EXAMPLE"),
            gui=EpicsGUIOptions(
                output_path=Path.cwd() / "opis" / "example.bob",
                title="Odin Example Detector",
            ),
        )
    ],
)

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

Without overriding _create_adapter_controller, the application will fail with:

RuntimeError: Controller 'ExampleOdinController' introspection of hinted sub controller
'DETECTOR' does not match defined type. Expected 'ExampleDetectorAdapterController'
got 'OdinAdapterController'.

Type hints for attributes or sub controllers that don’t exist will also fail:

RuntimeError: Controller `ExampleOdinController` failed to introspect hinted
controller `FOO` during initialisation

RuntimeError: Controller `ExampleOdinController` failed to introspect hinted
attribute `foo` during initialisation

Adding Commands

Parameters can now be set from the top screen, but acquisitions cannot be run yet. start and stop type hints should be added on ExampleDetectorAdapterController, along with acquire and stop command methods on ExampleOdinController.

Code 7

from pathlib import Path

from fastcs.attributes import AttrRW
from fastcs.connections import IPConnectionSettings
from fastcs.control_system import FastCS
from fastcs.controllers import BaseController
from fastcs.datatypes import Int, String
from fastcs.methods import Command, command
from fastcs.transports.epics import EpicsGUIOptions, EpicsIOCOptions
from fastcs.transports.epics.ca.transport import EpicsCATransport

from fastcs_odin.controllers import OdinController
from fastcs_odin.controllers.odin_adapter_controller import OdinAdapterController
from fastcs_odin.controllers.odin_data.frame_processor import (
    FrameProcessorAdapterController,
)
from fastcs_odin.http_connection import HTTPConnection
from fastcs_odin.io.config_fan_sender_attribute_io import ConfigFanAttributeIORef
from fastcs_odin.util import OdinParameter


class ExampleFrameProcessorAdapterController(FrameProcessorAdapterController):
    frames: AttrRW[int]


class ExampleDetectorAdapterController(OdinAdapterController):
    config_frames: AttrRW[int]

    start: Command
    stop: Command


class ExampleOdinController(OdinController):
    FP: ExampleFrameProcessorAdapterController
    DETECTOR: ExampleDetectorAdapterController

    @command()
    async def acquire(self):
        await self.FP.start_writing()
        await self.DETECTOR.start()

    @command()
    async def stop(self):
        await self.FP.stop_writing()
        await self.DETECTOR.stop()

    async def initialise(self):
        await super().initialise()

        self.file_path = AttrRW(
            String(),
            io_ref=ConfigFanAttributeIORef([self.FP.file_path]),
        )
        self.file_prefix = AttrRW(
            String(),
            io_ref=ConfigFanAttributeIORef([self.FP.file_prefix]),
        )
        self.frames = AttrRW(
            Int(),
            io_ref=ConfigFanAttributeIORef(
                [self.FP.frames, self.DETECTOR.config_frames]
            ),
        )

    def _create_adapter_controller(
        self,
        connection: HTTPConnection,
        parameters: list[OdinParameter],
        adapter: str,
        module: str,
    ) -> BaseController:
        match module:
            case "ExampleDetectorAdapter":
                return ExampleDetectorAdapterController(
                    connection, parameters, f"{self.API_PREFIX}/{adapter}", self._ios
                )
            case "FrameProcessorAdapter":
                return ExampleFrameProcessorAdapterController(
                    connection, parameters, f"{self.API_PREFIX}/{adapter}", self._ios
                )
            case _:
                return super()._create_adapter_controller(
                    connection, parameters, adapter, module
                )


fastcs = FastCS(
    ExampleOdinController(IPConnectionSettings("127.0.0.1", 8888)),
    [
        EpicsCATransport(
            EpicsIOCOptions(pv_prefix="EXAMPLE"),
            gui=EpicsGUIOptions(
                output_path=Path.cwd() / "opis" / "example.bob",
                title="Odin Example Detector",
            ),
        )
    ],
)

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

Note

FrameProcessorAdapterController already has start_writing and stop_writing defined statically, so they do not need to be added to ExampleFrameProcessorAdapterController.

After running and reloading Phoebus, an acquisition can now be started and stopped from the top screen.

Adding Status Attributes

The top screen can trigger acquisitions but has no status. status_acquiring and status_frames type hints should be added to ExampleDetectorAdapterController, and top-level summary attributes created that aggregate status from multiple sub controllers.

Code 8

from pathlib import Path

from fastcs.attributes import AttrR, AttrRW
from fastcs.connections import IPConnectionSettings
from fastcs.control_system import FastCS
from fastcs.controllers import BaseController
from fastcs.datatypes import Bool, Int, String
from fastcs.methods import Command, command
from fastcs.transports.epics import EpicsGUIOptions, EpicsIOCOptions
from fastcs.transports.epics.ca.transport import EpicsCATransport

from fastcs_odin.controllers import OdinController
from fastcs_odin.controllers.odin_adapter_controller import OdinAdapterController
from fastcs_odin.controllers.odin_data.frame_processor import (
    FrameProcessorAdapterController,
)
from fastcs_odin.http_connection import HTTPConnection
from fastcs_odin.io.config_fan_sender_attribute_io import ConfigFanAttributeIORef
from fastcs_odin.io.status_summary_attribute_io import StatusSummaryAttributeIORef
from fastcs_odin.util import OdinParameter


class ExampleFrameProcessorAdapterController(FrameProcessorAdapterController):
    frames: AttrRW[int]


class ExampleDetectorAdapterController(OdinAdapterController):
    config_frames: AttrRW[int]

    status_acquiring: AttrR[bool]
    status_frames: AttrR[int]

    start: Command
    stop: Command


class ExampleOdinController(OdinController):
    FP: ExampleFrameProcessorAdapterController
    DETECTOR: ExampleDetectorAdapterController

    @command()
    async def acquire(self):
        await self.FP.start_writing()
        await self.DETECTOR.start()

    @command()
    async def stop(self):
        await self.FP.stop_writing()
        await self.DETECTOR.stop()

    async def initialise(self):
        await super().initialise()

        self.file_path = AttrRW(
            String(),
            io_ref=ConfigFanAttributeIORef([self.FP.file_path]),
        )
        self.file_prefix = AttrRW(
            String(),
            io_ref=ConfigFanAttributeIORef([self.FP.file_prefix]),
        )
        self.frames = AttrRW(
            Int(),
            io_ref=ConfigFanAttributeIORef(
                [self.FP.frames, self.DETECTOR.config_frames]
            ),
        )

        self.acquiring = AttrR(
            Bool(),
            io_ref=StatusSummaryAttributeIORef(
                [], "", any, [self.FP.writing, self.DETECTOR.status_acquiring]
            ),
        )
        self.frames_captured = AttrR(
            Int(),
            io_ref=StatusSummaryAttributeIORef(
                [], "", min, [self.DETECTOR.status_frames, self.FP.frames_written]
            ),
        )

    def _create_adapter_controller(
        self,
        connection: HTTPConnection,
        parameters: list[OdinParameter],
        adapter: str,
        module: str,
    ) -> BaseController:
        match module:
            case "ExampleDetectorAdapter":
                return ExampleDetectorAdapterController(
                    connection, parameters, f"{self.API_PREFIX}/{adapter}", self._ios
                )
            case "FrameProcessorAdapter":
                return ExampleFrameProcessorAdapterController(
                    connection, parameters, f"{self.API_PREFIX}/{adapter}", self._ios
                )
            case _:
                return super()._create_adapter_controller(
                    connection, parameters, adapter, module
                )


fastcs = FastCS(
    ExampleOdinController(IPConnectionSettings("127.0.0.1", 8888)),
    [
        EpicsCATransport(
            EpicsIOCOptions(pv_prefix="EXAMPLE"),
            gui=EpicsGUIOptions(
                output_path=Path.cwd() / "opis" / "example.bob",
                title="Odin Example Detector",
            ),
        )
    ],
)

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

The StatusSummaryAttributeIORef aggregates values from multiple attributes. The any function is used for acquiring (true if any source is active) and min for frames_captured (the lowest count across sources).

Controller Logic

There is a risk that the file writer is slower to start than the detector and frames will be missed. wait_for_value can be used to ensure that the file writers have started before starting the detector.

Code 9

from pathlib import Path

from fastcs.attributes import AttrR, AttrRW
from fastcs.connections import IPConnectionSettings
from fastcs.control_system import FastCS
from fastcs.controllers import BaseController
from fastcs.datatypes import Bool, Int, String
from fastcs.methods import Command, command
from fastcs.transports.epics import EpicsGUIOptions, EpicsIOCOptions
from fastcs.transports.epics.ca.transport import EpicsCATransport

from fastcs_odin.controllers import OdinController
from fastcs_odin.controllers.odin_adapter_controller import OdinAdapterController
from fastcs_odin.controllers.odin_data.frame_processor import (
    FrameProcessorAdapterController,
)
from fastcs_odin.http_connection import HTTPConnection
from fastcs_odin.io.config_fan_sender_attribute_io import ConfigFanAttributeIORef
from fastcs_odin.io.status_summary_attribute_io import StatusSummaryAttributeIORef
from fastcs_odin.util import OdinParameter


class ExampleFrameProcessorAdapterController(FrameProcessorAdapterController):
    frames: AttrRW[int]


class ExampleDetectorAdapterController(OdinAdapterController):
    config_frames: AttrRW[int]

    status_acquiring: AttrR[bool]
    status_frames: AttrR[int]

    start: Command
    stop: Command


class ExampleOdinController(OdinController):
    FP: ExampleFrameProcessorAdapterController
    DETECTOR: ExampleDetectorAdapterController

    @command()
    async def acquire(self):
        await self.FP.start_writing()
        await self.FP.writing.wait_for_value(True, timeout=1)

        await self.DETECTOR.start()

    @command()
    async def stop(self):
        await self.FP.stop_writing()
        await self.DETECTOR.stop()

    async def initialise(self):
        await super().initialise()

        self.file_path = AttrRW(
            String(),
            io_ref=ConfigFanAttributeIORef([self.FP.file_path]),
        )
        self.file_prefix = AttrRW(
            String(),
            io_ref=ConfigFanAttributeIORef([self.FP.file_prefix]),
        )
        self.frames = AttrRW(
            Int(),
            io_ref=ConfigFanAttributeIORef(
                [self.FP.frames, self.DETECTOR.config_frames]
            ),
        )

        self.acquiring = AttrR(
            Bool(),
            io_ref=StatusSummaryAttributeIORef(
                [], "", any, [self.FP.writing, self.DETECTOR.status_acquiring]
            ),
        )
        self.frames_captured = AttrR(
            Int(),
            io_ref=StatusSummaryAttributeIORef(
                [], "", min, [self.DETECTOR.status_frames, self.FP.frames_written]
            ),
        )

    def _create_adapter_controller(
        self,
        connection: HTTPConnection,
        parameters: list[OdinParameter],
        adapter: str,
        module: str,
    ) -> BaseController:
        match module:
            case "ExampleDetectorAdapter":
                return ExampleDetectorAdapterController(
                    connection, parameters, f"{self.API_PREFIX}/{adapter}", self._ios
                )
            case "FrameProcessorAdapter":
                return ExampleFrameProcessorAdapterController(
                    connection, parameters, f"{self.API_PREFIX}/{adapter}", self._ios
                )
            case _:
                return super()._create_adapter_controller(
                    connection, parameters, adapter, module
                )


fastcs = FastCS(
    ExampleOdinController(IPConnectionSettings("127.0.0.1", 8888)),
    [
        EpicsCATransport(
            EpicsIOCOptions(pv_prefix="EXAMPLE"),
            gui=EpicsGUIOptions(
                output_path=Path.cwd() / "opis" / "example.bob",
                title="Odin Example Detector",
            ),
        )
    ],
)

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

Live View with Scan Methods

The live view adapter can be used to see frames as they pass through. A Waveform attribute can be added to display the image along with a @scan method to periodically query the live view adapter for images.

Code 10

from io import BytesIO
from pathlib import Path

import numpy as np
from fastcs.attributes import AttrR, AttrRW
from fastcs.connections import IPConnectionSettings
from fastcs.control_system import FastCS
from fastcs.controllers import BaseController
from fastcs.datatypes import Bool, Int, String, Waveform
from fastcs.methods import Command, command, scan
from fastcs.transports.epics import EpicsGUIOptions, EpicsIOCOptions
from fastcs.transports.epics.ca.transport import EpicsCATransport
from PIL import Image

from fastcs_odin.controllers import OdinController
from fastcs_odin.controllers.odin_adapter_controller import OdinAdapterController
from fastcs_odin.controllers.odin_data.frame_processor import (
    FrameProcessorAdapterController,
)
from fastcs_odin.http_connection import HTTPConnection
from fastcs_odin.io.config_fan_sender_attribute_io import ConfigFanAttributeIORef
from fastcs_odin.io.status_summary_attribute_io import StatusSummaryAttributeIORef
from fastcs_odin.util import OdinParameter


class ExampleFrameProcessorAdapterController(FrameProcessorAdapterController):
    frames: AttrRW[int]


class ExampleDetectorAdapterController(OdinAdapterController):
    config_frames: AttrRW[int]

    status_acquiring: AttrR[bool]
    status_frames: AttrR[int]

    start: Command
    stop: Command


class ExampleOdinController(OdinController):
    FP: ExampleFrameProcessorAdapterController
    DETECTOR: ExampleDetectorAdapterController

    live_view_image = AttrR(Waveform("uint8", shape=(256, 256)))

    @scan(1)
    async def monitor_live_view(self):
        response, image_bytes = await self.connection.get_bytes(
            f"{self.API_PREFIX}/live/image"
        )

        if response.status != 200:
            return

        image = Image.open(BytesIO(image_bytes))
        numpy_array = np.asarray(image)
        await self.live_view_image.update(numpy_array[:, :, 0])

    @command()
    async def acquire(self):
        await self.FP.start_writing()
        await self.FP.writing.wait_for_value(True, timeout=1)

        await self.DETECTOR.start()

    @command()
    async def stop(self):
        await self.FP.stop_writing()
        await self.DETECTOR.stop()

    async def initialise(self):
        await super().initialise()

        self.file_path = AttrRW(
            String(),
            io_ref=ConfigFanAttributeIORef([self.FP.file_path]),
        )
        self.file_prefix = AttrRW(
            String(),
            io_ref=ConfigFanAttributeIORef([self.FP.file_prefix]),
        )
        self.frames = AttrRW(
            Int(),
            io_ref=ConfigFanAttributeIORef(
                [self.FP.frames, self.DETECTOR.config_frames]
            ),
        )

        self.acquiring = AttrR(
            Bool(),
            io_ref=StatusSummaryAttributeIORef(
                [], "", any, [self.FP.writing, self.DETECTOR.status_acquiring]
            ),
        )
        self.frames_captured = AttrR(
            Int(),
            io_ref=StatusSummaryAttributeIORef(
                [], "", min, [self.DETECTOR.status_frames, self.FP.frames_written]
            ),
        )

    def _create_adapter_controller(
        self,
        connection: HTTPConnection,
        parameters: list[OdinParameter],
        adapter: str,
        module: str,
    ) -> BaseController:
        match module:
            case "ExampleDetectorAdapter":
                return ExampleDetectorAdapterController(
                    connection, parameters, f"{self.API_PREFIX}/{adapter}", self._ios
                )
            case "FrameProcessorAdapter":
                return ExampleFrameProcessorAdapterController(
                    connection, parameters, f"{self.API_PREFIX}/{adapter}", self._ios
                )
            case _:
                return super()._create_adapter_controller(
                    connection, parameters, adapter, module
                )


fastcs = FastCS(
    ExampleOdinController(IPConnectionSettings("127.0.0.1", 8888)),
    [
        EpicsCATransport(
            EpicsIOCOptions(pv_prefix="EXAMPLE"),
            gui=EpicsGUIOptions(
                output_path=Path.cwd() / "opis" / "example.bob",
                title="Odin Example Detector",
            ),
        )
    ],
)

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

Note

The image dimensions and dtype could be queried from LIVE.Shape and LIVE.FrameDtype at runtime to create live_view_image dynamically.

The EPICS CA transport does not support 2D arrays, so it will give a warning and the LiveViewImage PV will not be created via CA.

Adding PV Access

An EpicsPVATransport can be added alongside the existing EpicsCATransport to serve both transports simultaneously. PVA supports 2D array attributes, so the LiveViewImage PV will be available over PVA.

Code 11

from io import BytesIO
from pathlib import Path

import numpy as np
from fastcs.attributes import AttrR, AttrRW
from fastcs.connections import IPConnectionSettings
from fastcs.control_system import FastCS
from fastcs.controllers import BaseController
from fastcs.datatypes import Bool, Int, String, Waveform
from fastcs.methods import Command, command, scan
from fastcs.transports.epics import EpicsGUIOptions, EpicsIOCOptions
from fastcs.transports.epics.ca.transport import EpicsCATransport
from fastcs.transports.epics.pva.transport import EpicsPVATransport
from PIL import Image

from fastcs_odin.controllers import OdinController
from fastcs_odin.controllers.odin_adapter_controller import OdinAdapterController
from fastcs_odin.controllers.odin_data.frame_processor import (
    FrameProcessorAdapterController,
)
from fastcs_odin.http_connection import HTTPConnection
from fastcs_odin.io.config_fan_sender_attribute_io import ConfigFanAttributeIORef
from fastcs_odin.io.status_summary_attribute_io import StatusSummaryAttributeIORef
from fastcs_odin.util import OdinParameter


class ExampleFrameProcessorAdapterController(FrameProcessorAdapterController):
    frames: AttrRW[int]


class ExampleDetectorAdapterController(OdinAdapterController):
    config_frames: AttrRW[int]

    status_acquiring: AttrR[bool]
    status_frames: AttrR[int]

    start: Command
    stop: Command


class ExampleOdinController(OdinController):
    FP: ExampleFrameProcessorAdapterController
    DETECTOR: ExampleDetectorAdapterController

    live_view_image = AttrR(Waveform("uint8", shape=(256, 256)))

    @scan(1)
    async def monitor_live_view(self):
        response, image_bytes = await self.connection.get_bytes(
            f"{self.API_PREFIX}/live/image"
        )

        if response.status != 200:
            return

        image = Image.open(BytesIO(image_bytes))
        numpy_array = np.asarray(image)
        await self.live_view_image.update(numpy_array[:, :, 0])

    @command()
    async def acquire(self):
        await self.FP.start_writing()
        await self.FP.writing.wait_for_value(True, timeout=1)

        await self.DETECTOR.start()

    @command()
    async def stop(self):
        await self.FP.stop_writing()
        await self.DETECTOR.stop()

    async def initialise(self):
        await super().initialise()

        self.file_path = AttrRW(
            String(),
            io_ref=ConfigFanAttributeIORef([self.FP.file_path]),
        )
        self.file_prefix = AttrRW(
            String(),
            io_ref=ConfigFanAttributeIORef([self.FP.file_prefix]),
        )
        self.frames = AttrRW(
            Int(),
            io_ref=ConfigFanAttributeIORef(
                [self.FP.frames, self.DETECTOR.config_frames]
            ),
        )

        self.acquiring = AttrR(
            Bool(),
            io_ref=StatusSummaryAttributeIORef(
                [], "", any, [self.FP.writing, self.DETECTOR.status_acquiring]
            ),
        )
        self.frames_captured = AttrR(
            Int(),
            io_ref=StatusSummaryAttributeIORef(
                [], "", min, [self.DETECTOR.status_frames, self.FP.frames_written]
            ),
        )

    def _create_adapter_controller(
        self,
        connection: HTTPConnection,
        parameters: list[OdinParameter],
        adapter: str,
        module: str,
    ) -> BaseController:
        match module:
            case "ExampleDetectorAdapter":
                return ExampleDetectorAdapterController(
                    connection, parameters, f"{self.API_PREFIX}/{adapter}", self._ios
                )
            case "FrameProcessorAdapter":
                return ExampleFrameProcessorAdapterController(
                    connection, parameters, f"{self.API_PREFIX}/{adapter}", self._ios
                )
            case _:
                return super()._create_adapter_controller(
                    connection, parameters, adapter, module
                )


fastcs = FastCS(
    ExampleOdinController(IPConnectionSettings("127.0.0.1", 8888)),
    [
        EpicsCATransport(
            EpicsIOCOptions(pv_prefix="EXAMPLE"),
        ),
        EpicsPVATransport(
            EpicsIOCOptions(pv_prefix="EXAMPLE"),
            gui=EpicsGUIOptions(
                output_path=Path.cwd() / "opis" / "example.bob",
                title="Odin Example Detector",
            ),
        ),
    ],
)

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

An acquisition can be run and the live view watched (LiveViewImage button):

1. Set DETECTOR.Frames = 0 (continuous)

2. Press DETECTOR.Start

Logging and Tracing

There is a custom logger built into fastcs that allows structured logging and trace-level logging that can be enabled at runtime per-attribute.

Add INFO level log statements to the acquire and stop commands to record when writing starts and stops.

Code 12

from io import BytesIO
from pathlib import Path

import numpy as np
from fastcs.attributes import AttrR, AttrRW
from fastcs.connections import IPConnectionSettings
from fastcs.control_system import FastCS
from fastcs.controllers import BaseController
from fastcs.datatypes import Bool, Int, String, Waveform
from fastcs.logging import LogLevel, configure_logging, logger
from fastcs.methods import Command, command, scan
from fastcs.transports.epics import EpicsGUIOptions, EpicsIOCOptions
from fastcs.transports.epics.ca.transport import EpicsCATransport
from fastcs.transports.epics.pva.transport import EpicsPVATransport
from PIL import Image

from fastcs_odin.controllers import OdinController
from fastcs_odin.controllers.odin_adapter_controller import OdinAdapterController
from fastcs_odin.controllers.odin_data.frame_processor import (
    FrameProcessorAdapterController,
)
from fastcs_odin.http_connection import HTTPConnection
from fastcs_odin.io.config_fan_sender_attribute_io import ConfigFanAttributeIORef
from fastcs_odin.io.status_summary_attribute_io import StatusSummaryAttributeIORef
from fastcs_odin.util import OdinParameter


class ExampleFrameProcessorAdapterController(FrameProcessorAdapterController):
    frames: AttrRW[int]


class ExampleDetectorAdapterController(OdinAdapterController):
    config_frames: AttrRW[int]

    status_acquiring: AttrR[bool]
    status_frames: AttrR[int]

    start: Command
    stop: Command


class ExampleOdinController(OdinController):
    FP: ExampleFrameProcessorAdapterController
    DETECTOR: ExampleDetectorAdapterController

    live_view_image = AttrR(Waveform("uint8", shape=(256, 256)))

    @scan(1)
    async def monitor_live_view(self):
        response, image_bytes = await self.connection.get_bytes(
            f"{self.API_PREFIX}/live/image"
        )

        if response.status != 200:
            return

        image = Image.open(BytesIO(image_bytes))
        numpy_array = np.asarray(image)
        await self.live_view_image.update(numpy_array[:, :, 0])

    @command()
    async def acquire(self):
        logger.info("Starting writing")
        await self.FP.start_writing()
        await self.FP.writing.wait_for_value(True, timeout=1)

        await self.DETECTOR.start()

    @command()
    async def stop(self):
        logger.info("Stopping writing")
        await self.FP.stop_writing()
        await self.DETECTOR.stop()

    async def initialise(self):
        await super().initialise()

        self.file_path = AttrRW(
            String(),
            io_ref=ConfigFanAttributeIORef([self.FP.file_path]),
        )
        self.file_prefix = AttrRW(
            String(),
            io_ref=ConfigFanAttributeIORef([self.FP.file_prefix]),
        )
        self.frames = AttrRW(
            Int(),
            io_ref=ConfigFanAttributeIORef(
                [self.FP.frames, self.DETECTOR.config_frames]
            ),
        )

        self.acquiring = AttrR(
            Bool(),
            io_ref=StatusSummaryAttributeIORef(
                [], "", any, [self.FP.writing, self.DETECTOR.status_acquiring]
            ),
        )
        self.frames_captured = AttrR(
            Int(),
            io_ref=StatusSummaryAttributeIORef(
                [], "", min, [self.DETECTOR.status_frames, self.FP.frames_written]
            ),
        )

    def _create_adapter_controller(
        self,
        connection: HTTPConnection,
        parameters: list[OdinParameter],
        adapter: str,
        module: str,
    ) -> BaseController:
        match module:
            case "ExampleDetectorAdapter":
                return ExampleDetectorAdapterController(
                    connection, parameters, f"{self.API_PREFIX}/{adapter}", self._ios
                )
            case "FrameProcessorAdapter":
                return ExampleFrameProcessorAdapterController(
                    connection, parameters, f"{self.API_PREFIX}/{adapter}", self._ios
                )
            case _:
                return super()._create_adapter_controller(
                    connection, parameters, adapter, module
                )


configure_logging(LogLevel.TRACE)

fastcs = FastCS(
    ExampleOdinController(IPConnectionSettings("127.0.0.1", 8888)),
    [
        EpicsCATransport(
            EpicsIOCOptions(pv_prefix="EXAMPLE"),
        ),
        EpicsPVATransport(
            EpicsIOCOptions(pv_prefix="EXAMPLE"),
            gui=EpicsGUIOptions(
                output_path=Path.cwd() / "opis" / "example.bob",
                title="Odin Example Detector",
            ),
        ),
    ],
)

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

Running an acquisition will then produce log output:

[2025-12-23 12:07:31.615+0000 I] Starting writing
[2025-12-23 12:07:32.012+0000 I] Stopping writing

TRACE level logging must be configured for trace messages to appear. This will not produce additional output until tracing is enabled at runtime, although it will enable DEBUG level output. There are trace level messages logged throughout the fastcs codebase to enable debugging live systems. All classes that inherit Tracer can call log_event to emit trace messages, which can be enabled at runtime by calling enable_tracing. For example, Attributes:

In [1]: controller.FP[0].HDF.file_path.enable_tracing()
[2025-12-23 12:07:31.615+0000 T] Query for parameter [ParameterTreeAttributeIO] uri=api/0.1/fp/0/config/hdf/file/path, response={'path': '/tmp'}
...
In [2]: controller.FP[0].HDF.file_path.disable_tracing()