dodal.devices.util.epics_util#

Classes

SetWhenEnabled

A device that sets the proc field of a PV when it becomes enabled.

Functions

call_func(func)

epics_signal_put_wait(pv_name[, wait])

Creates a Component around an EpicsSignal that waits for a callback on a put.

run_functions_without_blocking(...[, ...])

Creates and initiates an asynchronous chaining of functions which return a status
dodal.devices.util.epics_util.epics_signal_put_wait(pv_name: str, wait: float = 3.0) → Component[EpicsSignal][source]#

Creates a Component around an EpicsSignal that waits for a callback on a put.

Parameters:

  • pv_name (str) – The name of the PV for the EpicsSignal

  • wait (str, optional) – The timeout to wait for a callback. Defaults to 3.0.

Returns:

An EpicsSignal that will wait for a callback.

Return type:

EpicsSignal

dodal.devices.util.epics_util.run_functions_without_blocking(functions_to_chain: Sequence[Callable[[], StatusBase]], timeout: float = 60.0, associated_obj: Device | None = None) → Status[source]#

Creates and initiates an asynchronous chaining of functions which return a status

Usage: This function can be used to take a series of status-returning functions and run them all sequentially and in the background by making use of callbacks. It also ensures exceptions on each returned status are propagated

Args: functions_to_chain( list(function - > StatusBase) ): A list of functions which each

return a status object

associated_obj (Device | None): The device that should be associated with the

returned status

Returns: Status: A status object which is marked as complete once all of the Status objects returned by the unwrapped functions have completed.

Parameters:

  • functions_to_chain (Sequence[Callable[[], StatusBase]])

  • timeout (float, default: 60.0)

  • associated_obj (Device | None, default: None)

Return type:

Status

dodal.devices.util.epics_util.call_func(func: Callable[[], StatusBase]) → StatusBase[source]#
Parameters:

func (Callable[[], StatusBase])

Return type:

StatusBase