L2SS-1203: add bst header test

lofar_station_client/statistics/writer/entry.py

@@ -3,8 +3,9 @@
 
 """Statistics writer parser and executor"""
 
-# too-many-locals, broad-except, raise-missing-from, too-many-arguments
-# pylint: disable=R0914, W0703, W0707, R0913
+# too-many-locals, broad-except, raise-missing-from,
+# too-many-branches, too-many-arguments
+# pylint: disable=R0914, W0703, W0707, R0912, R0913
 
 import argparse
 import logging
@@ -107,6 +108,30 @@ def _create_parser():
         default="",
         help="Antenna field to collect data for",
     )
+    parser.add_argument(
+        "-SDP",
+        "--sdp",
+        type=str,
+        choices=["", "1"],
+        default="",
+        help="SDP device to collect data for",
+    )
+    parser.add_argument(
+        "-TB",
+        "--tilebeam",
+        type=str,
+        choices=["", "LBA", "HBA", "HBA0", "HBA1"],
+        default="",
+        help="Tilebeam device to collect data for",
+    )
+    parser.add_argument(
+        "-DB",
+        "--digitalbeam",
+        type=str,
+        choices=["", "LBA", "HBA", "HBA0", "HBA1"],
+        default="",
+        help="Digitalbeam device to collect data for",
+    )
     return parser
 
 
@@ -259,11 +284,39 @@ def main():
     else:
         antennafield_device = None
 
+    if args.sdp:
+        sdp_device = _get_tango_device(tango_disabled, host, f"STAT/SDP/{args.sdp}")
+    else:
+        sdp_device = None
+
+    if args.tilebeam:
+        tilebeam_device = _get_tango_device(
+            tango_disabled, host, f"STAT/TileBeam/{args.sdp}"
+        )
+    else:
+        tilebeam_device = None
+
+    if args.digitalbeam:
+        digitalbeam_device = _get_tango_device(
+            tango_disabled, host, f"STAT/DigitalBeam/{args.digitalbeam}"
+        )
+    else:
+        digitalbeam_device = None
+
     # creates the TCP receiver that is given to the writer
     receiver = _create_receiver(filename, host, port)
 
     # create the writer
-    writer = _create_writer(mode, interval, output_dir, decimation, antennafield_device)
+    writer = _create_writer(
+        mode,
+        interval,
+        output_dir,
+        decimation,
+        antennafield_device,
+        sdp_device,
+        tilebeam_device,
+        digitalbeam_device,
+    )
 
     # start looping
     _start_loop(receiver, writer, reconnect, filename)

tests/statistics/test_writer.py

@@ -17,11 +17,35 @@ from lofar_station_client.statistics import reader
 from tests.test_devices import (
     FakeAntennaFieldDeviceProxy,
     FakeOffAntennaFieldDeviceProxy,
+    FakeDigitalBeamDeviceProxy,
 )
 from tests import base
 
 
-class TestStatisticsReaderWriterSST(base.TestCase):
+class TestStatisticsReaderWriter(base.TestCase):
+    """Parent TestStatistics class which exposes common internal methods"""
+
+    def _mock_get_tango_device(self, tango_disabled, host, device_name):
+        """Return our mocked DeviceProxies"""
+        if device_name == "STAT/AntennaField/LBA":
+            return FakeAntennaFieldDeviceProxy(device_name)
+        if device_name == "STAT/DigitalBeam/HBA":
+            return FakeDigitalBeamDeviceProxy(device_name)
+        raise ValueError(
+            f"Device not mocked, and thus not available in this test: {device_name}"
+        )
+
+    def _mock_get_tango_device_off(self, tango_disabled, host, device_name):
+        """Return our mocked DeviceProxies that simulate a device that is off"""
+        if device_name == "STAT/AntennaField/LBA":
+            return FakeOffAntennaFieldDeviceProxy(device_name)
+
+        raise ValueError(
+            f"Device not mocked, and thus not available in this test: {device_name}"
+        )
+
+
+class TestStatisticsReaderWriterSST(TestStatisticsReaderWriter):
     def _run_writer_reader(
         self, tmpdir: str, writer_argv: list
     ) -> (StatisticsData, StatisticsFileHeader):
@@ -71,24 +95,6 @@ class TestStatisticsReaderWriterSST(base.TestCase):
 
         return stat, file_header
 
-    def _mock_get_tango_device(self, tango_disabled, host, device_name):
-        """Return our mocked DeviceProxies"""
-        if device_name == "STAT/AntennaField/LBA":
-            return FakeAntennaFieldDeviceProxy(device_name)
-
-        raise ValueError(
-            f"Device not mocked, and thus not available in this test: {device_name}"
-        )
-
-    def _mock_get_tango_device_off(self, tango_disabled, host, device_name):
-        """Return our mocked DeviceProxies that simulate a device that is off"""
-        if device_name == "STAT/AntennaField/LBA":
-            return FakeOffAntennaFieldDeviceProxy(device_name)
-
-        raise ValueError(
-            f"Device not mocked, and thus not available in this test: {device_name}"
-        )
-
     def test_header_info(self):
         """Test whether the header info are inserted and collected in the proper way"""
         with TemporaryDirectory() as tmpdir:
@@ -139,6 +145,68 @@ class TestStatisticsReaderWriterSST(base.TestCase):
                 _ = self._run_writer_reader(tmpdir, writer_argv)
 
 
+class TestStatisticsReaderWriterBST(TestStatisticsReaderWriter):
+    def _run_writer_reader(
+        self, tmpdir: str, writer_argv: list
+    ) -> (StatisticsData, StatisticsFileHeader):
+        """Run the statistics writer with the given arguments,
+        and read and return the output."""
+        # default arguments for statistics writer
+        default_writer_sys_argv = [
+            sys.argv[0],
+            "--mode",
+            "BST",
+            "--file",
+            dirname(__file__) + "/SDP_BST_statistics_packets.bin",
+            "--output_dir",
+            tmpdir,
+        ]
+
+        with mock.patch.object(
+            entry.sys, "argv", default_writer_sys_argv + writer_argv
+        ):
+            with self.assertRaises(SystemExit):
+                entry.main()
+
+        # check if file was written
+        self.assertTrue(isfile(f"{tmpdir}/BST_2022-05-20-11-08-44.h5"))
+
+        # default arguments for statistics reader
+        default_reader_sys_argv = [
+            sys.argv[0],
+            "--files",
+            f"{tmpdir}/BST_2022-05-20-11-08-44.h5",
+            "--start_time",
+            "2021-09-20#07:40:08.937+00:00",
+            "--end_time",
+            "2021-10-04#07:50:08.937+00:00",
+        ]
+
+        # test statistics reader
+        with mock.patch.object(reader.sys, "argv", default_reader_sys_argv):
+            stat_parser = reader.setup_stat_parser()
+            bst_statistics = stat_parser.list_statistics()
+            self.assertIsNotNone(bst_statistics)
+            file_header = stat_parser.file_header
+
+        return file_header
+
+    def test_insert_tango_BST_statistics(self):
+        with TemporaryDirectory() as tmpdir:
+            writer_argv = [
+                "--digitalbeam",
+                "HBA",
+            ]
+
+            with mock.patch.object(
+                entry, "_get_tango_device", self._mock_get_tango_device
+            ):
+                file_header = self._run_writer_reader(tmpdir, writer_argv)
+
+            # Test some AntennField attributes, whether they match our mock
+            self.assertListEqual(list(range(0, 488)), file_header.subbands.tolist())
+
+
 class TestStatisticsWriter(base.TestCase):
     def test_xst(self):
         with TemporaryDirectory() as tmpdir: