`ophyd_async.tango.core`#

Package Contents#

Classes#

`AttributeProxy`	Used by the tango transport.
`AttributeConfig`	Base class for protocol classes.
`CommandProxy`	Tango proxy for commands.
`CommandConfig`	Base class for protocol classes.
`DevStateEnum`	All members should exist in the Backend, and there will be no extras.
`TangoSignalBackend`	Tango backend to connect signals over tango.
`TangoDevice`	General class for TangoDevices.
`TangoPolling`	Abstract base class for generic types.
`TangoDeviceConnector`	Defines how a `Device` should be connected and type hints processed.
`TangoLongStringTable`	An abstraction of a Table where each field is a column.
`TangoLongStringTableConverter`	Abstract base class for generic types.
`TangoDoubleStringTable`	An abstraction of a Table where each field is a column.
`TangoDoubleStringTableConverter`	Abstract base class for generic types.
`TangoConverter`	Abstract base class for generic types.
`TangoEnumConverter`	Abstract base class for generic types.
`TangoEnumArrayConverter`	Abstract base class for generic types.
`TangoDevStateConverter`	Abstract base class for generic types.
`TangoDevStateArrayConverter`	Abstract base class for generic types.
`TangoCommandBackend`	A backend for executing commands on a Tango device.

Functions#

`ensure_proper_executor`	Ensure decorated method has a proper asyncio executor.
`get_python_type`	Convert Tango types to Python types based on the configuration.
`get_dtype_extended`	For converting tango types to numpy datatype formats.
`get_source_metadata`
`get_tango_trl`	Get the tango resource locator.
`infer_python_type`	Infers the python type from the TRL.
`infer_signal_type`
`make_backend`
`parse_precision`
`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.
`try_to_cast_as_float`	Attempt to cast a value to float, returning None on failure.
`get_device_trl_and_attr`
`get_full_attr_trl`
`tango_command`	Create a `Command` backed by a Tango device command.
`tango_triggerable_command`	Create a `TriggerableCommand` backed by a void/void Tango device command.
`sig_from_types`
`make_converter`

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.

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, timeout: float | None = None) → AsyncStatus | None[source]#: Put value to TRL.

async get_config() → AttributeConfig[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.AttributeConfig[source]#

Bases: typing.Protocol

Base class for protocol classes.

Protocol classes are defined as::

class Proto(Protocol):
    def meth(self) -> int:
        ...

Such classes are primarily used with static type checkers that recognize structural subtyping (static duck-typing).

For example::

class C:
    def meth(self) -> int:
        return 0

def func(x: Proto) -> int:
    return x.meth()

func(C())  # Passes static type check

See PEP 544 for details. Protocol classes decorated with @typing.runtime_checkable act as simple-minded runtime protocols that check only the presence of given attributes, ignoring their type signatures. Protocol classes can be generic, they are defined as::

class GenProto(Protocol[T]):
    def meth(self) -> T:
        ...

data_type: int#: None

data_format: AttrDataFormat#: None

enum_labels: StdStringVector | list[str]#: None

format: str#: None

min_value: str#: None

max_value: str#: None

alarms: AttributeAlarmInfo#: None

unit: str#: None

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

Bases: ophyd_async.tango.core._tango_transport.TangoProxy

Tango proxy for commands.

device_proxy: DeviceProxy#: None

name: str#: None

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_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) → Any[source]#: Put value to TRL.

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

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.

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

Bases: typing.Protocol

Base class for protocol classes.

Protocol classes are defined as::

class Proto(Protocol):
    def meth(self) -> int:
        ...

Such classes are primarily used with static type checkers that recognize structural subtyping (static duck-typing).

For example::

class C:
    def meth(self) -> int:
        return 0

def func(x: Proto) -> int:
    return x.meth()

func(C())  # Passes static type check

See PEP 544 for details. Protocol classes decorated with @typing.runtime_checkable act as simple-minded runtime protocols that check only the presence of given attributes, ignoring their type signatures. Protocol classes can be generic, they are defined as::

class GenProto(Protocol[T]):
    def meth(self) -> T:
        ...

in_type: CmdArgType#: None

out_type: CmdArgType#: None

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

Bases: ophyd_async.core.StrictEnum

All members should exist in the Backend, and there will be no extras.

ON#: ‘ON’

OFF#: ‘OFF’

CLOSE#: ‘CLOSE’

OPEN#: ‘OPEN’

INSERT#: ‘INSERT’

EXTRACT#: ‘EXTRACT’

MOVING#: ‘MOVING’

STANDBY#: ‘STANDBY’

FAULT#: ‘FAULT’

INIT#: ‘INIT’

RUNNING#: ‘RUNNING’

ALARM#: ‘ALARM’

DISABLE#: ‘DISABLE’

UNKNOWN#: ‘UNKNOWN’

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, 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(config: AttributeConfig | CommandConfig, return_input_type: bool = False) → type[SignalDatatype] | None[source]#: Convert Tango types to Python types based on the configuration.

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

ophyd_async.tango.core.get_source_metadata(tango_resource: str, tr_configs: dict[str, AttributeConfig | CommandConfig]) → SignalMetadata[source]#

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) → tuple[type[SignalDatatype] | None, type[SignalDatatype] | None][source]#: Infers the python type from the TRL.

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

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

ophyd_async.tango.core.parse_precision(config: AttributeConfig)[source]#

ophyd_async.tango.core.tango_signal_r(datatype: type[SignalDatatypeT], read_trl: str, 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 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 = '', 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 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, 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 timeout: The timeout for the write operation name: The name of the Signal

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

Create a SignalX backed by 1 Tango Attribute/Command.

Deprecated since version 0.19: Use tango_triggerable_command instead.

Parameters

write_trl: The Attribute/Command to write its initial value to on execute timeout: The timeout for the command operation name: The name of the Signal

class ophyd_async.tango.core.TangoDevice(trl: str = '', support_events: bool = False, name: str = '', auto_fill_signals: bool = True)[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. 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.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, support_events: bool, auto_fill_signals: bool = True)[source]#

Bases: ophyd_async.core.DeviceConnector

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

set_trl(trl: str)[source]#

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.

class ophyd_async.tango.core.TangoLongStringTable(**kwargs)[source]#

Bases: ophyd_async.core.Table

An abstraction of a Table where each field is a column.

For example:

>>> from ophyd_async.core import Table, Array1D
>>> import numpy as np
>>> from collections.abc import Sequence
>>> class MyTable(Table):
...     a: Array1D[np.int8]
...     b: Sequence[str]
...
>>> t = MyTable(a=[1, 2], b=["x", "y"])
>>> len(t)  # the length is the number of rows
2
>>> t2 = t + t  # adding tables together concatenates them
>>> t2.a
array([1, 2, 1, 2], dtype=int8)
>>> t2.b
['x', 'y', 'x', 'y']
>>> t2[1]  # slice a row
array([(2, b'y')], dtype=[('a', 'i1'), ('b', 'S40')])

long: Array1D[int32]#: None

string: Sequence[str]#: None

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

Bases: ophyd_async.tango.core._converters.TangoConverter[ophyd_async.tango.core._utils.TangoLongStringTable]

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

write_value(value: TangoLongStringTable) → tuple[NDArray[int32], Sequence[str]][source]#: Convert from TangoLongStringTable to DevVarLongStringArray format.

value(value: tuple[NDArray[int32], Sequence[str]]) → TangoLongStringTable[source]#: Convert from DevVarLongStringArray format to TangoLongStringTable.

class ophyd_async.tango.core.TangoDoubleStringTable(**kwargs)[source]#

Bases: ophyd_async.core.Table

An abstraction of a Table where each field is a column.

For example:

>>> from ophyd_async.core import Table, Array1D
>>> import numpy as np
>>> from collections.abc import Sequence
>>> class MyTable(Table):
...     a: Array1D[np.int8]
...     b: Sequence[str]
...
>>> t = MyTable(a=[1, 2], b=["x", "y"])
>>> len(t)  # the length is the number of rows
2
>>> t2 = t + t  # adding tables together concatenates them
>>> t2.a
array([1, 2, 1, 2], dtype=int8)
>>> t2.b
['x', 'y', 'x', 'y']
>>> t2[1]  # slice a row
array([(2, b'y')], dtype=[('a', 'i1'), ('b', 'S40')])

double: Array1D[float64]#: None

string: Sequence[str]#: None

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

Bases: ophyd_async.tango.core._converters.TangoConverter[ophyd_async.tango.core._utils.TangoDoubleStringTable]

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

write_value(value: TangoDoubleStringTable) → tuple[NDArray[float64], Sequence[str]][source]#: Convert from TangoDoubleStringTable to DevVarDoubleStringArray format.

value(value: tuple[NDArray[float64], Sequence[str]]) → TangoDoubleStringTable[source]#: Convert from DevVarDoubleStringArray format to TangoDoubleStringTable.

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

Bases: typing.Generic[ophyd_async.core.SignalDatatypeT]

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

write_value(value: Any) → Any[source]#

value(value: Any) → Any[source]#

class ophyd_async.tango.core.TangoEnumConverter(labels: list[str])[source]#

Bases: ophyd_async.tango.core._converters.TangoConverter

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

write_value(value: str)[source]#

value(value: int)[source]#

class ophyd_async.tango.core.TangoEnumArrayConverter(labels: list[str])[source]#

Bases: ophyd_async.tango.core._converters.TangoConverter

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

write_value(value: NDArray[str_]) → NDArray[integer][source]#

value(value: NDArray[integer]) → NDArray[str_][source]#

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

Bases: ophyd_async.tango.core._converters.TangoConverter

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

write_value(value: str) → DevState[source]#

value(value: DevState) → str[source]#

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

Bases: ophyd_async.tango.core._converters.TangoConverter

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

write_value(value: NDArray[str_]) → NDArray[int_][source]#

value(value: NDArray[int_]) → NDArray[str_][source]#

ophyd_async.tango.core.try_to_cast_as_float(value: Any) → float | None[source]#: Attempt to cast a value to float, returning None on failure.

ophyd_async.tango.core.get_device_trl_and_attr(name: str)[source]#

ophyd_async.tango.core.get_full_attr_trl(device_trl: str, attr_name: str) → str[source]#

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

Bases: ophyd_async.core.CommandBackend[ophyd_async.tango.core._utils.P, ophyd_async.tango.core._utils.T]

A backend for executing commands on a Tango device.

Interfaces with a Tango device command via a CommandProxy, handling connection, type conversion, and execution while enforcing Tango’s limitations (no keyword arguments, single positional argument).

Parameters:

call_spec – Type signature of the Tango command, or None for void/void commands.
trl – The Tango Resource Locator of the command, e.g. tango://host:port/device/command.
device_proxy – An optional pre-configured DeviceProxy to use instead of creating one from trl.

set_trl(trl: str) → None[source]#

source(name: str) → str[source]#: Return source of command.

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

async execute(*args: args, **kwargs: kwargs) → T[source]#: Execute the command and return its result.

ophyd_async.tango.core.tango_command(call_spec: Callable[P, T], trl: str, device_proxy: DeviceProxy | None = None, *, timeout: float | None = DEFAULT_TIMEOUT, name: str = '') → Command[P, T][source]#

Create a Command backed by a Tango device command.

For void/void Tango commands use tango_triggerable_command instead. Tango commands only accept positional arguments; passing keyword arguments raises TypeError.

Parameters:

call_spec – A callable whose signature matches that of the Tango command, used to infer parameter and return types.
trl – The Tango Resource Locator of the command, e.g. tango://host:port/device/command.
device_proxy – An optional pre-configured DeviceProxy; if omitted one is created from trl.
timeout – Timeout in seconds for connecting to the Tango device.
name – Name for the command device node.

ophyd_async.tango.core.tango_triggerable_command(trl: str, device_proxy: DeviceProxy | None = None, *, timeout: float | None = DEFAULT_TIMEOUT, name: str = '') → TriggerableCommand[source]#

Create a TriggerableCommand backed by a void/void Tango device command.

Use this for Tango commands that take no arguments and return no value. For commands with arguments or a return type use tango_command.

Parameters:

trl – The Tango Resource Locator of the command, e.g. tango://host:port/device/command.
device_proxy – An optional pre-configured DeviceProxy; if omitted one is created from trl.
timeout – Timeout in seconds for connecting to the Tango device.
name – Name for the command device node.

ophyd_async.tango.core.sig_from_types(in_type: type[SignalDatatype] | None, out_type: type[SignalDatatype] | None)[source]#

ophyd_async.tango.core.make_converter(info: AttributeConfig | CommandConfig, datatype) → TangoConverter[source]#

ophyd_async.tango.core#

Package Contents#

Classes#

Functions#

API#

`ophyd_async.tango.core`#