From 82bfccfd72556cffdecf6c412a6e1b686f14c83c Mon Sep 17 00:00:00 2001 From: James Souter Date: Mon, 17 Feb 2025 15:56:49 +0000 Subject: [PATCH] Separate tango everything device server and client logic. Breaks tests requiring converters for tango commands --- .../tango/core/_tango_transport.py | 4 + src/ophyd_async/tango/core/_utils.py | 4 +- src/ophyd_async/tango/testing/__init__.py | 3 +- .../tango/testing/_one_of_everything.py | 281 ++++++++---------- tests/tango/conftest.py | 144 +++++++++ tests/tango/test_tango_signals.py | 269 ++++++++--------- 6 files changed, 395 insertions(+), 310 deletions(-) diff --git a/src/ophyd_async/tango/core/_tango_transport.py b/src/ophyd_async/tango/core/_tango_transport.py index 934740f4a9..49bc5a81fe 100644 --- a/src/ophyd_async/tango/core/_tango_transport.py +++ b/src/ophyd_async/tango/core/_tango_transport.py @@ -671,6 +671,10 @@ def make_converter(info: AttributeInfoEx | CommandInfo) -> TangoConverter: return TangoDevStateSpectrumConverter() elif info.data_format == AttrDataFormat.IMAGE: return TangoDevStateImageConverter() + else: # command info + match info.in_type: + case CmdArgType.DevState: + return TangoDevStateConverter() # default case return trivial converter return TangoConverter() diff --git a/src/ophyd_async/tango/core/_utils.py b/src/ophyd_async/tango/core/_utils.py index 874d19d98a..330a0d88b3 100644 --- a/src/ophyd_async/tango/core/_utils.py +++ b/src/ophyd_async/tango/core/_utils.py @@ -30,7 +30,9 @@ def get_full_attr_trl(device_trl: str, attr_name: str): def get_device_trl_and_attr(name: str): - # TODO put example of what the attr_name could look like here... + # trl can have form: + # domain/family/member/attr_name<#dbase=no> + # e.g. tango://127.0.0.1:8888/test/nodb/test#dbase=no re_str = ( r"([\.a-zA-Z0-9_-]*://)?([\.a-zA-Z0-9_-]+:[0-9]+/)?" r"([^#/]+/[^#/]+/[^#/]+/)([^#/]+)(#dbase=[a-z]+)?" diff --git a/src/ophyd_async/tango/testing/__init__.py b/src/ophyd_async/tango/testing/__init__.py index a09d0e0fda..7010c582c1 100644 --- a/src/ophyd_async/tango/testing/__init__.py +++ b/src/ophyd_async/tango/testing/__init__.py @@ -1,7 +1,6 @@ from ._one_of_everything import ( ExampleStrEnum, OneOfEverythingTangoDevice, - everything_signal_info, ) -__all__ = ["ExampleStrEnum", "OneOfEverythingTangoDevice", "everything_signal_info"] +__all__ = ["ExampleStrEnum", "OneOfEverythingTangoDevice"] diff --git a/src/ophyd_async/tango/testing/_one_of_everything.py b/src/ophyd_async/tango/testing/_one_of_everything.py index 577c3a9a76..5e4007b8f5 100644 --- a/src/ophyd_async/tango/testing/_one_of_everything.py +++ b/src/ophyd_async/tango/testing/_one_of_everything.py @@ -1,6 +1,5 @@ import textwrap from dataclasses import dataclass -from random import choice from typing import Any, Generic, TypeVar import numpy as np @@ -10,7 +9,6 @@ DTypeScalar_co, StrictEnum, ) -from ophyd_async.tango.core import DevStateEnum from ophyd_async.testing import float_array_value, int_array_value from tango import AttrDataFormat, AttrWriteType, DevState from tango.server import Device, attribute, command @@ -40,171 +38,144 @@ def float_image_value( return np.vstack((array_1d, array_1d)) +def _valid_command(dformat: AttrDataFormat, dtype: str): + if dtype == "DevUChar": + return False + if dformat != AttrDataFormat.SCALAR and dtype in ["DevState", "DevEnum"]: + return False + return True + + @dataclass class AttributeData(Generic[T]): name: str tango_type: str - py_type: type - initial_value: T - random_put_values: tuple[T, ...] - dformat = AttrDataFormat.SCALAR - cmd_name: str | None - - def random_value(self): - return choice(self.random_put_values) - - -class SpectrumData(AttributeData): - dformat = AttrDataFormat.SPECTRUM - - def random_value(self): - array = self.initial_value.copy() - for idx in range(len(array)): - array[idx] = choice(self.random_put_values) - return array - - -class ImageData(AttributeData): - dformat = AttrDataFormat.IMAGE - cmd_name = None + initial_scalar: T + initial_spectrum: Array1D + + +_all_attribute_definitions = [ + AttributeData( + "str", + "DevString", + "test_string", + np.array(["one", "two", "three"], dtype=str), + ), + AttributeData( + "bool", + "DevBoolean", + True, + np.array([False, True], dtype=bool), + ), + AttributeData("strenum", "DevEnum", 1, np.array([0, 1, 2])), + AttributeData("int8", "DevShort", 1, int_array_value(np.int8)), + AttributeData("uint8", "DevUChar", 1, int_array_value(np.uint8)), + AttributeData("int16", "DevShort", 1, int_array_value(np.int16)), + AttributeData("uint16", "DevUShort", 1, int_array_value(np.uint16)), + AttributeData("int32", "DevLong", 1, int_array_value(np.int32)), + AttributeData("uint32", "DevULong", 1, int_array_value(np.uint32)), + AttributeData("int64", "DevLong64", 1, int_array_value(np.int64)), + AttributeData("uint64", "DevULong64", 1, int_array_value(np.uint64)), + AttributeData("float32", "DevFloat", 1.234, float_array_value(np.float32)), + AttributeData("float64", "DevDouble", 1.234, float_array_value(np.float64)), + AttributeData( + "my_state", + "DevState", + DevState.INIT, + np.array([DevState.INIT, DevState.ON, DevState.MOVING], dtype=DevState), + ), +] - def random_value(self): - array = self.initial_value.copy() - for idx in range(array.shape[1]): - array[0, idx] = choice(self.random_put_values) - array[1, idx] = choice(self.random_put_values) - return array - -everything_signal_info = [] - - -def add_ads( - name: str, - tango_type: str, - py_type: type, - initial_scalar, - initial_spectrum, - choices, -): - scalar_cmd = f"{name}_cmd" if tango_type != "DevUChar" else None - everything_signal_info.append( - AttributeData(name, tango_type, py_type, initial_scalar, choices, scalar_cmd) - ) - spectrum_cmd = ( - f"{name}_spectrum_cmd" - if tango_type not in ["DevUChar", "DevState", "DevEnum"] - else None - ) - everything_signal_info.append( - SpectrumData( - f"{name}_spectrum", - tango_type, - Array1D[py_type], - initial_spectrum, - choices, - spectrum_cmd, +class OneOfEverythingTangoDevice(Device): + attr_values = {} + initial_values = {} + + def _add_attr(self, attr: attribute, initial_value): + self.attr_values[attr.name] = initial_value + self.initial_values[attr.name] = initial_value + self.add_attribute(attr) + self.set_change_event(attr.name, True, False) + + def add_scalar_attr(self, name: str, dtype: str, initial_value: Any): + attr = attribute( + name=name, + dtype=dtype, + dformat=AttrDataFormat.SCALAR, + access=AttrWriteType.READ_WRITE, + fget=self.read, + fset=self.write, + enum_labels=[e.value for e in ExampleStrEnum], ) - ) - everything_signal_info.append( - ImageData( - f"{name}_image", - tango_type, - np.ndarray[Any, np.dtype[py_type]], - np.vstack((initial_spectrum, initial_spectrum)), - choices, - None, + self._add_attr(attr, initial_value) + + def add_array_attrs(self, name: str, dtype: str, initial_value: np.ndarray): + spectrum_name = f"{name}_spectrum" + spectrum_attr = attribute( + name=spectrum_name, + dtype=dtype, + dformat=AttrDataFormat.SPECTRUM, + access=AttrWriteType.READ_WRITE, + fget=self.read, + fset=self.write, + max_dim_x=initial_value.shape[-1], + enum_labels=[e.value for e in ExampleStrEnum], ) - ) - - -add_ads( - "str", - "DevString", - str, - "test_string", - np.array(["one", "two", "three"], dtype=str), - ("four", "five", "six"), -) -add_ads( - "bool", - "DevBoolean", - bool, - True, - np.array([False, True], dtype=bool), - (False, True), -) -add_ads("strenum", "DevEnum", StrictEnum, 1, np.array([0, 1, 2]), (0, 1, 2)) -add_ads("int8", "DevShort", int, 1, int_array_value(np.int8), (1, 2, 3, 4, 5)) -add_ads("uint8", "DevUChar", int, 1, int_array_value(np.uint8), (1, 2, 3, 4, 5)) -add_ads("int16", "DevShort", int, 1, int_array_value(np.int16), (1, 2, 3, 4, 5)) -add_ads("uint16", "DevUShort", int, 1, int_array_value(np.uint16), (1, 2, 3, 4, 5)) -add_ads("int32", "DevLong", int, 1, int_array_value(np.int32), (1, 2, 3, 4, 5)) -add_ads("uint32", "DevULong", int, 1, int_array_value(np.uint32), (1, 2, 3, 4, 5)) -add_ads("int64", "DevLong64", int, 1, int_array_value(np.int64), (1, 2, 3, 4, 5)) -add_ads("uint64", "DevULong64", int, 1, int_array_value(np.uint64), (1, 2, 3, 4, 5)) -add_ads( - "float32", - "DevFloat", - float, - 1.234, - float_array_value(np.float32), - (1.234, 2.345, 3.456), -) -add_ads( - "float64", - "DevDouble", - float, - 1.234, - float_array_value(np.float64), - (1.234, 2.345, 3.456), -) -add_ads( - "my_state", - "DevState", - DevStateEnum, - DevState.INIT, - np.array([DevState.INIT, DevState.ON, DevState.MOVING], dtype=DevState), - (DevState.INIT, DevState.ON, DevState.MOVING), -) - + image_name = f"{name}_image" + image_attr = attribute( + name=image_name, + dtype=dtype, + dformat=AttrDataFormat.IMAGE, + access=AttrWriteType.READ_WRITE, + fget=self.read, + fset=self.write, + max_dim_x=initial_value.shape[-1], + max_dim_y=2, + enum_labels=[e.value for e in ExampleStrEnum], + ) + self._add_attr(spectrum_attr, initial_value) + # have image just be 2 of the initial spectrum stacked + self._add_attr(image_attr, np.vstack((initial_value, initial_value))) + + def add_scalar_command(self, name: str, dtype: str): + if _valid_command(AttrDataFormat.SCALAR, dtype): + self.add_command( + command( + f=getattr(self, f"{name}_cmd"), + dtype_in=dtype, + dtype_out=dtype, + dformat_in=AttrDataFormat.SCALAR, + dformat_out=AttrDataFormat.SCALAR, + ), + ) -class OneOfEverythingTangoDevice(Device): - attr_values = {} + def add_spectrum_command(self, name: str, dtype: str): + if _valid_command(AttrDataFormat.SPECTRUM, dtype): + self.add_command( + command( + f=getattr(self, f"{name}_spectrum_cmd"), + dtype_in=dtype, + dtype_out=dtype, + dformat_in=AttrDataFormat.SPECTRUM, + dformat_out=AttrDataFormat.SPECTRUM, + ), + ) def initialize_dynamic_attributes(self): - self.reset_values() - for attr_data in everything_signal_info: - attr = attribute( - name=attr_data.name, - dtype=attr_data.tango_type, - dformat=attr_data.dformat, - access=AttrWriteType.READ_WRITE, - fget=self.read, - fset=self.write, - max_dim_x=0 - if attr_data.dformat == AttrDataFormat.SCALAR - else attr_data.initial_value.shape[-1], - max_dim_y=2, - enum_labels=[e.value for e in ExampleStrEnum], + for attr_data in _all_attribute_definitions: + self.add_scalar_attr( + attr_data.name, attr_data.tango_type, attr_data.initial_scalar + ) + self.add_array_attrs( + attr_data.name, attr_data.tango_type, attr_data.initial_spectrum ) - self.add_attribute(attr) - self.set_change_event(attr.name, True, False) - if attr_data.cmd_name: - self.add_command( - command( - f=getattr(self, attr_data.cmd_name), - dtype_in=attr_data.tango_type, - dtype_out=attr_data.tango_type, - dformat_in=attr_data.dformat, - dformat_out=attr_data.dformat, - ) - ) + self.add_scalar_command(attr_data.name, attr_data.tango_type) + self.add_spectrum_command(attr_data.name, attr_data.tango_type) @command def reset_values(self): - for attr_data in everything_signal_info: - self.attr_values[attr_data.name] = attr_data.initial_value + for attr_name in self.attr_values: + self.attr_values[attr_name] = self.initial_values[attr_name] def read(self, attr): value = self.attr_values[attr.get_name()] @@ -222,6 +193,8 @@ def {}(self, arg): """ ) - for attr_data in everything_signal_info: - if attr_data.dformat != AttrDataFormat.IMAGE: + for attr_data in _all_attribute_definitions: + if _valid_command(AttrDataFormat.SCALAR, attr_data.tango_type): exec(echo_command_code.format(f"{attr_data.name}_cmd")) + if _valid_command(AttrDataFormat.SPECTRUM, attr_data.tango_type): + exec(echo_command_code.format(f"{attr_data.name}_spectrum_cmd")) diff --git a/tests/tango/conftest.py b/tests/tango/conftest.py index 5dde981e7e..ec0f406955 100644 --- a/tests/tango/conftest.py +++ b/tests/tango/conftest.py @@ -3,12 +3,25 @@ import socket import subprocess import sys +from dataclasses import dataclass from pathlib import Path +from random import choice +from typing import Any, Generic, TypeVar +import numpy as np import pytest +from ophyd_async.core import Array1D +from ophyd_async.tango.core import DevStateEnum +from ophyd_async.tango.testing import ExampleStrEnum +from ophyd_async.testing import ( + float_array_value, + int_array_value, +) from tango.asyncio_executor import set_global_executor +T = TypeVar("T") + @pytest.fixture(autouse=True) def reset_tango_asyncio(): @@ -52,3 +65,134 @@ def __exit__(self, A, B, C): @pytest.fixture(scope="module") def subprocess_helper(): return TangoSubprocessHelper + + +@dataclass +class AttributeData(Generic[T]): + name: str + py_type: type + initial: T + random_put_values: tuple[T, ...] + cmd_name: str | None + + def random_value(self): + return choice(self.random_put_values) + + +class SpectrumData(AttributeData): + def random_value(self): + array = self.initial.copy() + for idx in range(len(array)): + array[idx] = choice(self.random_put_values) + return array + + +class ImageData(AttributeData): + cmd_name = None + + def random_value(self): + array = self.initial.copy() + for idx in range(array.shape[1]): + array[0, idx] = choice(self.random_put_values) + array[1, idx] = choice(self.random_put_values) + return array + + +@pytest.fixture(scope="module") +def everything_signal_info(): + signal_info = {} + + def add_ads( + name: str, + tango_type: str, + py_type: type, + initial_scalar, + initial_spectrum, + choices, + ): + scalar_cmd = f"{name}_cmd" if tango_type != "DevUChar" else None + signal_info[name] = AttributeData( + name, py_type, initial_scalar, choices, scalar_cmd + ) + spectrum_cmd = ( + f"{name}_spectrum_cmd" + if tango_type not in ["DevUChar", "DevState", "DevEnum"] + else None + ) + signal_info[f"{name}_spectrum"] = SpectrumData( + f"{name}_spectrum", + Array1D[py_type], + initial_spectrum, + choices, + spectrum_cmd, + ) + signal_info[f"{name}_image"] = ImageData( + f"{name}_image", + np.ndarray[Any, np.dtype[py_type]], + np.vstack((initial_spectrum, initial_spectrum)), + choices, + None, + ) + + add_ads( + "str", + "DevString", + str, + "test_string", + np.array(["one", "two", "three"], dtype=str), + ("four", "five", "six"), + ) + add_ads( + "bool", + "DevBoolean", + bool, + True, + np.array([False, True], dtype=bool), + (False, True), + ) + add_ads( + "strenum", + "DevEnum", + ExampleStrEnum, + ExampleStrEnum.B, + np.array( + [ExampleStrEnum.A.value, ExampleStrEnum.B.value, ExampleStrEnum.C.value], + ), + (ExampleStrEnum.A.value, ExampleStrEnum.B.value, ExampleStrEnum.C.value), + ) + add_ads("int8", "DevShort", int, 1, int_array_value(np.int8), (1, 2, 3, 4, 5)) + add_ads("uint8", "DevUChar", int, 1, int_array_value(np.uint8), (1, 2, 3, 4, 5)) + add_ads("int16", "DevShort", int, 1, int_array_value(np.int16), (1, 2, 3, 4, 5)) + add_ads("uint16", "DevUShort", int, 1, int_array_value(np.uint16), (1, 2, 3, 4, 5)) + add_ads("int32", "DevLong", int, 1, int_array_value(np.int32), (1, 2, 3, 4, 5)) + add_ads("uint32", "DevULong", int, 1, int_array_value(np.uint32), (1, 2, 3, 4, 5)) + add_ads("int64", "DevLong64", int, 1, int_array_value(np.int64), (1, 2, 3, 4, 5)) + add_ads("uint64", "DevULong64", int, 1, int_array_value(np.uint64), (1, 2, 3, 4, 5)) + add_ads( + "float32", + "DevFloat", + float, + 1.234, + float_array_value(np.float32), + (1.234, 2.345, 3.456), + ) + add_ads( + "float64", + "DevDouble", + float, + 1.234, + float_array_value(np.float64), + (1.234, 2.345, 3.456), + ) + add_ads( + "my_state", + "DevState", + DevStateEnum, + DevStateEnum.INIT, + np.array( # TODO: make this work without the .values + [DevStateEnum.INIT.value, DevStateEnum.ON.value, DevStateEnum.MOVING.value] + ), + (DevStateEnum.INIT.value, DevStateEnum.ON.value, DevStateEnum.MOVING.value), + ) + + return signal_info diff --git a/tests/tango/test_tango_signals.py b/tests/tango/test_tango_signals.py index 815cb5412a..79fe9fc967 100644 --- a/tests/tango/test_tango_signals.py +++ b/tests/tango/test_tango_signals.py @@ -21,7 +21,6 @@ from ophyd_async.tango.testing import ( ExampleStrEnum, OneOfEverythingTangoDevice, - everything_signal_info, ) from ophyd_async.testing import MonitorQueue, assert_reading, assert_value @@ -110,39 +109,33 @@ async def assert_monitor_then_put( descriptor: dict, ): backend = signal._connector.backend - converter = signal._connector.backend.converter # type: ignore - converted_put = converter.value(put_value) # Make a monitor queue that will monitor for updates with MonitorQueue(signal) as q: assert dict(source=source, **descriptor) == await backend.get_datakey("") # Check initial value await q.assert_updates(initial_value) # Put to new value and check that - await backend.put(converted_put) - await q.assert_updates(converted_put) + await backend.put(put_value) + await q.assert_updates(put_value) # -------------------------------------------------------------------- @pytest.mark.asyncio -async def test_backend_get_put_monitor_attr(everything_device: TangoDevice): +async def test_backend_get_put_monitor_attr( + everything_device: TangoDevice, everything_signal_info +): await everything_device.connect() try: - for attr_data in everything_signal_info: + for attr_data in everything_signal_info.values(): signal = getattr(everything_device, attr_data.name) source = get_full_attr_trl(everything_device._connector.trl, attr_data.name) - initial = attr_data.initial_value - if "my_state" in attr_data.name or "strenum" in attr_data.name: - # signal_info initial_values use datatype that works on server backend - initial = signal._connector.backend.converter.value(initial) await asyncio.wait_for( assert_monitor_then_put( signal, source, - initial, + attr_data.initial, attr_data.random_value(), - get_test_descriptor( - attr_data.py_type, attr_data.initial_value, False - ), + get_test_descriptor(attr_data.py_type, attr_data.initial, False), ), timeout=10, # Timeout in seconds ) @@ -158,7 +151,7 @@ async def assert_put_read( source: str, put_value: T, descriptor: dict, - datatype: type[T] | None = None, + datatype: type[T] | None = None, # TODO reimplement this ): backend = signal._connector.backend # Make a monitor queue that will monitor for updates @@ -178,18 +171,22 @@ async def assert_put_read( # -------------------------------------------------------------------- @pytest.mark.asyncio -async def test_backend_get_put_monitor_cmd(everything_device: TangoDevice): +async def test_backend_get_put_monitor_cmd( + everything_device: TangoDevice, everything_signal_info +): await everything_device.connect() - for cmd_data in everything_signal_info: + for cmd_data in everything_signal_info.values(): if cmd_data.cmd_name is None: continue + put_value = cmd_data.random_value() # With the given datatype, check we have the correct initial value + if "strenum" in cmd_data.cmd_name: + # enum commands do not provide labels so we have to send the int index + put_value = [e.value for e in ExampleStrEnum].index(put_value) # and putting works - put_value = cmd_data.random_value() - name = f"{cmd_data.name}_cmd" - descriptor = get_test_descriptor(cmd_data.py_type, cmd_data.initial_value, True) - signal = getattr(everything_device, name) - source = get_full_attr_trl(everything_device._connector.trl, name) + descriptor = get_test_descriptor(cmd_data.py_type, cmd_data.initial, True) + signal = getattr(everything_device, cmd_data.cmd_name) + source = get_full_attr_trl(everything_device._connector.trl, cmd_data.cmd_name) await assert_put_read(signal, source, put_value, descriptor, cmd_data.py_type) # # With guessed datatype, check we can set it back to the initial value await assert_put_read(signal, source, put_value, descriptor) @@ -199,11 +196,9 @@ async def test_backend_get_put_monitor_cmd(everything_device: TangoDevice): # -------------------------------------------------------------------- @pytest.mark.asyncio -async def test_tango_signal_r( - everything_device_trl: str, -): +async def test_tango_signal_r(everything_device_trl: str, everything_signal_info): timeout = 0.2 - for attr_data in everything_signal_info: + for attr_data in everything_signal_info.values(): source = get_full_attr_trl(everything_device_trl, attr_data.name) signal = tango_signal_r( datatype=attr_data.py_type, @@ -222,11 +217,9 @@ async def test_tango_signal_r( # -------------------------------------------------------------------- @pytest.mark.asyncio -async def test_tango_signal_w( - everything_device_trl: str, -): +async def test_tango_signal_w(everything_device_trl: str, everything_signal_info): timeout = 0.2 - for attr_data in everything_signal_info: + for attr_data in everything_signal_info.values(): source = get_full_attr_trl(everything_device_trl, attr_data.name) signal = tango_signal_w( datatype=attr_data.py_type, @@ -235,13 +228,12 @@ async def test_tango_signal_w( name="test_signal", ) await signal.connect() # have to connect to get correct converter - converter = signal._connector.backend.converter # type: ignore - put_value = converter.value(attr_data.random_value()) + put_value = attr_data.random_value() + status = signal.set(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 @@ -257,11 +249,9 @@ async def test_tango_signal_w( # -------------------------------------------------------------------- @pytest.mark.asyncio -async def test_tango_signal_rw( - everything_device_trl: str, -): +async def test_tango_signal_rw(everything_device_trl: str, everything_signal_info): timeout = 0.2 - for attr_data in everything_signal_info: + for attr_data in everything_signal_info.values(): source = get_full_attr_trl(everything_device_trl, attr_data.name) signal = tango_signal_rw( datatype=attr_data.py_type, @@ -271,8 +261,7 @@ async def test_tango_signal_rw( name="test_signal", ) await signal.connect() - converter = signal._connector.backend.converter # type: ignore - put_value = converter.value(attr_data.random_value()) + put_value = attr_data.random_value() await signal.set(put_value, wait=True) await assert_value(signal, put_value) @@ -292,75 +281,6 @@ async def test_tango_signal_x(tango_test_device: str): assert status.done is True and status.success is True -_scalar_vals = { - "str": "test_string", - "bool": True, - "strenum": ExampleStrEnum.B, - "int8": 1, - "uint8": 1, - "int16": 1, - "uint16": 1, - "int32": 1, - "uint32": 1, - "int64": 1, - "uint64": 1, - "float32": 1.234, - "float64": 1.234, - "my_state": DevStateEnum.INIT, -} -_array_vals = { - "int8": np.array([-128, 127, 0, 1, 2, 3, 4], dtype=np.int8), - "uint8": np.array([0, 255, 0, 1, 2, 3, 4], dtype=np.uint8), - "int16": np.array([-32768, 32767, 0, 1, 2, 3, 4], dtype=np.int16), - "uint16": np.array([0, 65535, 0, 1, 2, 3, 4], dtype=np.uint16), - "int32": np.array([-2147483648, 2147483647, 0, 1, 2, 3, 4], dtype=np.int32), - "uint32": np.array([0, 4294967295, 0, 1, 2, 3, 4], dtype=np.uint32), - "int64": np.array( - [-9223372036854775808, 9223372036854775807, 0, 1, 2, 3, 4], - dtype=np.int64, - ), - "uint64": np.array([0, 18446744073709551615, 0, 1, 2, 3, 4], dtype=np.uint64), - "float32": np.array( - [ - -3.4028235e38, - 3.4028235e38, - 1.1754944e-38, - 1.4012985e-45, - 0, - 1.234, - 2.34e5, - 3.45e-6, - ], - dtype=np.float32, - ), - "float64": np.array( - [ - -1.79769313e308, - 1.79769313e308, - 2.22507386e-308, - 4.94065646e-324, - 0, - 1.234, - 2.34e5, - 3.45e-6, - ], - dtype=np.float64, - ), - "strenum": np.array( - [ExampleStrEnum.A.value, ExampleStrEnum.B.value, ExampleStrEnum.C.value], - dtype=str, - ), - "str": ["one", "two", "three"], - "bool": np.array([False, True]), - "my_state": np.array( - [DevStateEnum.INIT.value, DevStateEnum.ON.value, DevStateEnum.MOVING.value], - dtype=str, - ), -} - -_image_vals = {k: np.vstack((v, v)) for k, v in _array_vals.items()} - - async def assert_val_reading(signal, value, name=""): await assert_value(signal, value) await assert_reading(signal, {name: {"value": value}}) @@ -442,54 +362,97 @@ async def test_set_with_converter(everything_device_trl): ) -async def test_assert_val_reading_everything_tango(everything_device_trl): +async def test_assert_val_reading_everything_tango( + everything_device_trl, everything_signal_info +): + esi = everything_signal_info everything_device = TangoDevice(everything_device_trl) await everything_device.connect() await everything_device.reset_values.trigger() await asyncio.sleep(1) - await assert_val_reading(everything_device.str, _scalar_vals["str"]) - await assert_val_reading(everything_device.bool, _scalar_vals["bool"]) - await assert_val_reading(everything_device.strenum, _scalar_vals["strenum"]) - await assert_val_reading(everything_device.int8, _scalar_vals["int8"]) - await assert_val_reading(everything_device.uint8, _scalar_vals["uint8"]) - await assert_val_reading(everything_device.int16, _scalar_vals["int16"]) - await assert_val_reading(everything_device.uint16, _scalar_vals["uint16"]) - await assert_val_reading(everything_device.int32, _scalar_vals["int32"]) - await assert_val_reading(everything_device.uint32, _scalar_vals["uint32"]) - await assert_val_reading(everything_device.int64, _scalar_vals["int64"]) - await assert_val_reading(everything_device.uint64, _scalar_vals["uint64"]) - await assert_val_reading(everything_device.float32, _scalar_vals["float32"]) - await assert_val_reading(everything_device.float64, _scalar_vals["float64"]) - await assert_val_reading(everything_device.my_state, _scalar_vals["my_state"]) - - await assert_val_reading(everything_device.str_spectrum, _array_vals["str"]) - await assert_val_reading(everything_device.bool_spectrum, _array_vals["bool"]) - await assert_val_reading(everything_device.strenum_spectrum, _array_vals["strenum"]) - await assert_val_reading(everything_device.int8_spectrum, _array_vals["int8"]) - await assert_val_reading(everything_device.uint8_spectrum, _array_vals["uint8"]) - await assert_val_reading(everything_device.int16_spectrum, _array_vals["int16"]) - await assert_val_reading(everything_device.uint16_spectrum, _array_vals["uint16"]) - await assert_val_reading(everything_device.int32_spectrum, _array_vals["int32"]) - await assert_val_reading(everything_device.uint32_spectrum, _array_vals["uint32"]) - await assert_val_reading(everything_device.int64_spectrum, _array_vals["int64"]) - await assert_val_reading(everything_device.uint64_spectrum, _array_vals["uint64"]) - await assert_val_reading(everything_device.float32_spectrum, _array_vals["float32"]) - await assert_val_reading(everything_device.float64_spectrum, _array_vals["float64"]) + await assert_val_reading(everything_device.str, esi["str"].initial) + await assert_val_reading(everything_device.bool, esi["bool"].initial) + await assert_val_reading(everything_device.strenum, esi["strenum"].initial) + await assert_val_reading(everything_device.int8, esi["int8"].initial) + await assert_val_reading(everything_device.uint8, esi["uint8"].initial) + await assert_val_reading(everything_device.int16, esi["int16"].initial) + await assert_val_reading(everything_device.uint16, esi["uint16"].initial) + await assert_val_reading(everything_device.int32, esi["int32"].initial) + await assert_val_reading(everything_device.uint32, esi["uint32"].initial) + await assert_val_reading(everything_device.int64, esi["int64"].initial) + await assert_val_reading(everything_device.uint64, esi["uint64"].initial) + await assert_val_reading(everything_device.float32, esi["float32"].initial) + await assert_val_reading(everything_device.float64, esi["float64"].initial) + await assert_val_reading(everything_device.my_state, esi["my_state"].initial) + + await assert_val_reading( + everything_device.str_spectrum, esi["str_spectrum"].initial + ) + await assert_val_reading( + everything_device.bool_spectrum, esi["bool_spectrum"].initial + ) + await assert_val_reading( + everything_device.strenum_spectrum, esi["strenum_spectrum"].initial + ) await assert_val_reading( - everything_device.my_state_spectrum, _array_vals["my_state"] + everything_device.int8_spectrum, esi["int8_spectrum"].initial + ) + await assert_val_reading( + everything_device.uint8_spectrum, esi["uint8_spectrum"].initial + ) + await assert_val_reading( + everything_device.int16_spectrum, esi["int16_spectrum"].initial + ) + await assert_val_reading( + everything_device.uint16_spectrum, esi["uint16_spectrum"].initial + ) + await assert_val_reading( + everything_device.int32_spectrum, esi["int32_spectrum"].initial + ) + await assert_val_reading( + everything_device.uint32_spectrum, esi["uint32_spectrum"].initial + ) + await assert_val_reading( + everything_device.int64_spectrum, esi["int64_spectrum"].initial + ) + await assert_val_reading( + everything_device.uint64_spectrum, esi["uint64_spectrum"].initial + ) + await assert_val_reading( + everything_device.float32_spectrum, esi["float32_spectrum"].initial + ) + await assert_val_reading( + everything_device.float64_spectrum, esi["float64_spectrum"].initial + ) + await assert_val_reading( + everything_device.my_state_spectrum, esi["my_state_spectrum"].initial ) - await assert_val_reading(everything_device.str_image, _image_vals["str"]) - await assert_val_reading(everything_device.bool_image, _image_vals["bool"]) - await assert_val_reading(everything_device.strenum_image, _image_vals["strenum"]) - await assert_val_reading(everything_device.int8_image, _image_vals["int8"]) - await assert_val_reading(everything_device.uint8_image, _image_vals["uint8"]) - await assert_val_reading(everything_device.int16_image, _image_vals["int16"]) - await assert_val_reading(everything_device.uint16_image, _image_vals["uint16"]) - await assert_val_reading(everything_device.int32_image, _image_vals["int32"]) - await assert_val_reading(everything_device.uint32_image, _image_vals["uint32"]) - await assert_val_reading(everything_device.int64_image, _image_vals["int64"]) - await assert_val_reading(everything_device.uint64_image, _image_vals["uint64"]) - await assert_val_reading(everything_device.float32_image, _image_vals["float32"]) - await assert_val_reading(everything_device.float64_image, _image_vals["float64"]) - await assert_val_reading(everything_device.my_state_image, _image_vals["my_state"]) + await assert_val_reading(everything_device.str_image, esi["str_image"].initial) + await assert_val_reading(everything_device.bool_image, esi["bool_image"].initial) + await assert_val_reading( + everything_device.strenum_image, esi["strenum_image"].initial + ) + await assert_val_reading(everything_device.int8_image, esi["int8_image"].initial) + await assert_val_reading(everything_device.uint8_image, esi["uint8_image"].initial) + await assert_val_reading(everything_device.int16_image, esi["int16_image"].initial) + await assert_val_reading( + everything_device.uint16_image, esi["uint16_image"].initial + ) + await assert_val_reading(everything_device.int32_image, esi["int32_image"].initial) + await assert_val_reading( + everything_device.uint32_image, esi["uint32_image"].initial + ) + await assert_val_reading(everything_device.int64_image, esi["int64_image"].initial) + await assert_val_reading( + everything_device.uint64_image, esi["uint64_image"].initial + ) + await assert_val_reading( + everything_device.float32_image, esi["float32_image"].initial + ) + await assert_val_reading( + everything_device.float64_image, esi["float64_image"].initial + ) + await assert_val_reading( + everything_device.my_state_image, esi["my_state_image"].initial + )