DeviceController

class caqtus.device.DeviceController(
device_name: DeviceName,
shot_event_dispatcher: ShotEventDispatcher,
)

Bases: ABC, Generic[DeviceProxyType, _P]

Controls a device during a shot.

Subclasses must implement run_shot() to define the behavior of the associated device during a shot.

This class is generic in the type of device proxy that it controls and the extra arguments passed to the run_shot() method.

device_name

The name of the device being controlled.

abstractmethod async run_shot(
device: DeviceProxyType,
/,
*args: _P.args,
**kwargs: _P.kwargs,
) → None

Runs a shot on the device.

This method must call wait_all_devices_ready() exactly once.

Parameters:

  • device – An asynchronous proxy to the device being controlled.

  • args –

    Extra arguments than can be computed before the shot and that are required to run the shot.

    These arguments are at the discretion of the subclass implementation.

  • kwargs –

    Extra arguments than can be computed before the shot and that are required to run the shot.

    These arguments are at the discretion of the subclass implementation.

A typical implementation can look like this:

async def run_shot(self, device, instruction):
    await device.program(instruction)
    await self.wait_all_devices_ready()
    await device.trigger()

    data = await device.read_data()
    self.signal_data_acquired("some label", data)

Warning

Implementations of this method must be careful to handle cancellation at a checkpoint in case the shot is aborted.

final async wait_all_devices_ready() → ShotTimer

Wait for all devices to be ready for time-sensitive operations.

This method must be called once the device has been programmed for the shot and is ready to be triggered or to react to data acquisition signals.

It must be called exactly once in run_shot().

The method will wait for all devices to be ready before returning.

Returns:

A timer with its start time set to the time when this function returns.

final signal_data_acquired(
label: DataLabel,
data: Data,
) → None

Signals that data has been acquired from the device.

Parameters:

  • label – The label of the data.

  • data – The data that was acquired.

final async wait_data_acquired(
label: DataLabel,
) → Data

Waits until another device signals that some data has been acquired.

final async sleep(seconds: float) → None

Sleeps for a given number of seconds.