Modernise use of extra_data to get AGIPD control data

src/cal_tools/agipdlib.py

@@ -12,7 +12,9 @@ import h5py
 import numpy as np
 import sharedmem
 from dateutil import parser
-from extra_data import DataCollection, H5File, RunDirectory, by_id
+from extra_data import (
+    DataCollection, H5File, RunDirectory, by_id, PropertyNameError, SourceNameError,
+)
 
 from cal_tools import agipdalgs as calgs
 from cal_tools.agipdutils import (
@@ -51,14 +53,10 @@ class AgipdCtrl:
     def _get_num_cells_ctrl(self) -> Optional[int]:
         """Get number of cells from CONTROL source."""
         # Attempt to look for number of cells in slow data
-        ncell_src = (
-            self.ctrl_src, "bunchStructure.nPulses.value")
-        if (
-            ncell_src[0] in self.run_dc.all_sources and
-            ncell_src[1] in self.run_dc.keys_for_source(ncell_src[0])
-        ):
-            return int(self.run_dc[ncell_src].as_single_value(reduce_by='max'))
-        else:
+        try:
+            kd = self.run_dc[self.ctrl_src, "bunchStructure.nPulses"]
+            return int(kd.as_single_value(reduce_by='max'))
+        except (SourceNameError, PropertyNameError):
             return None
 
     def _get_num_cells_instr(self) -> int:
@@ -91,15 +89,11 @@ class AgipdCtrl:
     def _get_acq_rate_ctrl(self) -> Optional[float]:
         """Get acquisition (repetition) rate from CONTROL source."""
         # Attempt to look for acquisition rate in slow data
-        rep_rate_src = (
-            self.ctrl_src, "bunchStructure.repetitionRate.value")
-        if (
-            rep_rate_src[0] in self.run_dc.all_sources and
-            rep_rate_src[1] in self.run_dc.keys_for_source(rep_rate_src[0])
-        ):
-            # It is desired to loose precision here because the usage is
-            # about bucketing the rate for managing meta-data.
-            return round(float(self.run_dc[rep_rate_src].as_single_value()), 1)
+        try:
+            kd = self.run_dc[self.ctrl_src, "bunchStructure.repetitionRate"]
+            return round(float(kd.as_single_value()), 1)
+        except (SourceNameError, PropertyNameError):
+            return None
 
     def _get_acq_rate_instr(self) -> Optional[float]:
         """Get acquisition (repetition rate) from INSTRUMENT source."""
@@ -194,7 +188,7 @@ class AgipdCtrl:
             print("Set gain-setting to None for runs taken before 2020-01-31")
             return
 
-        if "gain.value" in self.run_dc.keys_for_source(self.ctrl_src):
+        if "gain.value" in self.run_dc[self.ctrl_src]:
             return self._get_gain_setting_ctrl()
 
         gain_setting = self._get_gain_setting_ctrl_old()
@@ -210,16 +204,11 @@ class AgipdCtrl:
 
     def get_gain_mode(self) -> int:
         """Returns the gain mode (adaptive or fixed) from slow data."""
-
-        if (
-            self.ctrl_src in self.run_dc.all_sources and
-            "gainModeIndex.value" in self.run_dc.keys_for_source(
-                self.ctrl_src)
-        ):
-            return AgipdGainMode(int(self.run_dc[
-                self.ctrl_src, "gainModeIndex"].as_single_value()))
-
-        return AgipdGainMode.ADAPTIVE_GAIN
+        try:
+            kd = self.run_dc[self.ctrl_src, "gainModeIndex"]
+            return AgipdGainMode(int(kd.as_single_value()))
+        except (SourceNameError, PropertyNameError):
+            return AgipdGainMode.ADAPTIVE_GAIN
 
     def get_bias_voltage(
         self,
@@ -257,16 +246,13 @@ class AgipdCtrl:
                 "highVoltage.actual.value")
             default_voltage = None
 
-        if (
-            voltage_src[0] in self.run_dc.all_sources and
-            voltage_src[1] in self.run_dc.keys_for_source(voltage_src[0])
-        ):
+        try:
             # Use RUN source for reading the bias voltage value.
             # As HED_DET_AGIPD500K2G has a hardware issue that leads
             # to storing arbitrary voltage values in the CONTROL source
             # array. e.g. /gpfs/exfel/exp/HED/202230/p900248/raw
             return int(self.run_dc.get_run_value(*voltage_src))
-        else:
+        except (SourceNameError, PropertyNameError):
             # TODO: Validate if removing this and
             # and using NB value for old RAW data.
             error = ("ERROR: Unable to read bias_voltage from"
@@ -287,15 +273,11 @@ class AgipdCtrl:
 
         :return: integration time
         """
-        if (
-            self.ctrl_src in self.run_dc.all_sources and
-            'integrationTime.value' in self.run_dc.keys_for_source(
-                self.ctrl_src)
-        ):
+        try:
             return int(self.run_dc[
                 self.ctrl_src, 'integrationTime'].as_single_value())
-
-        return 12
+        except (SourceNameError, PropertyNameError):
+            return 12
 
 
 @dataclass