diff --git a/lofar_station_client/devices.py b/lofar_station_client/devices.py index 60e4d567239a9b47a11d7fecbce6dfd7ec6819f0..0f9c2ee0a6186b0a514169e8836449652e07c56e 100644 --- a/lofar_station_client/devices.py +++ b/lofar_station_client/devices.py @@ -1,3 +1,5 @@ +""" Enhanced interfaces towards the station's Tango devices. """ + import ast from functools import lru_cache @@ -11,13 +13,15 @@ class LofarDeviceProxy(DeviceProxy): """A LOFAR-specific tango.DeviceProxy that provides a richer experience.""" - # cache attribute configurations, as they are not expected to change, - # but carry a cost to retrieve from the server. @lru_cache() def get_attribute_config(self, name): + """Get cached attribute configurations, as they are not expected to change.""" + return super().get_attribute_config(name) def read_attribute(self, name, extract_as=ExtractAs.Numpy): + """Read an attribute from the server.""" + attr = super().read_attribute(name, extract_as) # "format" property describes actual dimensions as a tuple (x, y, z, ...), @@ -30,11 +34,14 @@ class LofarDeviceProxy(DeviceProxy): return attr def write_attribute(self, name, value): + """Write an attribute to the server.""" + config = self.get_attribute_config(name) # 2D arrays also represent arrays of higher dimensionality. reshape them # to fit their original Tango shape before writing. if config.max_dim_y > 0: + value = numpy.array(value) value = value.reshape((config.max_dim_y, config.max_dim_x)) return super().write_attribute(name, value) diff --git a/tests/test_devices.py b/tests/test_devices.py new file mode 100644 index 0000000000000000000000000000000000000000..1a710e2f22f8784ef9187550f6b24c36f126fd4d --- /dev/null +++ b/tests/test_devices.py @@ -0,0 +1,71 @@ +import numpy + +from tests import base + +from lofar_station_client.devices import LofarDeviceProxy + +from tango.test_context import MultiDeviceTestContext +from tango.server import Device, attribute, AttrWriteType + + +class MyDevice(Device): + A = attribute( + dtype=((bool,),), + max_dim_x=4, + max_dim_y=6, + format="(2,3,4)", + access=AttrWriteType.READ_WRITE, + ) + + def init_device(self): + self.value_A = numpy.zeros((6, 4), dtype=bool) + + def read_A(self): + return self.value_A + + def write_A(self, value): + self.value_A = value + + +class LofarDeviceProxyTest(base.TestCase): + TEST_DEVICE_INFO = [ + { + "class": MyDevice, + "devices": [{"name": "STAT/MyDevice/1", "properties": {}, "memorized": {}}], + } + ] + + def test_write_3D_attribute_lists(self): + with MultiDeviceTestContext( + self.TEST_DEVICE_INFO, + process=True, + ) as context: + proxy = LofarDeviceProxy(context.get_device_access("STAT/MyDevice/1")) + + proxy.A = [ + [True, True, True, True], + [True, True, True, True], + [True, True, True, True], + ], [ + [False, False, False, False], + [False, False, False, False], + [False, False, False, False], + ] + + def test_write_3D_attribute_numpy(self): + with MultiDeviceTestContext( + self.TEST_DEVICE_INFO, + process=True, + ) as context: + proxy = LofarDeviceProxy(context.get_device_access("STAT/MyDevice/1")) + + proxy.A = numpy.zeros((2, 3, 4), dtype=bool) + + def test_read_3D_attribute(self): + with MultiDeviceTestContext( + self.TEST_DEVICE_INFO, + process=True, + ) as context: + proxy = LofarDeviceProxy(context.get_device_access("STAT/MyDevice/1")) + + self.assertEqual((2, 3, 4), proxy.A.shape)