variant in a separate dict and move retrieve_constant_and_time in init_constants

src/cal_tools/agipdlib.py

@@ -988,58 +988,7 @@ class AgipdCorrections:
                                    data=cntsv,
                                    fletcher32=True)
 
-    def retrieve_constant_and_time(self, karabo_id: str, karabo_da: str,
-                                   const_dict: Dict[str, Any],
-                                   cal_db_interface: str,
-                                   creation_time: str
-                                   ) -> Tuple[Dict[str, Any], Dict[str, Any]]:
-        """A wrapper around get_constant_from_db_and_time to get
-        constant image and creation time using a dictionary of
-        the names of the required constant from the data base
-        and their parameter operating conditions.
-
-        :param karabo_id: karabo identifier
-        :param karabo_da: karabo data aggregator
-        :param const_dict: (Dict) A dictionary with constants to retrieve
-        and their operating conditions.
-        e.g.{ "Offset": ["zeros", (128, 512, memory_cells, 3), "Dark"]
-              "Dark-cond": {Condition-name}: condition-value }
-        :param cal_db_interface: (Str) The port interface for
-        calibrationDBRemote
-        :param creation_time: (Str) Raw data creation time
-        :return:
-            cons_data: (np.array) Constant data
-            when: (Dict) A dictionary with all retrieved constants
-            and their creation-time. {"Constant-Name": "creation-time"}
-        """
-        when = {}
-        cons_data = {}
-        for cname, cval in const_dict.items():
-            condition = getattr(
-                Conditions, cval[2][0]).AGIPD(**cval[2][1])
-            cdata, md = get_from_db(
-                karabo_id=karabo_id,
-                karabo_da=karabo_da,
-                constant=getattr(Constants.AGIPD, cname)(),
-                condition=condition,
-                empty_constant=getattr(np, cval[0])(cval[1]),
-                cal_db_interface=cal_db_interface,
-                creation_time=creation_time,
-                verbosity=0,
-            )
-            cons_data[cname] = {
-                "data": cdata,
-                "variant": md.calibration_constant_version.variant
-            }
-
-            when[cname] = None
-            # Read the CCV begin at if constant was retrieved successfully.
-            if md and md.comm_db_success:
-                when[cname] = md.calibration_constant_version.begin_at
-
-        return cons_data, when
-
-    def init_constants(self, cons_data, when, module_idx):
+    def init_constants(self, cons_data, when, module_idx, variant):
         """
         For CI derived gain, a mean multiplication factor of 4.48 compared
         to medium gain is used, as no reliable CI data for all memory cells
@@ -1081,6 +1030,7 @@ class AgipdCorrections:
         :param when: A dictionary for the creation time
                      of each retrieved constant.
         :param module_idx: A module_idx index
+        :param variant: A dictionary for the variant of each retrieved CCV.
         :return:
         """
 
@@ -1088,32 +1038,29 @@ class AgipdCorrections:
         # assuming this method runs in parallel.
         calgs_opts = dict(num_threads=os.cpu_count() // len(self.offset))
 
-        calgs.transpose_constant(
-            self.offset[module_idx], cons_data["Offset"]["data"], **calgs_opts)  # noqa
-        calgs.transpose_constant(
-            self.noise[module_idx], cons_data["Noise"]["data"], **calgs_opts)
+        calgs.transpose_constant(self.offset[module_idx], cons_data['Offset'], **calgs_opts)
+        calgs.transpose_constant(self.noise[module_idx], cons_data['Noise'], **calgs_opts)
         if self.gain_mode is AgipdGainMode.ADAPTIVE_GAIN:
-            calgs.transpose_constant(
-                self.thresholds[module_idx],
-                cons_data["ThresholdsDark"]["data"][..., :3],
-                **calgs_opts
-            )
+            calgs.transpose_constant(self.thresholds[module_idx],
+                                     cons_data['ThresholdsDark'][..., :3],
+                                     **calgs_opts)
 
         if self.corr_bools.get("low_medium_gap"):
             t0 = self.thresholds[module_idx][0]
             t1 = self.thresholds[module_idx][1]
             t1[t1 <= 1.05 * t0] = np.iinfo(np.uint16).max
 
-        bpixels = cons_data["BadPixelsDark"]["data"].astype(np.uint32)
+        bpixels = cons_data["BadPixelsDark"].astype(np.uint32)
 
         if self.corr_bools.get("xray_corr"):
             if when["BadPixelsFF"]:
-                bpixels |= cons_data["BadPixelsFF"]["data"].astype(np.uint32)[
-                    ..., :bpixels.shape[2], None]
+                bpixels |= cons_data["BadPixelsFF"].astype(np.uint32)[...,
+                                                                      :bpixels.shape[2],  # noqa
+                                                                      None]
 
             if when["SlopesFF"]:  # Checking if constant was retrieved
 
-                slopesFF = cons_data["SlopesFF"]["data"]
+                slopesFF = cons_data["SlopesFF"]
                 # This could be used for backward compatibility
                 # for very old SlopesFF constants
                 if len(slopesFF.shape) == 4:
@@ -1150,16 +1097,15 @@ class AgipdCorrections:
         # add additional bad pixel information
         if any(self.pc_bools):
             if when["BadPixelsPC"]:
-                bppc = np.moveaxis(
-                    cons_data["BadPixelsPC"]["data"].astype(np.uint32), 0, 2)
+                bppc = np.moveaxis(cons_data["BadPixelsPC"].astype(np.uint32),
+                                   0, 2)
                 bpixels |= bppc[..., :bpixels.shape[2], None]
 
             # calculate relative gain from the constants
             rel_gain = np.ones((128, 512, self.max_cells, 3), np.float32)
 
             if when["SlopesPC"]:
-                slopesPC = cons_data["SlopesPC"]["data"].astype(
-                    np.float32, copy=False)
+                slopesPC = cons_data["SlopesPC"].astype(np.float32, copy=False)
 
                 # This will handle some historical data in a different format
                 # constant dimension injected first
@@ -1258,6 +1204,7 @@ class AgipdCorrections:
         # string of the device name.
         cons_data = dict()
         when = dict()
+        variant = dict()
         db_module = const_yaml[karabo_da]["physical-detector-unit"]
         for cname, mdata in const_yaml[karabo_da]["constants"].items():
             cons_data[cname] = {}
@@ -1265,16 +1212,15 @@ class AgipdCorrections:
             when[cname] = mdata["creation-time"]
             if when[cname]:
                 with h5py.File(mdata["file-path"], "r") as cf:
-                    cons_data[cname]["data"] = np.copy(cf[f"{base_key}/data"])
+                    cons_data[cname] = np.copy(cf[f"{base_key}/data"])
                     # The vairant attribute is missing for old constants.
                     if "variant" in cf[base_key].attrs.keys():
-                        cons_data[cname]["variant"] = cf[base_key].attrs["variant"]  # noqa
+                        variant[cname] = cf[base_key].attrs["variant"]  # noqa
             else:
                 # Create empty constant using the list elements
-                cons_data[cname]["data"] = getattr(
-                    np, mdata["file-path"][0])(mdata["file-path"][1])
+                cons_data[cname] = getattr(np, mdata["file-path"][0])(mdata["file-path"][1])  # noqa
 
-        self.init_constants(cons_data, when, module_idx)
+        self.init_constants(cons_data, when, module_idx, variant)
 
         return when
 
@@ -1353,11 +1299,32 @@ class AgipdCorrections:
             integration_time=integration_time
         )
 
-        cons_data, when = self.retrieve_constant_and_time(
-            karabo_id, karabo_da, const_dict, cal_db_interface, creation_time
-        )
+        when = {}
+        cons_data = {}
+        variant = {}
+
+        for cname, cval in const_dict.items():
+            condition = getattr(
+                Conditions, cval[2][0]).AGIPD(**cval[2][1])
+            cdata, md = get_from_db(
+                karabo_id=karabo_id,
+                karabo_da=karabo_da,
+                constant=getattr(Constants.AGIPD, cname)(),
+                condition=condition,
+                empty_constant=getattr(np, cval[0])(cval[1]),
+                cal_db_interface=cal_db_interface,
+                creation_time=creation_time,
+                verbosity=0,
+            )
+            cons_data[cname] = cdata
+            variant[cname] = md.calibration_constant_version.variant
+
+            when[cname] = None
+            # Read the CCV begin at if constant was retrieved successfully.
+            if md and md.comm_db_success:
+                when[cname] = md.calibration_constant_version.begin_at
 
-        self.init_constants(cons_data, when, module_idx)
+        self.init_constants(cons_data, when, module_idx, variant)
 
         return when
 

src/cal_tools/tools.py

@@ -642,8 +642,8 @@ def send_to_db(db_module: str, karabo_id: str, constant, condition,
     :param creation_time: Latest time for constant to be created
     :param timeout: Timeout for zmq request
     :param ntries: number of tries to contact the database,
-    ntries is set to 7 so that if the timeout started
-    at 30s last timeout will be ~ 1h.
+           ntries is set to 7 so that if the timeout started
+           at 30s last timeout will be ~ 1h.
     :param doraise: if True raise errors during communication with DB
     :param variant: A calibration constant version variant attribute
     for the constant file.

tests/test_cal_tools.py

@@ -335,6 +335,7 @@ def test_get_pdu_from_db(_agipd_const_cond):
                         "CAL_PHYSICAL_DETECTOR_UNIT-2_TEST"]
 
 
+# TODO add a marker for accessing zmq end_point
 @pytest.mark.requires_gpfs
 @pytest.mark.requires_caldb
 def test_initialize_from_db():