Merge branch 'L2SS-436-conjugate-some-xsts' into 'master'

L2SS-436: Conjugate XSTs if they are from the other triangle of the matrix...

Closes L2SS-436

See merge request !149

devices/devices/sdp/statistics_collector.py

@@ -131,6 +131,8 @@ class XSTCollector(StatisticsCollector):
 
             # Last value array we've constructed out of the packets
             "xst_blocks":            numpy.zeros((self.MAX_BLOCKS, self.BLOCK_LENGTH * self.BLOCK_LENGTH * self.VALUES_PER_COMPLEX), dtype=numpy.int64),
+            # Whether the values are actually conjugated and transposed
+            "xst_conjugated":        numpy.zeros((self.MAX_BLOCKS,), dtype=numpy.bool_),
             "xst_timestamps":        numpy.zeros((self.MAX_BLOCKS,), dtype=numpy.float64),
             "xst_subbands":          numpy.zeros((self.MAX_BLOCKS,), dtype=numpy.uint16),
             "integration_intervals": numpy.zeros((self.MAX_BLOCKS,), dtype=numpy.float32),
@@ -162,20 +164,29 @@ class XSTCollector(StatisticsCollector):
             if fields.first_baseline[antenna] % self.BLOCK_LENGTH != 0:
                 raise ValueError("Packet describes baselines starting at %s, but we require a multiple of BLOCK_LENGTH=%d" % (fields.first_baseline, self.MAX_INPUTS))
 
+        # Make sure we always have a baseline (a,b) with a>=b. If not, we swap the indices and mark that the data must be conjugated and transposed when processed.
+        first_baseline = fields.first_baseline
+        if first_baseline[0] < first_baseline[1]:
+            conjugated = True
+            first_baseline = (first_baseline[1], first_baseline[0])
+        else:
+            conjugated = False
+
         # the payload contains complex values for the block of baselines of size BLOCK_LENGTH x BLOCK_LENGTH
         # starting at baseline first_baseline.
         #
         # we honour this format, as we want to keep the metadata together with these blocks. we do need to put the blocks in a linear
         # and tight order, however, so we calculate a block index.
-        block_index = baseline_index(fields.first_baseline[0] // self.BLOCK_LENGTH, fields.first_baseline[1] // self.BLOCK_LENGTH)
+        block_index = baseline_index(first_baseline[0] // self.BLOCK_LENGTH, first_baseline[1] // self.BLOCK_LENGTH)
+
+        # We did enough checks on first_baseline for this to be a logic error in our code
+        assert 0 <= block_index < self.MAX_BLOCKS, f"Received block {block_index}, but have only room for {self.MAX_BLOCKS}. Block starts at baseline {first_baseline}."
 
         # process the packet
         self.parameters["nof_valid_payloads"][fields.gn_index] += numpy.uint64(1)
-
-        block_index = baseline_index(fields.first_baseline[0], fields.first_baseline[1])
-
         self.parameters["xst_blocks"][block_index][:fields.nof_statistics_per_packet] = fields.payload
         self.parameters["xst_timestamps"][block_index]        = numpy.float64(fields.timestamp().timestamp())
+        self.parameters["xst_conjugated"][block_index]        = conjugated
         self.parameters["xst_subbands"][block_index]          = numpy.uint16(fields.subband_index)
         self.parameters["integration_intervals"][block_index] = fields.integration_interval()
 
@@ -184,11 +195,16 @@ class XSTCollector(StatisticsCollector):
 
         matrix = numpy.zeros((self.MAX_INPUTS, self.MAX_INPUTS), dtype=numpy.complex64)
         xst_blocks = self.parameters["xst_blocks"]
+        xst_conjugated = self.parameters["xst_conjugated"]
 
         for block_index in range(self.MAX_BLOCKS):
             # convert real/imag int to complex float values. this works as real/imag come in pairs
             block = xst_blocks[block_index].astype(numpy.float32).view(numpy.complex64)
 
+            if xst_conjugated[block_index]:
+                # block is conjugated and transposed. process.
+                block = block.conjugate().transpose()
+
             # reshape into [a][b]
             block = block.reshape(self.BLOCK_LENGTH, self.BLOCK_LENGTH)
 

devices/devices/sdp/xst.py

@@ -118,6 +118,8 @@ class XST(Statistics):
     nof_payload_errors_R    = attribute_wrapper(comms_id=StatisticsClient, comms_annotation={"type": "statistics", "parameter": "nof_payload_errors"}, dims=(XSTCollector.MAX_FPGAS,), datatype=numpy.uint64)
     # latest XSTs
     xst_blocks_R            = attribute_wrapper(comms_id=StatisticsClient, comms_annotation={"type": "statistics", "parameter": "xst_blocks"}, dims=(XSTCollector.BLOCK_LENGTH * XSTCollector.BLOCK_LENGTH * XSTCollector.VALUES_PER_COMPLEX, XSTCollector.MAX_BLOCKS), datatype=numpy.int64)
+    # whether the values in the block are conjugated and transposed
+    xst_conjugated_R        = attribute_wrapper(comms_id=StatisticsClient, comms_annotation={"type": "statistics", "parameter": "xst_conjugated"}, dims=(XSTCollector.MAX_BLOCKS,), datatype=numpy.bool_)
     # reported timestamp for each row in the latest XSTs
     xst_timestamp_R         = attribute_wrapper(comms_id=StatisticsClient, comms_annotation={"type": "statistics", "parameter": "xst_timestamps"}, dims=(XSTCollector.MAX_BLOCKS,), datatype=numpy.uint64)
     # which subband the XSTs describe

devices/test/devices/test_statistics_collector.py

@@ -7,13 +7,16 @@ class TestXSTCollector(base.TestCase):
     def test_valid_packet(self):
         collector = XSTCollector()
 
-        # a valid packet as obtained from SDP, with 64-bit BE 1+1j as payload
-        packet =  b'X\x05\x00\x00\x00\x00\x00\x00\x10\x08\x00\x02\xfa\xef\x00f\x00\x00\x0c\x08\x01 \x14\x00\x00\x01!\xd9&z\x1b\xb3' + 288 * b'\x00\x00\x00\x00\x00\x00\x00\x01'
+        # a valid packet as obtained from SDP, with 64-bit BE 1+1j as payload at (12,0)
+        packet =  b'X\x05\x00\x00\x00\x00\x00\x00\x10\x08\x00\x02\xfa\xef\x00f\x0c\x00\x0c\x08\x01 \x14\x00\x00\x01!\xd9&z\x1b\xb3' + 288 * b'\x00\x00\x00\x00\x00\x00\x00\x01'
 
         # parse it ourselves to extract info nicely
         fields = XSTPacket(packet)
         fpga_index = fields.gn_index
 
+        # baseline indeed should be (12,0)
+        self.assertEqual((12,0), fields.first_baseline)
+
         # this should not throw
         collector.process_packet(packet)
 
@@ -41,10 +44,51 @@ class TestXSTCollector(base.TestCase):
                 else:
                     self.assertEqual(0+0j, xst_values[baseline_a][baseline_b], msg=f'element [{baseline_a}][{baseline_b}] was not in packet, but was written to the XST matrix.')
 
+    def test_conjugated_packet(self):
+        """ Test whether a packet with a baseline (a,b) with a<b will get its payload conjugated. """
+
+        collector = XSTCollector()
+
+        # a valid packet as obtained from SDP, with 64-bit BE 1+1j as payload, at baseline (0,12)
+        #                                                                       VV  VV
+        packet =  b'X\x05\x00\x00\x00\x00\x00\x00\x10\x08\x00\x02\xfa\xef\x00f\x00\x0c\x0c\x08\x01 \x14\x00\x00\x01!\xd9&z\x1b\xb3' + 288 * b'\x00\x00\x00\x00\x00\x00\x00\x01'
+
+        # parse it ourselves to extract info nicely
+        fields = XSTPacket(packet)
+
+        # baseline indeed should be (0,12)
+        self.assertEqual((0,12), fields.first_baseline)
+
+        # this should not throw
+        collector.process_packet(packet)
+
+        # counters should now be updated
+        self.assertEqual(1, collector.parameters["nof_packets"])
+        self.assertEqual(0, collector.parameters["nof_invalid_packets"])
+
+        # check whether the data ended up in the right block, and the rest is still zero
+        xst_values = collector.xst_values()
+
+        for baseline_a in range(collector.MAX_INPUTS):
+            for baseline_b in range(collector.MAX_INPUTS):
+                if baseline_b > baseline_a:
+                    # only scan top-left triangle
+                    continue
+
+                # use swapped indices!
+                baseline_a_was_in_packet = (fields.first_baseline[1] <= baseline_a < fields.first_baseline[1] + fields.nof_signal_inputs)
+                baseline_b_was_in_packet = (fields.first_baseline[0] <= baseline_b < fields.first_baseline[0] + fields.nof_signal_inputs)
+
+                if baseline_a_was_in_packet and baseline_b_was_in_packet:
+                    self.assertEqual(1-1j, xst_values[baseline_a][baseline_b], msg=f'element [{baseline_a}][{baseline_b}] did not end up conjugated in XST matrix.')
+                else:
+                    self.assertEqual(0+0j, xst_values[baseline_a][baseline_b], msg=f'element [{baseline_a}][{baseline_b}] was not in packet, but was written to the XST matrix.')
+
     def test_invalid_packet(self):
         collector = XSTCollector()
 
         # an invalid packet
+        #           V
         packet =  b'S\x05\x00\x00\x00\x00\x00\x00\x10\x08\x00\x02\xfa\xef\x00f\x00\x00\x0c\x08\x01 \x14\x00\x00\x01!\xd9&z\x1b\xb3' + 288 * b'\x00\x00\x00\x00\x00\x00\x00\x01'
 
         # this should throw
@@ -62,6 +106,7 @@ class TestXSTCollector(base.TestCase):
         collector = XSTCollector()
 
         # an valid packet with a payload error
+        #                                           V
         packet =  b'X\x05\x00\x00\x00\x00\x00\x00\x14\x08\x00\x02\xfa\xef\x00f\x00\x00\x0c\x08\x01 \x14\x00\x00\x01!\xd9&z\x1b\xb3' + 288 * b'\x00\x00\x00\x00\x00\x00\x00\x01'
 
         # parse it ourselves to extract info nicely