diff --git a/lofar_station_client/statistics/writer/entry.py b/lofar_station_client/statistics/writer/entry.py index 4336967a441400fa74156117528dcc5b956ea764..578b658f796d39f6218ddb1e8b2e063a154412ca 100644 --- a/lofar_station_client/statistics/writer/entry.py +++ b/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) diff --git a/tests/statistics/test_writer.py b/tests/statistics/test_writer.py index 3a44840846034be864cf71e5ee9221621672ab8b..6102b485805211b60a685636f21dc02a1834e4d5 100644 --- a/tests/statistics/test_writer.py +++ b/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: