Tango support

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.0rc1"]
 dev = [
     "ophyd_async[pva]",
     "ophyd_async[sim]",
     "ophyd_async[ca]",
+    "ophyd_async[tango]",
     "black",
     "flake8",
     "flake8-isort",
@@ -58,6 +60,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,
+    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_auto,
+    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,243 @@
+from __future__ import annotations
+
+from typing import (
+    Dict,
+    Optional,
+    Tuple,
+    Type,
+    TypeVar,
+    Union,
+    get_args,
+    get_origin,
+    get_type_hints,
+)
+
+from ophyd_async.core import (
+    DEFAULT_TIMEOUT,
+    Device,
+    DeviceVector,
+    Signal,
+)
+from ophyd_async.tango.signal import (
+    make_backend,
+    tango_signal_auto,
+)
+from tango import DeviceProxy as SyncDeviceProxy
+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: Optional[Union[AsyncDeviceProxy, SyncDeviceProxy]] = None
+    _polling: Tuple[bool, float, float, float] = (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: Optional[str] = None,
+        device_proxy: Optional[Union[AsyncDeviceProxy, SyncDeviceProxy]] = 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):
+                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():
+                if issubclass(type(child[1]), Signal):
+                    child[1]._backend.set_polling(*self._polling)  # noqa: SLF001
+                    child[1]._backend.allow_events(False)  # noqa: SLF001
+        if self._signal_polling:
+            for signal_name, polling in self._signal_polling.items():
+                if hasattr(self, signal_name):
+                    attr = getattr(self, signal_name)
+                    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: Optional[
+        Union[Tuple[float, float, float], Dict[str, Tuple[float, float, float]]]
+    ] = None,
+    signal_polling: Optional[Dict[str, Tuple[float, float, float]]] = 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.
+
+    Returns
+    -------
+    Callable
+        A class decorator that sets the `_polling` and `_signal_polling` attributes on
+        the decorated class.
+
+    Example
+    -------
+    Device-level and signal-level polling:
+    @tango_polling(
+        polling=(0.5, 1.0, 0.1),
+        signal_polling={
+            'signal1': (0.5, 1.0, 0.1),
+            'signal2': (1.0, 2.0, 0.2),
+        }
+    )
+    class MyTangoDevice(TangoDevice):
+        signal1: Signal
+        signal2: Signal
+    """
+    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):
+                setattr(device, name, origin())
+
+
+async def fill_proxy_entries(device: TangoDevice):
+    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: Union[Union[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: Union[Type[Device]]) -> Tuple[bool, Type[Device]]:
+    if get_origin(field) is DeviceVector:
+        return True, get_args(field)[0]
+    return False, field