Merge branch 'master' into L2SS-970-add-loki-instance

docker-compose/device-xst.yml

@@ -52,3 +52,5 @@ services:
       # can't know about our Docker port forwarding
       - l2ss-xst XST STAT -v -ORBendPoint giop:tcp:0:5706 -ORBendPointPublish giop:tcp:${HOSTNAME}:5706
     restart: unless-stopped
+    stop_signal: SIGINT # request a graceful shutdown of Tango
+    stop_grace_period: 2s

tangostationcontrol/tangostationcontrol/integration_test/default/devices/test_device_beamlet.py

@@ -11,6 +11,8 @@ from .base import AbstractTestBases
 
 from tangostationcontrol.integration_test.device_proxy import TestDeviceProxy
 
+from tango import DevState
+
 import numpy
 import numpy.testing
 import time
@@ -41,8 +43,6 @@ class TestDeviceBeamlet(AbstractTestBases.TestDeviceBase):
         sdp_proxy = self.setup_sdp()
 
         self.proxy.initialise()
-        # TODO(Corne): Update test so effects of attribute are asserted L2SS-984
-        # self.proxy.subband_select_RW = [0] * 488
         self.proxy.on()
 
         # The subband frequency of HBA subband 0 is 200 MHz,
@@ -59,13 +59,33 @@ class TestDeviceBeamlet(AbstractTestBases.TestDeviceBase):
 
         numpy.testing.assert_almost_equal(expected_bf_weights, calculated_bf_weights)
     
+    def test_subband_select_change(self):
+        # Setup configuration
+        sdp_proxy = self.setup_sdp()
+
+        # Change subband 
+        self.proxy.off()
+        self.proxy.initialise()
+        self.assertEqual(DevState.STANDBY, self.proxy.state())
+        self.proxy.subband_select_RW = [10] * 488
+        self.proxy.on()
+        self.assertEqual(DevState.ON, self.proxy.state())
+
+        # The subband frequency of HBA subband 10 is 201953125 Hz
+        # so its period is 4.95 ns ca, half period is 2.4758e-9
+        delays = numpy.array([[2.4758e-9] * 192] * 488)
+        calculated_bf_weights_subband_10 = self.proxy.calculate_bf_weights(delays.flatten())
+        
+        self.assertEqual(-2**14, c_short(49152).value) # check our calculations
+        expected_bf_weights_10 = numpy.array([49152] * 192 * 488, dtype=numpy.uint32)
+        numpy.testing.assert_almost_equal(expected_bf_weights_10, calculated_bf_weights_subband_10)
+
     def test_sdp_clock_change(self):
         # Setup configuration
         sdp_proxy = self.setup_sdp()
 
         self.proxy.initialise()
-        # TODO(Corne): Update test so effects of attribute are asserted L2SS-984
-        # self.proxy.subband_select_RW = list(range(488))
+        self.proxy.subband_select_RW = numpy.array(list(range(317)) + [316] + list(range(318,488)), dtype=numpy.uint32)
         self.proxy.on()
 
         # any non-zero delay should result in different weights for different clocks
@@ -87,3 +107,16 @@ class TestDeviceBeamlet(AbstractTestBases.TestDeviceBase):
         self.assertTrue((calculated_bf_weights_200 != calculated_bf_weights_160).all())
         self.assertTrue((calculated_bf_weights_200 == calculated_bf_weights_200_v2).all())
 
+        # change subbands
+        self.proxy.off()
+        self.proxy.initialise()
+        self.proxy.subband_select_RW = [317] * 488
+        self.proxy.on()
+        calculated_bf_weights_200_v3 = self.proxy.calculate_bf_weights(delays.flatten())
+        self.assertTrue((calculated_bf_weights_200_v2 != calculated_bf_weights_200_v3).all())
+
+        sdp_proxy.clock_RW = 160 * 1000000
+        time.sleep(1) # wait for beamlet device to process change event
+        calculated_bf_weights_160_v2 = self.proxy.calculate_bf_weights(delays.flatten())
+        self.assertTrue((calculated_bf_weights_160 != calculated_bf_weights_160_v2).all())
+

tangostationcontrol/tangostationcontrol/integration_test/default/devices/test_device_digitalbeam.py

@@ -14,6 +14,7 @@ from tangostationcontrol.devices.antennafield import AntennaQuality, AntennaUse
 from .base import AbstractTestBases
 
 import numpy
+import time
 
 class TestDeviceDigitalBeam(AbstractTestBases.TestDeviceBase):
 
@@ -34,7 +35,7 @@ class TestDeviceDigitalBeam(AbstractTestBases.TestDeviceBase):
         self.addCleanup(TestDeviceProxy.test_device_turn_off, self.recv_iden)
 
         self.recv_proxy = self.setup_recv_proxy()
-        self.beamlet_proxy = self.setup_beamlet_proxy()
+        self.beamlet_proxy = self.initialise_beamlet_proxy()
 
     def setup_recv_proxy(self):
         recv_proxy = TestDeviceProxy(self.recv_iden)
@@ -43,6 +44,12 @@ class TestDeviceDigitalBeam(AbstractTestBases.TestDeviceBase):
         recv_proxy.set_defaults()
         return recv_proxy
 
+    def initialise_beamlet_proxy(self):
+        beamlet_proxy = TestDeviceProxy(self.beamlet_iden)
+        beamlet_proxy.off()
+        beamlet_proxy.initialise()
+        return beamlet_proxy
+    
     def setup_beamlet_proxy(self):
         beamlet_proxy = TestDeviceProxy(self.beamlet_iden)
         beamlet_proxy.off()
@@ -73,19 +80,22 @@ class TestDeviceDigitalBeam(AbstractTestBases.TestDeviceBase):
         antennafield_proxy.boot()
         return antennafield_proxy
 
-    def test_pointing_to_zenith(self):
+    def test_pointing_to_zenith_clock_change(self):
+        self.addCleanup(TestDeviceProxy.test_device_turn_off, self.beamlet_iden)
         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.sdp_proxy = self.setup_sdp_proxy()
+
+        self.beamlet_proxy = self.initialise_beamlet_proxy()
+        self.beamlet_proxy.on()
 
-        # TODO(Corne): Update these and ensure their effects is asserted in tests L2SS-984
-        # Setup beamlet configuration
-        # self.beamlet_proxy.clock_RW = 200 * 1000000
-        # self.beamlet_proxy.subband_select_RW = list(range(488))
+        # Set first (default) clock configuration
+        self.sdp_proxy.clock_RW = 200 * 1000000
+        time.sleep(1)
 
         self.proxy.initialise()
         self.proxy.Tracking_enabled_RW = False
@@ -95,7 +105,52 @@ class TestDeviceDigitalBeam(AbstractTestBases.TestDeviceBase):
         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()))
+        FPGA_bf_weights_xx_yy_clock200 = self.beamlet_proxy.FPGA_bf_weights_xx_yy_RW.flatten()
+        self.assertNotEqual(0, sum(FPGA_bf_weights_xx_yy_clock200))
+
+        self.beamlet_proxy = self.initialise_beamlet_proxy()
+        self.beamlet_proxy.on()
+
+        # Change clock configuration
+        self.sdp_proxy.clock_RW = 160 * 1000000
+        time.sleep(1)
+
+        FPGA_bf_weights_xx_yy_clock160 = self.beamlet_proxy.FPGA_bf_weights_xx_yy_RW.flatten()
+        # Assert some values are different
+        self.assertNotEqual(sum(FPGA_bf_weights_xx_yy_clock160), sum(FPGA_bf_weights_xx_yy_clock200))
+    
+    def test_pointing_to_zenith_subband_change(self):
+        self.addCleanup(TestDeviceProxy.test_device_turn_off, self.beamlet_iden)
+        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.sdp_proxy = self.setup_sdp_proxy()
+
+        self.beamlet_proxy = self.initialise_beamlet_proxy()
+        self.beamlet_proxy.subband_select_RW = numpy.array(list(range(317)) + [316] + list(range(318,488)), dtype=numpy.uint32)
+        self.beamlet_proxy.on()
+
+        self.proxy.initialise()
+        self.proxy.Tracking_enabled_RW = False
+        self.proxy.on()
+
+        # Point to Zenith
+        self.proxy.set_pointing(numpy.array([["AZELGEO", "0deg", "90deg"]] * 488).flatten())
+        # Store values with first subband configuration
+        FPGA_bf_weights_xx_yy_subband_v1 = self.beamlet_proxy.FPGA_bf_weights_xx_yy_RW.flatten()
+
+        # Restart beamlet proxy
+        self.beamlet_proxy = self.initialise_beamlet_proxy()
+        self.beamlet_proxy.subband_select_RW = [317] * 488
+        self.beamlet_proxy.on()
+
+        # Store values with second subband configuration
+        FPGA_bf_weights_xx_yy_subband_v2 = self.beamlet_proxy.FPGA_bf_weights_xx_yy_RW.flatten()
+        # Assert some values are different
+        self.assertNotEqual(sum(FPGA_bf_weights_xx_yy_subband_v1), sum(FPGA_bf_weights_xx_yy_subband_v2))
 
     def test_set_pointing_masked_enable(self):
         """Verify that only selected inputs are written"""