`ophyd_async.tango.core`#

Package Contents#

Classes#

`AttributeProxy`	Used by the tango transport.
`CommandProxy`	Tango proxy for commands.
`TangoSignalBackend`	Tango backend to connect signals over tango.
`TangoDevice`	General class for TangoDevices.
`TangoReadable`	General class for TangoDevices.
`TangoPolling`	Abstract base class for generic types.
`TangoDeviceConnector`	Defines how a `Device` should be connected and type hints processed.

Functions#

`ensure_proper_executor`	Ensure decorated method has a proper asyncio executor.
`get_python_type`	For converting between recieved tango types and python primatives.
`get_dtype_extended`	For converting tango types to numpy datatype formats.
`get_trl_descriptor`	Create a descriptor from a tango resource locator.
`get_tango_trl`	Get the tango resource locator.
`infer_python_type`	Infers the python type from the TRL.
`infer_signal_type`
`make_backend`
`tango_signal_r`	Create a `SignalR` backed by 1 Tango Attribute/Command.
`tango_signal_rw`	Create a `SignalRW` backed by 1 or 2 Tango Attribute/Command.
`tango_signal_w`	Create a `SignalW` backed by 1 Tango Attribute/Command.
`tango_signal_x`	Create a `SignalX` backed by 1 Tango Attribute/Command.

API#

class ophyd_async.tango.core.AttributeProxy(device_proxy: DeviceProxy, name: str)[source]#

Bases: ophyd_async.tango.core._tango_transport.TangoProxy

Used by the tango transport.

exception: BaseException | None#: None

async connect() → None[source]#

Perform actions after proxy is connected.

e.g. check if signal can be subscribed.

async get() → object[source]#: Get value from TRL.

async get_w_value() → object[source]#: Get last written value from TRL.

async put(value: object | None, wait: bool = True, timeout: float | None = None) → AsyncStatus | None[source]#: Put value to TRL.

async get_config() → AttributeInfoEx[source]#: Get TRL config async.

async get_reading() → Reading[source]#: Get reading from TRL.

has_subscription() → bool[source]#: Indicate that this trl already subscribed.

subscribe_callback(callback: Callback | None)[source]#: Subscribe tango CHANGE event to callback.

unsubscribe_callback()[source]#: Delete CHANGE event subscription.

async poll()[source]#

Poll the attribute and call the callback if the value has changed.

Only callback if value has changed by more than the absolute or relative change. This function is used when an attribute that does not support events is cached or a callback is passed to it.

set_polling(allow_polling: bool = False, polling_period: float = 0.5, abs_change: float | None = None, rel_change: float | None = 0.1)[source]#: Set the polling parameters.

class ophyd_async.tango.core.CommandProxy(device_proxy: DeviceProxy, name: str)[source]#

Bases: ophyd_async.tango.core._tango_transport.TangoProxy

Tango proxy for commands.

abstractmethod subscribe_callback(callback: Callback | None) → None[source]#: Subscribe tango CHANGE event to callback.

abstractmethod unsubscribe_callback() → None[source]#: Delete CHANGE event subscription.

async get() → object[source]#: Get value from TRL.

async get_w_value() → object[source]#: Get last written value from TRL.

async connect() → None[source]#

Perform actions after proxy is connected.

e.g. check if signal can be subscribed.

async put(value: object | None, wait: bool = True, timeout: float | None = None) → AsyncStatus | None[source]#: Put value to TRL.

async get_config() → CommandInfo[source]#: Get TRL config async.

async get_reading() → Reading[source]#: Get reading from TRL.

set_polling(allow_polling: bool = False, polling_period: float = 0.5, abs_change: float | None = None, rel_change: float | None = 0.1)[source]#: Set polling parameters.

ophyd_async.tango.core.ensure_proper_executor(func: Callable[P, Coroutine[Any, Any, R]]) → Callable[P, Coroutine[Any, Any, R]][source]#: Ensure decorated method has a proper asyncio executor.

class ophyd_async.tango.core.TangoSignalBackend(datatype: type[SignalDatatypeT] | None, read_trl: str = '', write_trl: str = '', device_proxy: DeviceProxy | None = None)[source]#

Bases: ophyd_async.core.SignalBackend[ophyd_async.core.SignalDatatypeT]

Tango backend to connect signals over tango.

classmethod datatype_allowed(dtype: Any) → bool[source]#

set_trl(read_trl: str = '', write_trl: str = '')[source]#

source(name: str, read: bool) → str[source]#

Return source of signal.

Parameters:

name – The name of the signal, which can be used or discarded.
read – True if we want the source for reading, False if writing.

async connect(timeout: float) → None[source]#: Connect to underlying hardware.

async put(value: SignalDatatypeT | None, wait=True, timeout=None) → None[source]#: Put a value to the PV, if wait then wait for completion.

async get_datakey(source: str) → DataKey[source]#: Metadata like source, dtype, shape, precision, units.

async get_reading() → Reading[SignalDatatypeT][source]#: Return the current value, timestamp and severity.

async get_value() → SignalDatatypeT[source]#: Return the current value.

async get_setpoint() → SignalDatatypeT[source]#: Return the point that a signal was requested to move to.

set_callback(callback: Callback | None) → None[source]#: Observe changes to the current value, timestamp and severity.

set_polling(allow_polling: bool = True, polling_period: float = 0.1, abs_change: float | None = None, rel_change: float | None = 0.1)[source]#

allow_events(allow: bool = True)[source]#

ophyd_async.tango.core.get_python_type(tango_type: CmdArgType) → tuple[bool, object, str][source]#: For converting between recieved tango types and python primatives.

ophyd_async.tango.core.get_dtype_extended(datatype) → object | None[source]#: For converting tango types to numpy datatype formats.

ophyd_async.tango.core.get_trl_descriptor(datatype: type | None, tango_resource: str, tr_configs: dict[str, AttributeInfoEx | CommandInfo]) → DataKey[source]#: Create a descriptor from a tango resource locator.

async ophyd_async.tango.core.get_tango_trl(full_trl: str, device_proxy: DeviceProxy | TangoProxy | None, timeout: float) → TangoProxy[source]#: Get the tango resource locator.

async ophyd_async.tango.core.infer_python_type(trl: str = '', proxy: DeviceProxy | None = None) → object | NDArray | type[DevState] | IntEnum[source]#: Infers the python type from the TRL.

async ophyd_async.tango.core.infer_signal_type(trl, proxy: DeviceProxy | None = None) → type[Signal] | None[source]#

ophyd_async.tango.core.make_backend(datatype: type[SignalDatatypeT] | None, read_trl: str = '', write_trl: str = '', device_proxy: DeviceProxy | None = None) → TangoSignalBackend[source]#

ophyd_async.tango.core.tango_signal_r(datatype: type[SignalDatatypeT], read_trl: str, device_proxy: DeviceProxy | None = None, timeout: float = DEFAULT_TIMEOUT, name: str = '') → SignalR[SignalDatatypeT][source]#

Create a SignalR backed by 1 Tango Attribute/Command.

Parameters

datatype: Check that the Attribute/Command is of this type read_trl: The Attribute/Command to read and monitor device_proxy: If given, this DeviceProxy will be used timeout: The timeout for the read operation name: The name of the Signal

ophyd_async.tango.core.tango_signal_rw(datatype: type[SignalDatatypeT], read_trl: str, write_trl: str = '', device_proxy: DeviceProxy | None = None, timeout: float = DEFAULT_TIMEOUT, name: str = '') → SignalRW[SignalDatatypeT][source]#

Create a SignalRW backed by 1 or 2 Tango Attribute/Command.

Parameters

datatype: Check that the Attribute/Command is of this type read_trl: The Attribute/Command to read and monitor write_trl: If given, use this Attribute/Command to write to, otherwise use read_trl device_proxy: If given, this DeviceProxy will be used timeout: The timeout for the read and write operations name: The name of the Signal

ophyd_async.tango.core.tango_signal_w(datatype: type[SignalDatatypeT], write_trl: str, device_proxy: DeviceProxy | None = None, timeout: float = DEFAULT_TIMEOUT, name: str = '') → SignalW[SignalDatatypeT][source]#

Create a SignalW backed by 1 Tango Attribute/Command.

Parameters

datatype: Check that the Attribute/Command is of this type write_trl: The Attribute/Command to write to device_proxy: If given, this DeviceProxy will be used timeout: The timeout for the write operation name: The name of the Signal

ophyd_async.tango.core.tango_signal_x(write_trl: str, device_proxy: DeviceProxy | None = None, timeout: float = DEFAULT_TIMEOUT, name: str = '') → SignalX[source]#

Create a SignalX backed by 1 Tango Attribute/Command.

Parameters

write_trl: The Attribute/Command to write its initial value to on execute device_proxy: If given, this DeviceProxy will be used timeout: The timeout for the command operation name: The name of the Signal

class ophyd_async.tango.core.TangoDevice(trl: str | None = None, device_proxy: DeviceProxy | None = None, support_events: bool = False, name: str = '')[source]#

Bases: ophyd_async.core.Device

General class for TangoDevices.

Extends Device to provide attributes for Tango devices.

Parameters:

trl – Tango resource locator, typically of the device server.
device_proxy – Asynchronous or synchronous DeviceProxy object for the device. If not provided, an asynchronous DeviceProxy object will be created using the trl and awaited when the device is connected.

trl: str = <Multiline-String>#

proxy: DeviceProxy | None#: None

class ophyd_async.tango.core.TangoReadable(trl: str | None = None, device_proxy: DeviceProxy | None = None, name: str = '')[source]#

Bases: ophyd_async.tango.core._base_device.TangoDevice, ophyd_async.core.StandardReadable

General class for TangoDevices.

Extends Device to provide attributes for Tango devices.

Parameters:

trl – Tango resource locator, typically of the device server.
device_proxy – Asynchronous or synchronous DeviceProxy object for the device. If not provided, an asynchronous DeviceProxy object will be created using the trl and awaited when the device is connected.

class ophyd_async.tango.core.TangoPolling[source]#

Bases: typing.Generic[ophyd_async.tango.core._base_device.T]

Abstract base class for generic types.

A generic type is typically declared by inheriting from this class parameterized with one or more type variables. For example, a generic mapping type might be defined as::

class Mapping(Generic[KT, VT]): def getitem(self, key: KT) -> VT: … # Etc.

This class can then be used as follows::

def lookup_name(mapping: Mapping[KT, VT], key: KT, default: VT) -> VT: try: return mapping[key] except KeyError: return default

ophyd_polling_period: float#: 0.1

abs_change: T | None#: None

rel_change: T | None#: None

class ophyd_async.tango.core.TangoDeviceConnector(trl: str | None, device_proxy: DeviceProxy | None, support_events: bool)[source]#

Bases: ophyd_async.core.DeviceConnector

Defines how a Device should be connected and type hints processed.

create_children_from_annotations(device: Device)[source]#

Use when children can be created from introspecting the hardware.

Some control systems allow introspection of a device to determine what children it has. To allow this to work nicely with typing we add these hints to the Device like so::

my_signal: SignalRW[int]
my_device: MyDevice

This method will be run during Device.__init__, and is responsible for turning all of those type hints into real Signal and Device instances.

Subsequent runs of this function should do nothing, to allow it to be called early in Devices that need to pass references to their children during __init__.

async connect_mock(device: Device, mock: LazyMock)[source]#

Use during Device.connect with mock=True.

This is called when there is no cached connect done in mock=True mode. It connects the Device and all its children in mock mode.

async connect_real(device: Device, timeout: float, force_reconnect: bool)[source]#

Use during Device.connect with mock=False.

This is called when there is no cached connect done in mock=False mode. It connects the Device and all its children in real mode in parallel.

ophyd_async.tango.core#

Package Contents#

Classes#

Functions#

API#

This Page

`ophyd_async.tango.core`#