Tango support (#437)

Adding support for the Tango control system

This relies on PyTango (https://pypi.org/project/pytango/) ophyd-async devices to asynchronous PyTango DeviceProxy objects. The control strategy relies on a shared resource called `proxy` found in the new TangoDevice class. By passing this proxy to the TangoSignalBackend of its signals, a proxy to attributes or commands of the Tango device can be established.

    1. New TangoDevice and TangoReadable device classes.
    2. Automated inference of the existence unannotated signals
    3. Monitoring via Tango events with optional polling of attributes.
    4. Tango sensitive signals are constructed by attaching a TangoSignalBackend to a Signal object or may be built using new tango_signal_* constructor methods.
    5. Signal objects with a Tango backend and Tango devices should behave the same as those with EPICS or other backends.

    1. As of this commit, typed commands are not supported in Ophyd-Async so Tango command signals with a type other than None are automatically built as SignalRW as a workaround.
    2. Tango commands with different input/output types are not supported.
    3. Pipes are not supported.

    1. Extension of Device and StandardReadable to support shared resources such as the DeviceProxy.
    2. Extension of the Tango backend to support typed commands.
    3. Extension of the Tango backend to support pipes.

Contact:
Devin Burke
Research software scientist
Deutsches Elektronen-Synchrotron (DESY)
[email protected]

---------

Co-authored-by: matveyev <[email protected]>
Co-authored-by: Devin Burke <[email protected]>

docs/examples/tango_demo.py

@@ -0,0 +1,54 @@
+import asyncio
+
+import bluesky.plan_stubs as bps
+import bluesky.plans as bp
+from bluesky import RunEngine
+
+from ophyd_async.tango.demo import (
+ DemoCounter,
+ DemoMover,
+ TangoDetector,
+)
+from tango.test_context import MultiDeviceTestContext
+
+content = (
+ {
+ "class": DemoMover,
+ "devices": [{"name": "demo/motor/1"}],
+ },
+ {
+ "class": DemoCounter,
+ "devices": [{"name": "demo/counter/1"}, {"name": "demo/counter/2"}],
+ },
+)
+
+tango_context = MultiDeviceTestContext(content)
+
+
+async def main():
+ with tango_context:
+ detector = TangoDetector(
+ trl="",
+ name="detector",
+ counters_kwargs={"prefix": "demo/counter/", "count": 2},
+ mover_kwargs={"trl": "demo/motor/1"},
+ )
+ await detector.connect()
+
+ RE = RunEngine()
+
+ RE(bps.read(detector))
+ RE(bps.mv(detector, 0))
+ RE(bp.count(list(detector.counters.values())))
+
+ set_status = detector.set(1.0)
+ await asyncio.sleep(0.1)
+ stop_status = detector.stop()
+ await set_status
+ await stop_status
+ assert all([set_status.done, stop_status.done])
+ assert all([set_status.success, stop_status.success])
+
+
+if __name__ == "__main__":
+ asyncio.run(main())

pyproject.toml

@@ -33,10 +33,12 @@ requires-python = ">=3.10"
 ca = ["aioca>=1.6"]
 pva = ["p4p"]
 sim = ["h5py"]
+tango = ["pytango>=10.0.0"]
 dev = [
  "ophyd_async[pva]",
  "ophyd_async[sim]",
  "ophyd_async[ca]",
+ "ophyd_async[tango]",
  "black",
  "flake8",
  "flake8-isort",
@@ -59,6 +61,7 @@ dev = [
  "pytest-asyncio",
  "pytest-cov",
  "pytest-faulthandler",
+ "pytest-forked",
  "pytest-rerunfailures",
  "pytest-timeout",
  "ruff",

src/ophyd_async/tango/__init__.py

@@ -0,0 +1,45 @@
+from .base_devices import (
+ TangoDevice,
+ TangoReadable,
+ tango_polling,
+)
+from .signal import (
+ AttributeProxy,
+ CommandProxy,
+ TangoSignalBackend,
+ __tango_signal_auto,
+ ensure_proper_executor,
+ get_dtype_extended,
+ get_python_type,
+ get_tango_trl,
+ get_trl_descriptor,
+ infer_python_type,
+ infer_signal_character,
+ make_backend,
+ tango_signal_r,
+ tango_signal_rw,
+ tango_signal_w,
+ tango_signal_x,
+)
+
+__all__ = [
+ "TangoDevice",
+ "TangoReadable",
+ "tango_polling",
+ "TangoSignalBackend",
+ "get_python_type",
+ "get_dtype_extended",
+ "get_trl_descriptor",
+ "get_tango_trl",
+ "infer_python_type",
+ "infer_signal_character",
+ "make_backend",
+ "AttributeProxy",
+ "CommandProxy",
+ "ensure_proper_executor",
+ "__tango_signal_auto",
+ "tango_signal_r",
+ "tango_signal_rw",
+ "tango_signal_w",
+ "tango_signal_x",
+]

src/ophyd_async/tango/base_devices/__init__.py

@@ -0,0 +1,4 @@
+from ._base_device import TangoDevice, tango_polling
+from ._tango_readable import TangoReadable
+
+__all__ = ["TangoDevice", "TangoReadable", "tango_polling"]

src/ophyd_async/tango/base_devices/_base_device.py

@@ -0,0 +1,225 @@
+from __future__ import annotations
+
+from typing import (
+ TypeVar,
+ get_args,
+ get_origin,
+ get_type_hints,
+)
+
+from ophyd_async.core import (
+ DEFAULT_TIMEOUT,
+ Device,
+ Signal,
+)
+from ophyd_async.tango.signal import (
+ TangoSignalBackend,
+ __tango_signal_auto,
+ make_backend,
+)
+from tango import DeviceProxy as DeviceProxy
+from tango.asyncio import DeviceProxy as AsyncDeviceProxy
+
+T = TypeVar("T")
+
+
+class TangoDevice(Device):
+ """
+ General class for TangoDevices. Extends Device to provide attributes for Tango
+ devices.
+
+ Parameters
+ ----------
+ trl: str
+ Tango resource locator, typically of the device server.
+ device_proxy: Optional[Union[AsyncDeviceProxy, SyncDeviceProxy]]
+ 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 = ""
+ proxy: DeviceProxy | None = None
+ _polling: tuple[bool, float, float | None, float | None] = (False, 0.1, None, 0.1)
+ _signal_polling: dict[str, tuple[bool, float, float, float]] = {}
+ _poll_only_annotated_signals: bool = True
+
+ def __init__(
+ self,
+ trl: str | None = None,
+ device_proxy: DeviceProxy | None = None,
+ name: str = "",
+ ) -> None:
+ self.trl = trl if trl else ""
+ self.proxy = device_proxy
+ tango_create_children_from_annotations(self)
+ super().__init__(name=name)
+
+ def set_trl(self, trl: str):
+ """Set the Tango resource locator."""
+ if not isinstance(trl, str):
+ raise ValueError("TRL must be a string.")
+ self.trl = trl
+
+ async def connect(
+ self,
+ mock: bool = False,
+ timeout: float = DEFAULT_TIMEOUT,
+ force_reconnect: bool = False,
+ ):
+ if self.trl and self.proxy is None:
+ self.proxy = await AsyncDeviceProxy(self.trl)
+ elif self.proxy and not self.trl:
+ self.trl = self.proxy.name()
+
+ # Set the trl of the signal backends
+ for child in self.children():
+ if isinstance(child[1], Signal):
+ if isinstance(child[1]._backend, TangoSignalBackend): # noqa: SLF001
+ resource_name = child[0].lstrip("_")
+ read_trl = f"{self.trl}/{resource_name}"
+ child[1]._backend.set_trl(read_trl, read_trl) # noqa: SLF001
+
+ if self.proxy is not None:
+ self.register_signals()
+ await _fill_proxy_entries(self)
+
+ # set_name should be called again to propagate the new signal names
+ self.set_name(self.name)
+
+ # Set the polling configuration
+ if self._polling[0]:
+ for child in self.children():
+ child_type = type(child[1])
+ if issubclass(child_type, Signal):
+ if isinstance(child[1]._backend, TangoSignalBackend): # noqa: SLF001 # type: ignore
+ child[1]._backend.set_polling(*self._polling) # noqa: SLF001 # type: ignore
+ child[1]._backend.allow_events(False) # noqa: SLF001 # type: ignore
+ if self._signal_polling:
+ for signal_name, polling in self._signal_polling.items():
+ if hasattr(self, signal_name):
+ attr = getattr(self, signal_name)
+ if isinstance(attr._backend, TangoSignalBackend): # noqa: SLF001
+ attr._backend.set_polling(*polling) # noqa: SLF001
+ attr._backend.allow_events(False) # noqa: SLF001
+
+ await super().connect(mock=mock, timeout=timeout)
+
+ # Users can override this method to register new signals
+ def register_signals(self):
+ pass
+
+
+def tango_polling(
+ polling: tuple[float, float, float]
+ | dict[str, tuple[float, float, float]]
+ | None = None,
+ signal_polling: dict[str, tuple[float, float, float]] | None = None,
+):
+ """
+ Class decorator to configure polling for Tango devices.
+
+ This decorator allows for the configuration of both device-level and signal-level
+ polling for Tango devices. Polling is useful for device servers that do not support
+ event-driven updates.
+
+ Parameters
+ ----------
+ polling : Optional[Union[Tuple[float, float, float],
+ Dict[str, Tuple[float, float, float]]]], optional
+ Device-level polling configuration as a tuple of three floats representing the
+ polling interval, polling timeout, and polling delay. Alternatively,
+ a dictionary can be provided to specify signal-level polling configurations
+ directly.
+ signal_polling : Optional[Dict[str, Tuple[float, float, float]]], optional
+ Signal-level polling configuration as a dictionary where keys are signal names
+ and values are tuples of three floats representing the polling interval, polling
+ timeout, and polling delay.
+ """
+ if isinstance(polling, dict):
+ signal_polling = polling
+ polling = None
+
+ def decorator(cls):
+ if polling is not None:
+ cls._polling = (True, *polling)
+ if signal_polling is not None:
+ cls._signal_polling = {k: (True, *v) for k, v in signal_polling.items()}
+ return cls
+
+ return decorator
+
+
+def tango_create_children_from_annotations(
+ device: TangoDevice, included_optional_fields: tuple[str, ...] = ()
+):
+ """Initialize blocks at __init__ of `device`."""
+ for name, device_type in get_type_hints(type(device)).items():
+ if name in ("_name", "parent"):
+ continue
+
+ # device_type, is_optional = _strip_union(device_type)
+ # if is_optional and name not in included_optional_fields:
+ # continue
+ #
+ # is_device_vector, device_type = _strip_device_vector(device_type)
+ # if is_device_vector:
+ # n_device_vector = DeviceVector()
+ # setattr(device, name, n_device_vector)
+
+ # else:
+ origin = get_origin(device_type)
+ origin = origin if origin else device_type
+
+ if issubclass(origin, Signal):
+ type_args = get_args(device_type)
+ datatype = type_args[0] if type_args else None
+ backend = make_backend(datatype=datatype, device_proxy=device.proxy)
+ setattr(device, name, origin(name=name, backend=backend))
+
+ elif issubclass(origin, Device) or isinstance(origin, Device):
+ assert callable(origin), f"{origin} is not callable."
+ setattr(device, name, origin())
+
+
+async def _fill_proxy_entries(device: TangoDevice):
+ if device.proxy is None:
+ raise RuntimeError(f"Device proxy is not connected for {device.name}")
+ proxy_trl = device.trl
+ children = [name.lstrip("_") for name, _ in device.children()]
+ proxy_attributes = list(device.proxy.get_attribute_list())
+ proxy_commands = list(device.proxy.get_command_list())
+ combined = proxy_attributes + proxy_commands
+
+ for name in combined:
+ if name not in children:
+ full_trl = f"{proxy_trl}/{name}"
+ try:
+ auto_signal = await __tango_signal_auto(
+ trl=full_trl, device_proxy=device.proxy
+ )
+ setattr(device, name, auto_signal)
+ except RuntimeError as e:
+ if "Commands with different in and out dtypes" in str(e):
+ print(
+ f"Skipping {name}. Commands with different in and out dtypes"
+ f" are not supported."
+ )
+ continue
+ raise e
+
+
+# def _strip_union(field: T | T) -> tuple[T, bool]:
+# if get_origin(field) is Union:
+# args = get_args(field)
+# is_optional = type(None) in args
+# for arg in args:
+# if arg is not type(None):
+# return arg, is_optional
+# return field, False
+#
+#
+# def _strip_device_vector(field: type[Device]) -> tuple[bool, type[Device]]:
+# if get_origin(field) is DeviceVector:
+# return True, get_args(field)[0]
+# return False, field

src/ophyd_async/tango/base_devices/_tango_readable.py

@@ -0,0 +1,33 @@
+from __future__ import annotations
+
+from ophyd_async.core import (
+ StandardReadable,
+)
+from ophyd_async.tango.base_devices._base_device import TangoDevice
+from tango import DeviceProxy
+
+
+class TangoReadable(TangoDevice, StandardReadable):
+ """
+ General class for readable TangoDevices. Extends StandardReadable to provide
+ attributes for Tango devices.
+
+ Usage: to proper signals mount should be awaited:
+ new_device = await TangoDevice(<tango_device>)
+
+ Attributes
+ ----------
+ trl : str
+ Tango resource locator, typically of the device server.
+ proxy : AsyncDeviceProxy
+ AsyncDeviceProxy object for the device. This is created when the
+ device is connected.
+ """
+
+ def __init__(
+ self,
+ trl: str | None = None,
+ device_proxy: DeviceProxy | None = None,
+ name: str = "",
+ ) -> None:
+ TangoDevice.__init__(self, trl, device_proxy=device_proxy, name=name)

src/ophyd_async/tango/demo/__init__.py

@@ -0,0 +1,12 @@
+from ._counter import TangoCounter
+from ._detector import TangoDetector
+from ._mover import TangoMover
+from ._tango import DemoCounter, DemoMover
+
+__all__ = [
+ "DemoCounter",
+ "DemoMover",
+ "TangoCounter",
+ "TangoMover",
+ "TangoDetector",
+]