L2SS-1199: refactor injected devices

lofar_station_client/statistics/writer/entry.py

@@ -4,8 +4,8 @@
 """Statistics writer parser and executor"""
 
 # too-many-locals, broad-except, raise-missing-from,
-# too-many-branches, too-many-arguments, too-many-function-args
-# pylint: disable=R0914, W0703, W0707, R0912, R0913, E1121
+# too-many-branches
+# pylint: disable=R0914, W0703, W0707, R0912
 
 import argparse
 import logging
@@ -158,11 +158,7 @@ def _create_writer(
     interval,
     output_dir,
     decimation,
-    antennafield_device: DeviceProxy = None,
-    sdp_device: DeviceProxy = None,
-    tilebeam_device: DeviceProxy = None,
-    digitalbeam_device: DeviceProxy = None,
-    stationmanager_device: DeviceProxy = None,
+    devices: dict[str, DeviceProxy],
 ):
     """Create the writer"""
     if mode == "XST":
@@ -170,28 +166,21 @@ def _create_writer(
             new_file_time_interval=interval,
             file_location=output_dir,
             decimation_factor=decimation,
-            antennafield_device=antennafield_device,
-            sdp_device=sdp_device,
-            tilebeam_device=tilebeam_device,
-            stationmanager_device=stationmanager_device,
+            devices=devices,
         )
     if mode == "SST":
         return SstHdf5Writer(
             new_file_time_interval=interval,
             file_location=output_dir,
             decimation_factor=decimation,
-            antennafield_device=antennafield_device,
-            sdp_device=sdp_device,
-            tilebeam_device=tilebeam_device,
-            stationmanager_device=stationmanager_device,
+            devices=devices,
         )
     if mode == "BST":
         return BstHdf5Writer(
             new_file_time_interval=interval,
             file_location=output_dir,
             decimation_factor=decimation,
-            digitalbeam_device=digitalbeam_device,
-            stationmanager_device=stationmanager_device,
+            devices=devices,
         )
     logger.fatal("Invalid mode: %s", mode)
     sys.exit(1)
@@ -283,44 +272,48 @@ def main():
         logger.setLevel(logging.DEBUG)
         logger.debug("Setting loglevel to DEBUG")
 
+    # Create tango devices dictionary
+    devices = {
+        "antennafield": None,
+        "sdp": None,
+        "tilebeam": None,
+        "digitalbeam": None,
+        "stationmanager": None,
+    }
+
     if args.antennafield:
         antennafield_device = _get_tango_device(
             tango_disabled, host, f"STAT/AntennaField/{args.antennafield}"
         )
+        devices["antennafield"] = antennafield_device
 
         if antennafield_device and filename:
             logger.warning(
                 "Combining live metadata from AntennaField \
                 device with statistics read from disk."
             )
-    else:
-        antennafield_device = None
 
     if args.sdp:
         sdp_device = _get_tango_device(tango_disabled, host, f"STAT/SDP/{args.sdp}")
-    else:
-        sdp_device = None
+        devices["sdp"] = sdp_device
 
     if args.tilebeam:
         tilebeam_device = _get_tango_device(
             tango_disabled, host, f"STAT/TileBeam/{args.tilebeam}"
         )
-    else:
-        tilebeam_device = None
+        devices["tilebeam"] = tilebeam_device
 
     if args.digitalbeam:
         digitalbeam_device = _get_tango_device(
             tango_disabled, host, f"STAT/DigitalBeam/{args.digitalbeam}"
         )
-    else:
-        digitalbeam_device = None
+        devices["digitalbeam"] = digitalbeam_device
 
     if args.stationmanager:
         stationmanager_device = _get_tango_device(
             tango_disabled, host, f"STAT/StationManager/{args.stationmanager}"
         )
-    else:
-        stationmanager_device = None
+        devices["stationmanager"] = stationmanager_device
 
     # creates the TCP receiver that is given to the writer
     receiver = _create_receiver(filename, host, port)
@@ -331,11 +324,7 @@ def main():
         interval,
         output_dir,
         decimation,
-        antennafield_device,
-        sdp_device,
-        tilebeam_device,
-        digitalbeam_device,
-        stationmanager_device,
+        devices,
     )
 
     # start looping

lofar_station_client/statistics/writer/hdf5.py

@@ -49,9 +49,7 @@ __all__ = [
 def _get_station_version(device: DeviceProxy) -> str:
     """Retrieve the Lofar Station Control version"""
     try:
-        if device is not None:
         return device.version_R
-        return ""
     except Exception:
         return ""
 
@@ -59,9 +57,7 @@ def _get_station_version(device: DeviceProxy) -> str:
 def _get_station_name(device: DeviceProxy) -> str:
     """Retrieve the Station name from the StationManager device"""
     try:
-        if device is not None:
         return device.station_name_R
-        return ""
     except Exception:
         return ""
 
@@ -139,11 +135,7 @@ class HDF5Writer(ABC):
         file_location,
         statistics_mode,
         decimation_factor,
-        antennafield_device: DeviceProxy = None,
-        sdp_device: DeviceProxy = None,
-        tilebeam_device: DeviceProxy = None,
-        digitalbeam_device: DeviceProxy = None,
-        stationmanager_device: DeviceProxy = None,
+        devices: dict[str, DeviceProxy],
     ):
         # all variables that deal with the matrix that's currently being decoded
         self.file: StatisticsDataFile = None
@@ -170,11 +162,11 @@ class HDF5Writer(ABC):
         self.mode = statistics_mode.upper()
 
         # Set devices if any, defaults to None
-        self.antennafield_device = antennafield_device
-        self.sdp_device = sdp_device
-        self.tilebeam_device = tilebeam_device
-        self.digitalbeam_device = digitalbeam_device
-        self.stationmanager_device = stationmanager_device
+        self.antennafield_device = devices["antennafield"]
+        self.sdp_device = devices["sdp"]
+        self.tilebeam_device = devices["tilebeam"]
+        self.digitalbeam_device = devices["digitalbeam"]
+        self.stationmanager_device = devices["stationmanager"]
 
         # By default, select all the values from SDP
         self.antenna_selection: List[int] = None
@@ -186,7 +178,9 @@ class HDF5Writer(ABC):
                     self.antennafield_device.Antenna_to_SDP_Mapping_R
                 )
             except DevFailed:
-                logger.exception("Failed to read from %s", antennafield_device.name())
+                logger.exception(
+                    "Failed to read from %s", devices["antennafield"].name()
+                )
             else:
                 # select the values from SDP that represent the antennas in this field
                 self.antenna_selection = list(
@@ -529,21 +523,14 @@ class SstHdf5Writer(HDF5Writer):
         new_file_time_interval,
         file_location,
         decimation_factor,
-        antennafield_device: DeviceProxy = None,
-        sdp_device: DeviceProxy = None,
-        tilebeam_device: DeviceProxy = None,
-        stationmanager_device: DeviceProxy = None,
+        devices: dict[str, DeviceProxy],
     ):
         super().__init__(
             new_file_time_interval,
             file_location,
             HDF5Writer.SST_MODE,
             decimation_factor,
-            antennafield_device=antennafield_device,
-            sdp_device=sdp_device,
-            tilebeam_device=tilebeam_device,
-            digitalbeam_device=None,
-            stationmanager_device=stationmanager_device,
+            devices=devices,
         )
 
     def decoder(self, packet):
@@ -578,19 +565,14 @@ class BstHdf5Writer(HDF5Writer):
         new_file_time_interval,
         file_location,
         decimation_factor,
-        digitalbeam_device: DeviceProxy = None,
-        stationmanager_device: DeviceProxy = None,
+        devices: dict[str, DeviceProxy],
     ):
         super().__init__(
             new_file_time_interval,
             file_location,
             HDF5Writer.BST_MODE,
             decimation_factor,
-            antennafield_device=None,
-            sdp_device=None,
-            tilebeam_device=None,
-            digitalbeam_device=digitalbeam_device,
-            stationmanager_device=stationmanager_device,
+            devices=devices,
         )
 
     def decoder(self, packet):
@@ -615,22 +597,15 @@ class XstHdf5Writer(HDF5Writer):
         new_file_time_interval,
         file_location,
         decimation_factor,
-        antennafield_device,
-        sdp_device,
-        tilebeam_device,
+        devices: dict[str, DeviceProxy],
         subband_index,
-        stationmanager_device,
     ):
         super().__init__(
             new_file_time_interval,
             file_location,
             HDF5Writer.XST_MODE,
             decimation_factor,
-            antennafield_device,
-            tilebeam_device,
-            sdp_device,
-            digitalbeam_device=None,
-            stationmanager_device=stationmanager_device,
+            devices=devices,
         )
         self.subband_index = subband_index
 
@@ -680,10 +655,7 @@ class ParallelXstHdf5Writer:
         new_file_time_interval,
         file_location,
         decimation_factor,
-        antennafield_device,
-        sdp_device,
-        tilebeam_device,
-        stationmanager_device,
+        devices: dict[str, DeviceProxy],
     ):
         # maintain a dedicated HDF5Writer per subband
         self.writers = {}
@@ -695,11 +667,8 @@ class ParallelXstHdf5Writer:
                 new_file_time_interval,
                 file_location,
                 decimation_factor,
-                antennafield_device,
-                sdp_device,
-                tilebeam_device,
+                devices,
                 subband,
-                stationmanager_device,
             )
 
         self.new_writer = new_writer