secop_ophyd.SECoPDevices#

Module Contents#

Classes#

ParameterType

Annotation for Parameter Signals, defines the path to the parameter in the secclient module dict

PropertyType

Annotation for Module Property Signals, defines the path to the property

SECoPDeviceConnector

SECoPCMDDevice

Command devices that have Signals for command args, return values and a signal for triggering command execution (SignalX). They themselves are triggerable.

SECoPDevice

SECoPNodeDevice

SECoPCommunicatorDevice

SECoPReadableDevice

Standard readable SECoP device, corresponding to a SECoP module with the interface class “Readable”

SECoPTriggerableDevice

Standard triggerable SECoP device, corresponding to a SECoP module with the0s interface class “Triggerable”

SECoPWritableDevice

SECoPMoveableDevice

Standard movable SECoP device, corresponding to a SECoP module with the interface class “Drivable”

Functions#

Data#

API#

secop_ophyd.SECoPDevices.DISABLED = 0#
secop_ophyd.SECoPDevices.IDLE = 100#
secop_ophyd.SECoPDevices.STANDBY = 130#
secop_ophyd.SECoPDevices.PREPARED = 150#
secop_ophyd.SECoPDevices.WARN = 200#
secop_ophyd.SECoPDevices.WARN_STANDBY = 230#
secop_ophyd.SECoPDevices.WARN_PREPARED = 250#
secop_ophyd.SECoPDevices.NSTABLE = 270#
secop_ophyd.SECoPDevices.BUSY = 300#
secop_ophyd.SECoPDevices.DISABLING = 310#
secop_ophyd.SECoPDevices.INITIALIZING = 320#
secop_ophyd.SECoPDevices.PREPARING = 340#
secop_ophyd.SECoPDevices.STARTING = 360#
secop_ophyd.SECoPDevices.RAMPING = 370#
secop_ophyd.SECoPDevices.STABILIZING = 380#
secop_ophyd.SECoPDevices.FINALIZING = 390#
secop_ophyd.SECoPDevices.ERROR = 400#
secop_ophyd.SECoPDevices.ERROR_STANDBY = 430#
secop_ophyd.SECoPDevices.ERROR_PREPARED = 450#
secop_ophyd.SECoPDevices.UNKNOWN = 401#
secop_ophyd.SECoPDevices.IGNORED_PROPS = ['meaning', '_plotly']#
secop_ophyd.SECoPDevices.clean_identifier(anystring)[source]#
secop_ophyd.SECoPDevices.secop_enum_name_to_python(member_name: str) str[source]#

Convert SECoP enum member name to Python identifier.

Examples:

‘Low Energy’ -> ‘LOW_ENERGY’ ‘high-power’ -> ‘HIGH_POWER’ ‘Mode 1’ -> ‘MODE_1’

Parameters:

member_name – Original SECoP enum member name

Returns:

Python-compatible identifier in UPPER_CASE

secop_ophyd.SECoPDevices.format_assigned(device: ophyd_async.core.StandardReadable, signal: ophyd_async.core.SignalR) bool[source]#
secop_ophyd.SECoPDevices.is_read_signal(device: ophyd_async.core.StandardReadable, signal: ophyd_async.core.SignalR | ophyd_async.core.SignalRW) bool[source]#
secop_ophyd.SECoPDevices.is_config_signal(device: ophyd_async.core.StandardReadable, signal: ophyd_async.core.SignalR | ophyd_async.core.SignalRW) bool[source]#
class secop_ophyd.SECoPDevices.ParameterType[source]#

Annotation for Parameter Signals, defines the path to the parameter in the secclient module dict

class secop_ophyd.SECoPDevices.PropertyType[source]#

Annotation for Module Property Signals, defines the path to the property

class secop_ophyd.SECoPDevices.SECoPDeviceConnector(sri: str, auto_fill_signals: bool = True, loglevel=logging.INFO, logdir: str | None = None)[source]#

Bases: ophyd_async.core.DeviceConnector

sri: str = None#
module: str | None = None#
node_id: str = None#
set_module(module_name: str)[source]#
create_children_from_annotations(device: ophyd_async.core.Device)[source]#
fill_backend_with_path(backend: secop_ophyd.SECoPSignal.SECoPBackend, annotations: list[Any])[source]#
async connect_mock(device: ophyd_async.core.Device, mock: ophyd_async.core.LazyMock)[source]#
async connect_real(device: ophyd_async.core.Device, timeout: float, force_reconnect: bool)[source]#
class secop_ophyd.SECoPDevices.SECoPCMDDevice(path: secop_ophyd.util.Path, secclient: secop_ophyd.AsyncFrappyClient.AsyncFrappyClient)[source]#

Bases: ophyd_async.core.StandardReadable, bluesky.protocols.Flyable, bluesky.protocols.Triggerable

Command devices that have Signals for command args, return values and a signal for triggering command execution (SignalX). They themselves are triggerable.

Once the CMD Device is triggered, the command args are retrieved from the ‘argument’ Signal. The command message is sent to the SEC Node and the return value is written to ‘result’ signal.

Initialization

Initialize the CMD Device

Parameters:
  • path (Path) – Path to the command in the secclient module dict

  • secclient (AsyncFrappyClient) – SECoP client providing communication to the SEC Node

trigger() ophyd_async.core.AsyncStatus[source]#

Triggers the SECoPCMDDevice and sends command message to SEC Node. Command argument is taken form ‘argument’ Signal, and return value is written in the ‘return’ Signal

Returns:

A Status object, that is marked Done once the answer from the

SEC Node is received :rtype: AsyncStatus

kickoff() ophyd_async.core.AsyncStatus[source]#
complete() ophyd_async.core.AsyncStatus[source]#
collect() Iterator[bluesky.protocols.PartialEvent][source]#
class secop_ophyd.SECoPDevices.SECoPDevice(sri: str = '', name: str = '', connector: secop_ophyd.SECoPDevices.SECoPDeviceConnector | None = None, loglevel=logging.INFO, logdir: str | None = None)[source]#

Bases: ophyd_async.core.StandardReadable

hinted_signals: list[str] = []#
set_module(module_name: str)[source]#
async connect(mock: bool | ophyd_async.core.LazyMock = False, timeout: float = DEFAULT_TIMEOUT, force_reconnect: bool = False)[source]#
generate_cmd_plan(cmd_dev: secop_ophyd.SECoPDevices.SECoPCMDDevice, argument_type: Type | None = None, return_type: Type | None = None)[source]#
class secop_ophyd.SECoPDevices.SECoPNodeDevice(sec_node_uri: str = '', name: str = '', loglevel=logging.INFO, logdir: str | None = None)[source]#

Bases: secop_ophyd.SECoPDevices.SECoPDevice

hinted_signals: list[str] = []#
async connect(mock=False, timeout=DEFAULT_TIMEOUT, force_reconnect=False)[source]#
descriptiveDataChange(module, description)[source]#
nodeStateChange(online, state)[source]#

called when the state of the connection changes

‘online’ is True when connected or reconnecting, False when disconnected or connecting ‘state’ is the connection state as a string

class_from_instance(path_to_module: str | None = None)[source]#
class secop_ophyd.SECoPDevices.SECoPCommunicatorDevice(sri: str = '', name: str = '', connector: secop_ophyd.SECoPDevices.SECoPDeviceConnector | None = None, loglevel=logging.INFO, logdir: str | None = None)[source]#

Bases: secop_ophyd.SECoPDevices.SECoPDevice

hinted_signals: list[str] = []#
class secop_ophyd.SECoPDevices.SECoPReadableDevice(sri: str = '', name: str = '', connector: secop_ophyd.SECoPDevices.SECoPDeviceConnector | None = None, loglevel=logging.INFO, logdir: str | None = None)[source]#

Bases: secop_ophyd.SECoPDevices.SECoPDevice, bluesky.protocols.Triggerable, bluesky.protocols.Subscribable

Standard readable SECoP device, corresponding to a SECoP module with the interface class “Readable”

Initialization

Initializes the SECoPReadableDevice

Parameters:
  • secclient (AsyncFrappyClient) – SECoP client providing communication to the SEC Node

  • module_name (str) – Name of the SEC Node module that is represented by this device

hinted_signals: list[str] = ['value']#
async connect(mock=False, timeout=DEFAULT_TIMEOUT, force_reconnect=False)[source]#
async wait_for_idle()[source]#

asynchronously waits until module is IDLE again. this is helpful, for running commands that are not done immediately

observe_status_change(monitored_status_code: int)[source]#
trigger() ophyd_async.core.AsyncStatus[source]#
subscribe(function: ophyd_async.core._utils.Callback[dict[str, bluesky.protocols.Reading]]) None[source]#

Subscribe to updates in the reading

clear_sub(function: ophyd_async.core._utils.Callback) None[source]#

Remove a subscription.

class secop_ophyd.SECoPDevices.SECoPTriggerableDevice(sri: str = '', name: str = '', connector: secop_ophyd.SECoPDevices.SECoPDeviceConnector | None = None, loglevel=logging.INFO, logdir: str | None = None)[source]#

Bases: secop_ophyd.SECoPDevices.SECoPReadableDevice, bluesky.protocols.Stoppable

Standard triggerable SECoP device, corresponding to a SECoP module with the0s interface class “Triggerable”

Initialization

Initialize SECoPTriggerableDevice

Parameters:
  • secclient (AsyncFrappyClient) – SECoP client providing communication to the SEC Node

  • module_name (str) – ame of the SEC Node module that is represented by this device

hinted_signals: list[str] = ['value']#
class secop_ophyd.SECoPDevices.SECoPWritableDevice(sri: str = '', name: str = '', connector: secop_ophyd.SECoPDevices.SECoPDeviceConnector | None = None, loglevel=logging.INFO, logdir: str | None = None)[source]#

Bases: secop_ophyd.SECoPDevices.SECoPReadableDevice

hinted_signals: list[str] = ['target', 'value']#
class secop_ophyd.SECoPDevices.SECoPMoveableDevice(sri: str = '', name: str = '', connector: secop_ophyd.SECoPDevices.SECoPDeviceConnector | None = None, loglevel=logging.INFO, logdir: str | None = None)[source]#

Bases: secop_ophyd.SECoPDevices.SECoPReadableDevice, bluesky.protocols.Locatable, bluesky.protocols.Stoppable

Standard movable SECoP device, corresponding to a SECoP module with the interface class “Drivable”

Initialization

Initialize SECoPMovableDevice

Parameters:
  • secclient (AsyncFrappyClient) – SECoP client providing communication to the SEC Node

  • module_name (str) – ame of the SEC Node module that is represented by this device

hinted_signals: list[str] = ['target', 'value']#
async connect(mock=False, timeout=DEFAULT_TIMEOUT, force_reconnect=False)[source]#
set(new_target, timeout: Optional[float] = None) ophyd_async.core.AsyncStatus[source]#

Sends new target to SEC Nonde and waits until module is IDLE again

Parameters:
  • new_target (_type_) – new taget/setpoint for module

  • timeout (Optional[float], optional) – timeout for set operation, defaults to None

Returns:

Asyncstatus that gets set to Done once module is IDLE again

Return type:

AsyncStatus

async stop(success=True)[source]#

Calls stop command on the SEC Node module

Parameters:

success (bool, optional) – True: device is stopped as planned False: something has gone wrong (defaults to True)

async locate() bluesky.protocols.Location[source]#
secop_ophyd.SECoPDevices.class_from_interface(mod_properties: dict)[source]#
secop_ophyd.SECoPDevices.IF_CLASSES = None#