diff --git a/tangostationcontrol/tangostationcontrol/devices/lofar_device.py b/tangostationcontrol/tangostationcontrol/devices/lofar_device.py
index c04033f343e5ffb118fb31931b60262a6279c9d1..068c9cee675f57f8a503fe8f4036bc2e98b80433 100644
--- a/tangostationcontrol/tangostationcontrol/devices/lofar_device.py
+++ b/tangostationcontrol/tangostationcontrol/devices/lofar_device.py
@@ -11,14 +11,18 @@
"""
-# PyTango imports
-from tango.server import attribute, command, Device, DeviceMeta
-from tango import AttrWriteType, DevState, DebugIt, Attribute, DeviceProxy, AttrDataFormat, DevSource, DevDouble
+from collections.abc import Sequence
import time
import math
+from typing import List
+
import numpy
import textwrap
+# PyTango imports
+from tango.server import attribute, command, Device, DeviceMeta
+from tango import AttrWriteType, DevState, DebugIt, Attribute, DeviceProxy, AttrDataFormat, DevSource, DevDouble
+
# Additional import
from tangostationcontrol import __version__ as version
from tangostationcontrol.clients.attribute_wrapper import attribute_wrapper
@@ -34,6 +38,14 @@ import logging
logger = logging.getLogger()
+# TODO(Corne): Remove this in L2SS-940
+def sequence_not_str(obj):
+ """Separate sequences / collections from str, byte or bytearray"""
+
+ return (isinstance(obj, Sequence) or isinstance(obj, numpy.ndarray)) and not \
+ isinstance(obj, (str, bytes, bytearray))
+
+
class lofar_device(Device, metaclass=DeviceMeta):
"""
@@ -92,6 +104,59 @@ class lofar_device(Device, metaclass=DeviceMeta):
return self.get_state() in INITIALISED_STATES
+ # TODO(Corne): Actually implement locking in L2SS-940
+ def atomic_read_modify_write_attribute(
+ self, values: numpy.ndarray, proxy: DeviceProxy, attribute: str, sparse=None
+ ):
+ """Atomatically read-modify-write the attribute on the given proxy"""
+
+ current_values = proxy.read_attribute(attribute).value
+ self.merge_write(values, current_values, sparse)
+ proxy.write_attribute(attribute, values)
+
+ # TODO(Corne): Update docstring in L2SS-940
+ def merge_write(
+ self, merge_values: List[any], current_values: List[any], mask_or_sparse=None
+ ):
+ """Merge values as retrieved from :py:func:`~map_write` with current_values
+
+ This method will modify the contents of merge_values.
+
+ To be used by the :py:class:`~AntennaField` device to remove sparse fields
+ from mapped_values with recently retrieved current_values from RECV device.
+
+ :param merge_values: values as retrieved from :py:func:`~map_write`
+ :param current_values: values retrieved from RECV device on specific attribute
+ :param sparse: The value to identify sparse entries
+ """
+
+ if mask_or_sparse is not None and sequence_not_str(mask_or_sparse):
+ self._merge_write_mask(
+ merge_values, current_values, mask_or_sparse
+ )
+ else:
+ self._merge_write_delimiter(
+ merge_values, current_values, mask_or_sparse
+ )
+
+ def _merge_write_delimiter(
+ self, merge_values: List[any], current_values: List[any], sparse=None
+ ):
+ for idx, value in enumerate(merge_values):
+ if sequence_not_str(value):
+ self._merge_write_delimiter(merge_values[idx], current_values[idx], sparse)
+ elif value == sparse:
+ merge_values[idx] = current_values[idx]
+
+ def _merge_write_mask(
+ self, merge_values: List[any], current_values: List[any], mask: List[any]
+ ):
+ for idx, value in enumerate(merge_values):
+ if sequence_not_str(value):
+ self._merge_write_mask(merge_values[idx], current_values[idx], mask[idx])
+ elif not mask[idx]:
+ merge_values[idx] = current_values[idx]
+
@log_exceptions()
def init_device(self):
""" Instantiates the device in the OFF state. """
diff --git a/tangostationcontrol/tangostationcontrol/devices/sdp/digitalbeam.py b/tangostationcontrol/tangostationcontrol/devices/sdp/digitalbeam.py
index 93c91ef791fff255aa2d56e1e1e5c34cdd0f3dbf..a310eb5e44b10ac5b4f08048bd0a0850896fb3a2 100644
--- a/tangostationcontrol/tangostationcontrol/devices/sdp/digitalbeam.py
+++ b/tangostationcontrol/tangostationcontrol/devices/sdp/digitalbeam.py
@@ -224,9 +224,6 @@ class DigitalBeam(beam_device):
beam_weights = self.beamlet_proxy.calculate_bf_weights(fpga_delays.flatten())
beam_weights = beam_weights.reshape((Beamlet.N_PN, Beamlet.A_PN * Beamlet.N_POL * Beamlet.N_BEAMLETS_CTRL))
- # Filter out unwanted antennas (they get a weight of 0)
- beam_weights *= self._map_inputs_on_polarised_inputs(self._input_select)
-
return beam_weights
@TimeIt()
@@ -234,8 +231,13 @@ class DigitalBeam(beam_device):
"""
Uploads beam weights based on a given pointing direction 2D array (96 tiles x 3 parameters)
"""
- # Write weights to SDP
- self.beamlet_proxy.FPGA_bf_weights_xx_yy_RW = beam_weights
+
+ self.atomic_read_modify_write_attribute(
+ beam_weights,
+ self.beamlet_proxy,
+ "FPGA_bf_weights_xx_yy_RW",
+ self._map_inputs_on_polarised_inputs(self._input_select)
+ )
# Record where we now point to, now that we've updated the weights.
# Only record pointings per beamlet, not which antennas took part
diff --git a/tangostationcontrol/tangostationcontrol/integration_test/default/devices/test_device_digitalbeam.py b/tangostationcontrol/tangostationcontrol/integration_test/default/devices/test_device_digitalbeam.py
index b06b60ce82f76b493450f1828bf5a31d8e8a928a..d4ba9f993a56dd3e6f20c03ef43e06e6ebfb02c4 100644
--- a/tangostationcontrol/tangostationcontrol/integration_test/default/devices/test_device_digitalbeam.py
+++ b/tangostationcontrol/tangostationcontrol/integration_test/default/devices/test_device_digitalbeam.py
@@ -21,22 +21,30 @@ class TestDeviceDigitalBeam(AbstractTestBases.TestDeviceBase):
antenna_qualities_only_second = numpy.array([AntennaQuality.BROKEN] + [AntennaQuality.OK] + [AntennaQuality.BROKEN] * 94)
antenna_use_ok = numpy.array([AntennaUse.AUTO] * 96)
+ antennafield_iden = "STAT/AntennaField/1"
+ beamlet_iden = "STAT/Beamlet/1"
+ recv_iden = "STAT/RECV/1"
+ sdp_iden = "STAT/SDP/1"
+
def setUp(self):
"""Intentionally recreate the device object in each test"""
super().setUp("STAT/DigitalBeam/1")
+ self.addCleanup(TestDeviceProxy.test_device_turn_off, self.beamlet_iden)
+ self.addCleanup(TestDeviceProxy.test_device_turn_off, self.recv_iden)
+
self.recv_proxy = self.setup_recv_proxy()
self.beamlet_proxy = self.setup_beamlet_proxy()
def setup_recv_proxy(self):
- recv_proxy = TestDeviceProxy("STAT/RECV/1")
+ recv_proxy = TestDeviceProxy(self.recv_iden)
recv_proxy.off()
recv_proxy.warm_boot()
recv_proxy.set_defaults()
return recv_proxy
def setup_beamlet_proxy(self):
- beamlet_proxy = TestDeviceProxy("STAT/Beamlet/1")
+ beamlet_proxy = TestDeviceProxy(self.beamlet_iden)
beamlet_proxy.off()
beamlet_proxy.warm_boot()
beamlet_proxy.set_defaults()
@@ -44,7 +52,7 @@ class TestDeviceDigitalBeam(AbstractTestBases.TestDeviceBase):
def setup_sdp_proxy(self):
# setup SDP, on which this device depends
- sdp_proxy = TestDeviceProxy("STAT/SDP/1")
+ sdp_proxy = TestDeviceProxy(self.sdp_iden)
sdp_proxy.off()
sdp_proxy.warm_boot()
sdp_proxy.set_defaults()
@@ -53,19 +61,24 @@ class TestDeviceDigitalBeam(AbstractTestBases.TestDeviceBase):
def setup_antennafield_proxy(self, antenna_qualities, antenna_use):
# setup AntennaField
NR_TILES = 48
- antennafield_proxy = TestDeviceProxy("STAT/AntennaField/1")
+ antennafield_proxy = TestDeviceProxy(self.antennafield_iden)
control_mapping = [[1,i] for i in range(NR_TILES)]
- antennafield_proxy.put_property({"RECV_devices": ["STAT/RECV/1"],
- "Control_to_RECV_mapping": numpy.array(control_mapping).flatten(),
+ antennafield_proxy.put_property({"RECV_devices": [self.recv_iden],
+ "HBAT_Control_to_RECV_mapping": numpy.array(control_mapping).flatten(),
'Antenna_Quality': antenna_qualities, 'Antenna_Use': antenna_use})
antennafield_proxy.off()
antennafield_proxy.boot()
return antennafield_proxy
def test_pointing_to_zenith(self):
+ self.addCleanup(TestDeviceProxy.test_device_turn_off, self.sdp_iden)
+ self.addCleanup(
+ TestDeviceProxy.test_device_turn_off, self.antennafield_iden
+ )
+
self.setup_antennafield_proxy(self.antenna_qualities_ok, self.antenna_use_ok)
self.setup_sdp_proxy()
- self.setup_recv_proxy()
+
# Setup beamlet configuration
self.beamlet_proxy.clock_RW = 200 * 1000000
self.beamlet_proxy.subband_select = list(range(488))
@@ -75,13 +88,86 @@ class TestDeviceDigitalBeam(AbstractTestBases.TestDeviceBase):
self.proxy.on()
# Point to Zenith
- self.proxy.set_pointing(numpy.array([["AZELGEO","0deg","90deg"]] * 488).flatten())
+ self.proxy.set_pointing(numpy.array([["AZELGEO", "0deg", "90deg"]] * 488).flatten())
# beam weights should now be non-zero, we don't actually check their values for correctness
self.assertNotEqual(0, sum(self.beamlet_proxy.FPGA_bf_weights_xx_yy_RW.flatten()))
+
+ def test_set_pointing_masked_enable(self):
+ """Verify that only selected inputs are written"""
+
+ self.addCleanup(TestDeviceProxy.test_device_turn_off, self.sdp_iden)
+ self.addCleanup(
+ TestDeviceProxy.test_device_turn_off, self.antennafield_iden
+ )
+
+ self.setup_antennafield_proxy(self.antenna_qualities_ok, self.antenna_use_ok)
+ self.setup_sdp_proxy()
+
+ # Setup beamlet configuration
+ self.beamlet_proxy.clock_RW = 200 * 1000000
+ self.beamlet_proxy.subband_select = list(range(488))
+
+ self.proxy.initialise()
+ self.proxy.Tracking_enabled_RW = False
+ self.proxy.on()
+
+ all_zeros = numpy.array([[0] * 5856] * 16)
+ self.beamlet_proxy.FPGA_bf_weights_xx_yy_RW = all_zeros
+
+ # Enable all inputs
+ self.proxy.input_select_RW = numpy.array([[True] * 488] * 96)
+
+ self.proxy.set_pointing(
+ numpy.array([["AZELGEO", "0deg", "90deg"]] * 488).flatten()
+ )
+
+ # Verify all zeros are replaced with other values for all inputs
+ self.assertTrue(numpy.any(numpy.not_equal(
+ all_zeros, self.beamlet_proxy.FPGA_bf_weights_xx_yy_RW
+ )))
+
+ def test_set_pointing_masked_disable(self):
+ """Verify that only diabled inputs are unchanged"""
+
+ self.addCleanup(TestDeviceProxy.test_device_turn_off, self.sdp_iden)
+ self.addCleanup(
+ TestDeviceProxy.test_device_turn_off, self.antennafield_iden
+ )
+
+ self.setup_antennafield_proxy(self.antenna_qualities_ok, self.antenna_use_ok)
+ self.setup_sdp_proxy()
+
+ # Setup beamlet configuration
+ self.beamlet_proxy.clock_RW = 200 * 1000000
+ self.beamlet_proxy.subband_select = list(range(488))
+
+ self.proxy.initialise()
+ self.proxy.Tracking_enabled_RW = False
+ self.proxy.on()
+
+ non_zeros = numpy.array([[16] * 5856] * 16)
+ self.beamlet_proxy.FPGA_bf_weights_xx_yy_RW = non_zeros
+
+ # Disable all inputs
+ self.proxy.input_select_RW = numpy.array([[False] * 488] * 96)
+
+ self.proxy.set_pointing(
+ numpy.array([["AZELGEO", "0deg", "90deg"]] * 488).flatten()
+ )
+
+ # Verify all zeros are replaced with other values for all inputs
+ numpy.testing.assert_equal(
+ non_zeros, self.beamlet_proxy.FPGA_bf_weights_xx_yy_RW
+ )
def test_input_select_with_all_antennas_ok(self):
- """ Verify if input and antenna select are correctly calculated following Antennafield.Antenna_Usage_Mask """
+ """Verify if input and antenna select are correctly calculated following Antennafield.Antenna_Usage_Mask"""
+
+ self.addCleanup(
+ TestDeviceProxy.test_device_turn_off, self.antennafield_iden
+ )
+
antennafield_proxy = self.setup_antennafield_proxy(self.antenna_qualities_ok, self.antenna_use_ok)
numpy.testing.assert_equal(numpy.array([True] * 96), antennafield_proxy.Antenna_Usage_Mask_R)
self.setUp()
@@ -92,7 +178,12 @@ class TestDeviceDigitalBeam(AbstractTestBases.TestDeviceBase):
numpy.testing.assert_equal(expected_antenna_select, self.proxy.antenna_select_RW)
def test_input_select_with_only_second_antenna_ok(self):
- """ Verify if input and antenna select are correctly calculated following Antennafield.Antenna_Usage_Mask """
+ """Verify if input and antenna select are correctly calculated following Antennafield.Antenna_Usage_Mask"""
+
+ self.addCleanup(
+ TestDeviceProxy.test_device_turn_off, self.antennafield_iden
+ )
+
antennafield_proxy = self.setup_antennafield_proxy(self.antenna_qualities_only_second, self.antenna_use_ok)
numpy.testing.assert_equal(numpy.array([False] + [True] + [False] * 94), antennafield_proxy.Antenna_Usage_Mask_R)
self.setUp()
diff --git a/tangostationcontrol/tangostationcontrol/integration_test/device_proxy.py b/tangostationcontrol/tangostationcontrol/integration_test/device_proxy.py
index 93f84dc98a789c2e0fa5451529728dbed2e73f79..4c519a41184e933e6d9bc720df75f494f7b87051 100644
--- a/tangostationcontrol/tangostationcontrol/integration_test/device_proxy.py
+++ b/tangostationcontrol/tangostationcontrol/integration_test/device_proxy.py
@@ -22,6 +22,7 @@ class TestDeviceProxy(DeviceProxy):
# See also https://www.tango-controls.org/community/forum/c/development/python/attribute-direct-reading-from-device-when-polling-is-turned-on/
self.set_source(DevSource.DEV)
+
@staticmethod
def test_device_turn_off(endpoint):
d = TestDeviceProxy(endpoint)
diff --git a/tangostationcontrol/tangostationcontrol/test/devices/test_digitalbeam_device.py b/tangostationcontrol/tangostationcontrol/test/devices/test_digitalbeam_device.py
new file mode 100644
index 0000000000000000000000000000000000000000..0d19f7cab3b6284caed0e6a6e63f701957f5c582
--- /dev/null
+++ b/tangostationcontrol/tangostationcontrol/test/devices/test_digitalbeam_device.py
@@ -0,0 +1,110 @@
+# -*- coding: utf-8 -*-
+#
+# This file is part of the LOFAR 2.0 Station Software
+#
+#
+#
+# Distributed under the terms of the APACHE license.
+# See LICENSE.txt for more info.
+
+# Builtin regular libraries
+import copy
+
+# External regular libraries
+import numpy
+
+# Internal regular imports
+from tangostationcontrol.devices.sdp import digitalbeam
+
+# Builtin test libraries
+from unittest import mock
+import unittest
+
+# External test libraries
+from tango.test_context import DeviceTestContext
+
+# Internal test imports
+from tangostationcontrol.test.devices import device_base
+
+
+class TestDigitalBeamDevice(device_base.DeviceTestCase):
+
+ def setUp(self):
+ # DeviceTestCase setUp patches lofar_device DeviceProxy
+ super(TestDigitalBeamDevice, self).setUp()
+
+ @unittest.skip("Test for manual use, enable at most one (process=false)")
+ @mock.patch.object(digitalbeam.DigitalBeam, "_wait_to_apply_weights")
+ @mock.patch.object(digitalbeam.DigitalBeam, "_compute_weights")
+ @mock.patch.object(digitalbeam, "DeviceProxy")
+ def test_apply_weights_disabled(self, m_proxy, m_compute, m_wait):
+ """Verify won't overwrite digitalbeam data if no input_selected"""
+
+ input_data = numpy.array([["AZELGEO", "0deg", "90deg"]] * 488).flatten()
+ current_data = numpy.array([[16384] * 5856] * 16)
+
+ m_proxy.return_value = mock.Mock(
+ read_attribute=mock.Mock(
+ return_value=mock.Mock(value=copy.copy(current_data))
+ ),
+ Antenna_Usage_Mask_R=numpy.array([0] * 96),
+ Antenna_Field_Reference_ITRF_R=mock.MagicMock(),
+ HBAT_reference_ITRF_R=numpy.array([[0] * 3] * 96)
+ )
+
+ new_data = numpy.array(
+ [[16384] * 2928 + [0] * 2928] * 16
+ )
+ m_compute.return_value = copy.copy(new_data)
+
+ with DeviceTestContext(
+ digitalbeam.DigitalBeam, process=False,
+ ) as proxy:
+ proxy.initialise()
+ proxy.Tracking_enabled_RW = False
+ proxy.input_select_RW = numpy.array([[False] * 488] * 96)
+
+ proxy.set_pointing(input_data)
+
+ numpy.testing.assert_equal(
+ m_proxy.return_value.write_attribute.call_args[0][0],
+ current_data
+ )
+
+ @unittest.skip("Test for manual use, enable at most one (process=false)")
+ @mock.patch.object(digitalbeam.DigitalBeam, "_wait_to_apply_weights")
+ @mock.patch.object(digitalbeam.DigitalBeam, "_compute_weights")
+ @mock.patch.object(digitalbeam, "DeviceProxy")
+ def test_apply_weights_enabled(self, m_proxy, m_compute, m_wait):
+ """Verify can overwrite digitalbeam data if input_selected"""
+
+ input_data = numpy.array([["AZELGEO", "0deg", "90deg"]] * 488).flatten()
+ current_data = numpy.array([[16384] * 5856] * 16)
+
+ m_proxy.return_value = mock.Mock(
+ read_attribute=mock.Mock(
+ return_value=mock.Mock(value=current_data)
+ ),
+ Antenna_Usage_Mask_R=numpy.array([0] * 96),
+ Antenna_Field_Reference_ITRF_R=mock.MagicMock(),
+ HBAT_reference_ITRF_R=numpy.array([[0] * 3] * 96)
+ )
+
+ new_data = numpy.array(
+ [[16384] * 2928 + [0] * 2928] * 16
+ )
+ m_compute.return_value = copy.copy(new_data)
+
+ with DeviceTestContext(
+ digitalbeam.DigitalBeam, process=False,
+ ) as proxy:
+ proxy.initialise()
+ proxy.Tracking_enabled_RW = False
+ proxy.input_select_RW = numpy.array([[True] * 488] * 96)
+
+ proxy.set_pointing(input_data)
+
+ numpy.testing.assert_equal(
+ m_proxy.return_value.write_attribute.call_args[0][0],
+ new_data
+ )