diff --git a/src/ophyd_async/tango/core/_tango_transport.py b/src/ophyd_async/tango/core/_tango_transport.py index 6de2c85e96..77394905f6 100644 --- a/src/ophyd_async/tango/core/_tango_transport.py +++ b/src/ophyd_async/tango/core/_tango_transport.py @@ -189,7 +189,7 @@ async def _write(): raise TimeoutError(f"{self._name} attr put failed: Timeout") from te except DevFailed as de: raise RuntimeError( - f"{self._name} device" f" failure: {de.args[0].desc}" + f"{self._name} device failure: {de.args[0].desc}" ) from de else: diff --git a/tests/tango/test_tango_signals.py b/tests/tango/test_tango_signals.py index 63c6c25ee9..0ec269527e 100644 --- a/tests/tango/test_tango_signals.py +++ b/tests/tango/test_tango_signals.py @@ -1,8 +1,10 @@ import asyncio import textwrap import time +from dataclasses import dataclass from enum import Enum, IntEnum from random import choice +from typing import Generic import numpy as np import numpy.typing as npt @@ -59,8 +61,19 @@ class TestEnum(IntEnum): ("enum", "DevEnum", TestEnum, (TestEnum.A, TestEnum.B)), ) -ATTRIBUTES_SET = [] -COMMANDS_SET = [] + +@dataclass +class AttributeData(Generic[T]): + pv: str + tango_type: str + d_format: AttrDataFormat + py_type: type[T] + initial_value: T + put_value: T + + +ATTRIBUTES_SET: list[AttributeData] = [] +COMMANDS_SET: list[AttributeData] = [] for type_name, tango_type_name, py_type, values in BASE_TYPES_SET: # pytango test utils currently fail to handle bool pytest.approx @@ -68,7 +81,7 @@ class TestEnum(IntEnum): continue ATTRIBUTES_SET.extend( [ - ( + AttributeData( f"{type_name}_scalar_attr", tango_type_name, AttrDataFormat.SCALAR, @@ -76,7 +89,7 @@ class TestEnum(IntEnum): choice(values), choice(values), ), - ( + AttributeData( f"{type_name}_spectrum_attr", tango_type_name, AttrDataFormat.SPECTRUM, @@ -84,7 +97,7 @@ class TestEnum(IntEnum): [choice(values), choice(values), choice(values)], [choice(values), choice(values), choice(values)], ), - ( + AttributeData( f"{type_name}_image_attr", tango_type_name, AttrDataFormat.IMAGE, @@ -105,7 +118,7 @@ class TestEnum(IntEnum): continue else: COMMANDS_SET.append( - ( + AttributeData( f"{type_name}_scalar_cmd", tango_type_name, AttrDataFormat.SCALAR, @@ -118,7 +131,7 @@ class TestEnum(IntEnum): continue else: COMMANDS_SET.append( - ( + AttributeData( f"{type_name}_spectrum_cmd", tango_type_name, AttrDataFormat.SPECTRUM, @@ -148,11 +161,11 @@ class EchoDevice(Device): attr_values = {} def initialize_dynamic_attributes(self): - for name, typ, form, _, _, _ in ATTRIBUTES_SET: + for attr_data in ATTRIBUTES_SET: attr = attribute( - name=name, - dtype=typ, - dformat=form, + name=attr_data.pv, + dtype=attr_data.tango_type, + dformat=attr_data.d_format, access=AttrWriteType.READ_WRITE, fget=self.read, fset=self.write, @@ -161,15 +174,15 @@ def initialize_dynamic_attributes(self): enum_labels=[member.name for member in TestEnum], ) self.add_attribute(attr) - self.set_change_event(name, True, False) + self.set_change_event(attr_data.pv, True, False) - for name, typ, form, _, _, _ in COMMANDS_SET: + for cmd_data in COMMANDS_SET: cmd = command( - f=getattr(self, name), - dtype_in=typ, - dformat_in=form, - dtype_out=typ, - dformat_out=form, + f=getattr(self, cmd_data.pv), + dtype_in=cmd_data.tango_type, + dformat_in=cmd_data.d_format, + dtype_out=cmd_data.tango_type, + dformat_out=cmd_data.d_format, ) self.add_command(cmd) @@ -188,8 +201,8 @@ def echo_command(self, arg): """ ) - for name, _, _, _, _, _ in COMMANDS_SET: - exec(echo_command_code.replace("echo_command", name)) + for cmd_data in COMMANDS_SET: + exec(echo_command_code.replace("echo_command", cmd_data.pv)) # -------------------------------------------------------------------- @@ -319,33 +332,23 @@ async def assert_monitor_then_put( # -------------------------------------------------------------------- @pytest.mark.asyncio -@pytest.mark.parametrize( - "pv, tango_type, d_format, py_type, initial_value, put_value", - ATTRIBUTES_SET, - ids=[x[0] for x in ATTRIBUTES_SET], -) -async def test_backend_get_put_monitor_attr( - echo_device: str, - pv: str, - tango_type: str, - d_format: AttrDataFormat, - py_type: type[T], - initial_value: T, - put_value: T, -): +async def test_backend_get_put_monitor_attr(echo_device: str): try: - # Set a timeout for the operation to prevent it from running indefinitely - await asyncio.wait_for( - assert_monitor_then_put( - echo_device, - pv, - initial_value, - put_value, - get_test_descriptor(py_type, initial_value, False), - py_type, - ), - timeout=100, # Timeout in seconds - ) + for attr_data in ATTRIBUTES_SET: + # Set a timeout for the operation to prevent it from running indefinitely + await asyncio.wait_for( + assert_monitor_then_put( + echo_device, + attr_data.pv, + attr_data.initial_value, + attr_data.put_value, + get_test_descriptor( + attr_data.py_type, attr_data.initial_value, False + ), + attr_data.py_type, + ), + timeout=100, # Timeout in seconds + ) except asyncio.TimeoutError: pytest.fail("Test timed out") except Exception as e: @@ -381,384 +384,237 @@ async def assert_put_read( # -------------------------------------------------------------------- @pytest.mark.asyncio -@pytest.mark.parametrize( - "pv, tango_type, d_format, py_type, initial_value, put_value", - COMMANDS_SET, - ids=[x[0] for x in COMMANDS_SET], -) async def test_backend_get_put_monitor_cmd( echo_device: str, - pv: str, - tango_type: str, - d_format: AttrDataFormat, - py_type: type[T], - initial_value: T, - put_value: T, ): - # With the given datatype, check we have the correct initial value and putting works - descriptor = get_test_descriptor(py_type, initial_value, True) - await assert_put_read(echo_device, pv, put_value, descriptor, py_type) - # # With guessed datatype, check we can set it back to the initial value - await assert_put_read(echo_device, pv, put_value, descriptor) - tasks = [t for t in asyncio.all_tasks() if t is not asyncio.current_task()] - await asyncio.gather(*tasks) - del echo_device + for cmd_data in COMMANDS_SET: + # With the given datatype, check we have the correct initial value + # and putting works + descriptor = get_test_descriptor(cmd_data.py_type, cmd_data.initial_value, True) + await assert_put_read( + echo_device, cmd_data.pv, cmd_data.put_value, descriptor, cmd_data.py_type + ) + # # With guessed datatype, check we can set it back to the initial value + await assert_put_read(echo_device, cmd_data.pv, cmd_data.put_value, descriptor) + tasks = [t for t in asyncio.all_tasks() if t is not asyncio.current_task()] + await asyncio.gather(*tasks) # -------------------------------------------------------------------- @pytest.mark.asyncio -@pytest.mark.parametrize( - "pv, tango_type, d_format, py_type, initial_value, put_value, use_proxy", - [ - ( - pv, - tango_type, - d_format, - py_type, - initial_value, - put_value, - use_proxy, - ) - for ( - pv, - tango_type, - d_format, - py_type, - initial_value, - put_value, - ) in ATTRIBUTES_SET - for use_proxy in [True, False] - ], - ids=[f"{x[0]}_{use_proxy}" for x in ATTRIBUTES_SET for use_proxy in [True, False]], -) async def test_tango_signal_r( echo_device: str, - pv: str, - tango_type: str, - d_format: AttrDataFormat, - py_type: type[T], - initial_value: T, - put_value: T, - use_proxy: bool, ): - await prepare_device(echo_device, pv, initial_value) - source = echo_device + "/" + pv - proxy = await DeviceProxy(echo_device) if use_proxy else None - timeout = 0.2 - signal = tango_signal_r( - datatype=py_type, - read_trl=source, - device_proxy=proxy, - timeout=timeout, - name="test_signal", - ) - await signal.connect() - reading = await signal.read() - assert_close(reading["test_signal"]["value"], initial_value) + for use_proxy in [True, False]: + proxy = await DeviceProxy(echo_device) if use_proxy else None + for attr_data in ATTRIBUTES_SET: + await prepare_device(echo_device, attr_data.pv, attr_data.initial_value) + source = echo_device + "/" + attr_data.pv + signal = tango_signal_r( + datatype=attr_data.py_type, + read_trl=source, + device_proxy=proxy, + timeout=timeout, + name="test_signal", + ) + await signal.connect() + reading = await signal.read() + assert_close(reading["test_signal"]["value"], attr_data.initial_value) # -------------------------------------------------------------------- @pytest.mark.asyncio -@pytest.mark.parametrize( - "pv, tango_type, d_format, py_type, initial_value, put_value, use_proxy", - [ - ( - pv, - tango_type, - d_format, - py_type, - initial_value, - put_value, - use_proxy, - ) - for ( - pv, - tango_type, - d_format, - py_type, - initial_value, - put_value, - ) in ATTRIBUTES_SET - for use_proxy in [True, False] - ], - ids=[f"{x[0]}_{use_proxy}" for x in ATTRIBUTES_SET for use_proxy in [True, False]], -) async def test_tango_signal_w( echo_device: str, - pv: str, - tango_type: str, - d_format: AttrDataFormat, - py_type: type[T], - initial_value: T, - put_value: T, - use_proxy: bool, ): - await prepare_device(echo_device, pv, initial_value) - source = echo_device + "/" + pv - proxy = await DeviceProxy(echo_device) if use_proxy else None - - timeout = 0.2 - signal = tango_signal_w( - datatype=py_type, - write_trl=source, - device_proxy=proxy, - timeout=timeout, - name="test_signal", - ) - await signal.connect() - status = signal.set(put_value, wait=True, timeout=timeout) - await status - assert status.done is True and status.success is True + for use_proxy in [True, False]: + proxy = await DeviceProxy(echo_device) if use_proxy else None + timeout = 0.2 + for attr_data in ATTRIBUTES_SET: + await prepare_device(echo_device, attr_data.pv, attr_data.initial_value) + source = echo_device + "/" + attr_data.pv + + signal = tango_signal_w( + datatype=attr_data.py_type, + write_trl=source, + device_proxy=proxy, + timeout=timeout, + name="test_signal", + ) + await signal.connect() + status = signal.set(attr_data.put_value, wait=True, timeout=timeout) + await status + assert status.done is True and status.success is True - status = signal.set(put_value, wait=False, timeout=timeout) - await status - assert status.done is True and status.success is True + status = signal.set(attr_data.put_value, wait=False, timeout=timeout) + await status + assert status.done is True and status.success is True - status = signal.set(put_value, wait=True) - await status - assert status.done is True and status.success is True + status = signal.set(attr_data.put_value, wait=True) + await status + assert status.done is True and status.success is True - status = signal.set(put_value, wait=False) - await status - assert status.done is True and status.success is True + status = signal.set(attr_data.put_value, wait=False) + await status + assert status.done is True and status.success is True # -------------------------------------------------------------------- @pytest.mark.asyncio -@pytest.mark.parametrize( - "pv, tango_type, d_format, py_type, initial_value, put_value, use_proxy", - [ - ( - pv, - tango_type, - d_format, - py_type, - initial_value, - put_value, - use_proxy, - ) - for ( - pv, - tango_type, - d_format, - py_type, - initial_value, - put_value, - ) in ATTRIBUTES_SET - for use_proxy in [True, False] - ], - ids=[f"{x[0]}_{use_proxy}" for x in ATTRIBUTES_SET for use_proxy in [True, False]], -) async def test_tango_signal_rw( echo_device: str, - pv: str, - tango_type: str, - d_format: AttrDataFormat, - py_type: type[T], - initial_value: T, - put_value: T, - use_proxy: bool, ): - await prepare_device(echo_device, pv, initial_value) - source = echo_device + "/" + pv - proxy = await DeviceProxy(echo_device) if use_proxy else None - timeout = 0.2 - signal = tango_signal_rw( - datatype=py_type, - read_trl=source, - write_trl=source, - device_proxy=proxy, - timeout=timeout, - name="test_signal", - ) - await signal.connect() - reading = await signal.read() - assert_close(reading["test_signal"]["value"], initial_value) - await signal.set(put_value) - location = await signal.locate() - assert_close(location["setpoint"], put_value) - assert_close(location["readback"], put_value) + for use_proxy in [True, False]: + proxy = await DeviceProxy(echo_device) if use_proxy else None + for attr_data in ATTRIBUTES_SET: + await prepare_device(echo_device, attr_data.pv, attr_data.initial_value) + source = echo_device + "/" + attr_data.pv + + signal = tango_signal_rw( + datatype=attr_data.py_type, + read_trl=source, + write_trl=source, + device_proxy=proxy, + timeout=timeout, + name="test_signal", + ) + await signal.connect() + reading = await signal.read() + assert_close(reading["test_signal"]["value"], attr_data.initial_value) + await signal.set(attr_data.put_value) + location = await signal.locate() + assert_close(location["setpoint"], attr_data.put_value) + assert_close(location["readback"], attr_data.put_value) # -------------------------------------------------------------------- @pytest.mark.asyncio -@pytest.mark.parametrize("use_proxy", [True, False]) -async def test_tango_signal_x(tango_test_device: str, use_proxy: bool): - proxy = await DeviceProxy(tango_test_device) if use_proxy else None +async def test_tango_signal_x(tango_test_device: str): timeout = 0.2 - signal = tango_signal_x( - write_trl=tango_test_device + "/" + "clear", - device_proxy=proxy, - timeout=timeout, - name="test_signal", - ) - await signal.connect() - status = signal.trigger() - await status - assert status.done is True and status.success is True + for use_proxy in [True, False]: + proxy = await DeviceProxy(tango_test_device) if use_proxy else None + signal = tango_signal_x( + write_trl=tango_test_device + "/" + "clear", + device_proxy=proxy, + timeout=timeout, + name="test_signal", + ) + await signal.connect() + status = signal.trigger() + await status + assert status.done is True and status.success is True # -------------------------------------------------------------------- @pytest.mark.asyncio @pytest.mark.skip("Not sure if we need tango_signal_auto") -@pytest.mark.parametrize( - "pv, tango_type, d_format, py_type, initial_value, put_value, use_proxy", - [ - ( - pv, - tango_type, - d_format, - py_type, - initial_value, - put_value, - use_proxy, - ) - for ( - pv, - tango_type, - d_format, - py_type, - initial_value, - put_value, - ) in ATTRIBUTES_SET - for use_proxy in [True, False] - ], - ids=[f"{x[0]}_{use_proxy}" for x in ATTRIBUTES_SET for use_proxy in [True, False]], -) async def test_tango_signal_auto_attrs( echo_device: str, - pv: str, - tango_type: str, - d_format: AttrDataFormat, - py_type: type[T], - initial_value: T, - put_value: T, - use_proxy: bool, ): - await prepare_device(echo_device, pv, initial_value) - source = echo_device + "/" + pv - proxy = await DeviceProxy(echo_device) if use_proxy else None timeout = 0.2 - - async def _test_signal(dtype, proxy): - signal = await __tango_signal_auto( - datatype=dtype, - trl=source, - device_proxy=proxy, - timeout=timeout, - name="test_signal", - ) - assert type(signal) is SignalRW - await signal.connect() - reading = await signal.read() - value = reading["test_signal"]["value"] - if isinstance(value, np.ndarray): - value = value.tolist() - assert_close(value, initial_value) - - await signal.set(put_value, wait=True, timeout=timeout) - reading = await signal.read() - value = reading["test_signal"]["value"] - if isinstance(value, np.ndarray): - value = value.tolist() - assert_close(value, put_value) - - dtype = py_type - await _test_signal(dtype, proxy) + for use_proxy in [True, False]: + proxy = await DeviceProxy(echo_device) if use_proxy else None + for attr_data in ATTRIBUTES_SET: + await prepare_device(echo_device, attr_data.pv, attr_data.initial_value) + source = echo_device + "/" + attr_data.pv + + async def _test_signal(dtype, proxy, source, initial_value, put_value): + signal = await __tango_signal_auto( + datatype=dtype, + trl=source, + device_proxy=proxy, + timeout=timeout, + name="test_signal", + ) + assert type(signal) is SignalRW + await signal.connect() + reading = await signal.read() + value = reading["test_signal"]["value"] + if isinstance(value, np.ndarray): + value = value.tolist() + assert_close(value, initial_value) + + await signal.set(put_value, wait=True, timeout=timeout) + reading = await signal.read() + value = reading["test_signal"]["value"] + if isinstance(value, np.ndarray): + value = value.tolist() + assert_close(value, put_value) + + await _test_signal( + attr_data.dtype, + proxy, + source, + attr_data.initial_value, + attr_data.put_value, + ) # -------------------------------------------------------------------- @pytest.mark.asyncio @pytest.mark.skip("Not sure if we need tango_signal_auto") @pytest.mark.parametrize( - "pv, tango_type, d_format, py_type, initial_value, put_value, use_dtype, use_proxy", + "use_dtype, use_proxy", [ - ( - pv, - tango_type, - d_format, - py_type, - initial_value, - put_value, - use_dtype, - use_proxy, - ) - for ( - pv, - tango_type, - d_format, - py_type, - initial_value, - put_value, - ) in COMMANDS_SET - for use_dtype in [True, False] - for use_proxy in [True, False] - ], - ids=[ - f"{x[0]}_{use_dtype}_{use_proxy}" - for x in COMMANDS_SET + (use_dtype, use_proxy) for use_dtype in [True, False] for use_proxy in [True, False] ], ) async def test_tango_signal_auto_cmds( echo_device: str, - pv: str, - tango_type: str, - d_format: AttrDataFormat, - py_type: type[T], - initial_value: T, - put_value: T, use_dtype: bool, use_proxy: bool, ): - source = echo_device + "/" + pv timeout = 0.2 + proxy = await DeviceProxy(echo_device) if use_proxy else None - async def _test_signal(dtype, proxy): - signal = await __tango_signal_auto( - datatype=dtype, - trl=source, - device_proxy=proxy, - name="test_signal", - timeout=timeout, - ) - # Ophyd SignalX does not support types - assert type(signal) in [SignalR, SignalRW, SignalW] - await signal.connect() - assert signal - reading = await signal.read() - assert reading["test_signal"]["value"] is None + for cmd_data in COMMANDS_SET: + source = echo_device + "/" + cmd_data.pv + + async def _test_signal(dtype, proxy, source, put_value): + signal = await __tango_signal_auto( + datatype=dtype, + trl=source, + device_proxy=proxy, + name="test_signal", + timeout=timeout, + ) + # Ophyd SignalX does not support types + assert type(signal) in [SignalR, SignalRW, SignalW] + await signal.connect() + assert signal + reading = await signal.read() + assert reading["test_signal"]["value"] is None - await signal.set(put_value, wait=True, timeout=0.1) - reading = await signal.read() - value = reading["test_signal"]["value"] - if isinstance(value, np.ndarray): - value = value.tolist() - assert_close(value, put_value) + await signal.set(put_value, wait=True, timeout=0.1) + reading = await signal.read() + value = reading["test_signal"]["value"] + if isinstance(value, np.ndarray): + value = value.tolist() + assert_close(value, put_value) - proxy = await DeviceProxy(echo_device) if use_proxy else None - dtype = py_type if use_dtype else None - await _test_signal(dtype, proxy) + dtype = cmd_data.py_type if use_dtype else None + await _test_signal(dtype, proxy, source, cmd_data.put_value) # -------------------------------------------------------------------- @pytest.mark.asyncio @pytest.mark.skip("Not sure if we need tango_signal_auto") -@pytest.mark.parametrize("use_proxy", [True, False]) async def test_tango_signal_auto_cmds_void(tango_test_device: str, use_proxy: bool): - proxy = await DeviceProxy(tango_test_device) if use_proxy else None - signal = await __tango_signal_auto( - datatype=None, - trl=tango_test_device + "/" + "clear", - device_proxy=proxy, - ) - assert type(signal) is SignalX - await signal.connect() - assert signal - await signal.trigger(wait=True) + for use_proxy in [True, False]: + proxy = await DeviceProxy(tango_test_device) if use_proxy else None + signal = await __tango_signal_auto( + datatype=None, + trl=tango_test_device + "/" + "clear", + device_proxy=proxy, + ) + assert type(signal) is SignalX + await signal.connect() + assert signal + await signal.trigger(wait=True) # -------------------------------------------------------------------- diff --git a/tests/tango/test_tango_transport.py b/tests/tango/test_tango_transport.py index 51048c4e9c..56387d3ef1 100644 --- a/tests/tango/test_tango_transport.py +++ b/tests/tango/test_tango_transport.py @@ -235,7 +235,6 @@ async def test_attribute_proxy_get(tango_test_device, attr): async def test_attribute_proxy_put(tango_test_device, attr, wait): device_proxy = await DeviceProxy(tango_test_device) attr_proxy = AttributeProxy(device_proxy, attr) - old_value = await attr_proxy.get() new_value = old_value + 1 status = await attr_proxy.put(new_value, wait=wait, timeout=0.1) @@ -244,7 +243,6 @@ async def test_attribute_proxy_put(tango_test_device, attr, wait): else: if not wait: raise AssertionError("If wait is False, put should return a status object") - await asyncio.sleep(1.0) updated_value = await attr_proxy.get() if isinstance(new_value, np.ndarray): assert np.all(updated_value == new_value)