Allow overriding calibration constant ID

src/calng/base_correction.py

@@ -366,6 +366,23 @@ class BaseCorrection(PythonDevice):
         self._buffer_lock = threading.Lock()
         self.KARABO_SLOT(self.loadMostRecentConstants)
         self.KARABO_SLOT(self.requestScene)
+        # TODO: the CalCatFriend could add these for us
+        # note: overly complicated slot function creation necessary for closure to work
+        def make_wrapper_capturing_constant(constant):
+            def aux():
+                self.calcat_friend.get_specific_constant_version_and_call_me_back(
+                    constant, self._load_constant_to_gpu
+                )
+
+            return aux
+
+        for constant in self._constant_enum_class:
+            slot_name = f"foundConstants.{constant.name}.overrideConstantVersion"
+            meth_name = slot_name.replace(".", "_")
+            self.KARABO_SLOT(
+                make_wrapper_capturing_constant(constant),
+                slotName=meth_name,
+            )
 
     def preReconfigure(self, config):
         for path in config.getPaths():

src/calng/calcat_utils.py

@@ -22,6 +22,7 @@ from karabo.bound import (
     BOOL_ELEMENT,
     DOUBLE_ELEMENT,
     NODE_ELEMENT,
+    SLOT_ELEMENT,
     STRING_ELEMENT,
     UINT32_ELEMENT,
 )
@@ -93,6 +94,22 @@ def _add_status_schema_from_enum(schema, prefix, enum_class):
             .readOnly()
             .initialValue("")
             .commit(),
+            STRING_ELEMENT(schema)
+            .key(f"{constant_node}.constantVersionId")
+            .description(
+                "This field is editable - if for any reason a specific constant "
+                "version is desired, the constant version ID (as used in CalCat) can "
+                "be set here and the slot below can be called to load this particular "
+                "version, overriding the automatic loading of latest constants."
+            )
+            .assignmentOptional()
+            .defaultValue("")
+            .reconfigurable()
+            .commit(),
+            SLOT_ELEMENT(schema)
+            .key(f"{constant_node}.overrideConstantVersion")
+            .displayedName("Override constant version")
+            .commit()
         )
 
 
@@ -248,9 +265,9 @@ class BaseCalcatFriend:
         """Helper to get value from attached device schema"""
         return self.device.get(f"{self.param_prefix}.{key}")
 
-    def _set_status(self, key, value):
+    def _set_status(self, constant, key, value):
         """Helper to update information about found constants on device"""
-        self.device.set(f"{self.status_prefix}.{key}", value)
+        self.device.set(f"{self.status_prefix}.{constant.name}.{key}", value)
 
     @functools.cached_property
     def detector_id(self):
@@ -328,18 +345,18 @@ class BaseCalcatFriend:
             constant = self._constant_enum_class[constant]
 
         calibration_id = self.calibration_id(constant.name)
-        self._set_status(f"{constant.name}.calibrationId", calibration_id)
+        self._set_status(constant, f"calibrationId", calibration_id)
 
         condition = self._constants_need_conditions[constant]()
         condition_id = self.condition_id(
             self._karabo_da_to_float_uuid[karabo_da], condition
         )
-        self._set_status(f"{constant.name}.conditionId", condition_id)
+        self._set_status(constant, f"conditionId", condition_id)
 
         constant_id = self.constant_id(
             calibration_id=calibration_id, condition_id=condition_id
         )
-        self._set_status(f"{constant.name}.constantId", constant_id)
+        self._set_status(constant, f"constantId", constant_id)
 
         resp = CalibrationConstantVersion.get_by_uk(
             self.client,
@@ -352,7 +369,8 @@ class BaseCalcatFriend:
         timestamp = resp["data"][
             "begin_at"
         ]  # TODO: check which key we like (also has begin_validity_at)
-        self._set_status(f"{constant.name}.createdAt", timestamp)
+        self._set_status(constant, f"createdAt", timestamp)
+        self._set_status(constant, f"constantVersionId", resp["data"]["id"])
 
         file_path = (
             self.caldb_store / resp["data"]["path_to_file"] / resp["data"]["file_name"]
@@ -361,10 +379,34 @@ class BaseCalcatFriend:
         with h5py.File(file_path, "r") as fd:
             constant_data = np.array(fd[resp["data"]["data_set_name"]]["data"])
         self.cached_constants[constant] = constant_data
-        self._set_status(f"{constant.name}.found", True)
+        self._set_status(constant, f"found", True)
 
         return constant_data
 
+    def get_specific_constant_version(self, constant):
+        # TODO: warn if PDU or constant type does not match
+        # TODO: warn if result is list (happens for empty version ID)
+        constant_version_id = self.device.get(
+            f"{self.status_prefix}.{constant.name}.constantVersionId"
+        )
+
+        resp = CalibrationConstantVersion.get_by_id(self.client, constant_version_id)
+        _check_resp(resp)
+        file_path = (
+            self.caldb_store / resp["data"]["path_to_file"] / resp["data"]["file_name"]
+        )
+        # TODO: handle FileNotFoundError if we are led astray
+        with h5py.File(file_path, "r") as fd:
+            constant_data = np.array(fd[resp["data"]["data_set_name"]]["data"])
+        self.cached_constants[constant] = constant_data
+        self._set_status(constant, f"createdAt", resp["data"]["begin_at"])
+        self._set_status(constant, f"calibrationId", "manual override")
+        self._set_status(constant, f"conditionId", "manual override")
+        self._set_status(constant, f"constantId", "manual override")
+        self._set_status(constant, f"constantVersionId", constant_version_id)
+        self._set_status(constant, f"found", True)
+        return constant_data
+
     def get_constant_version_and_call_me_back(
         self, constant, callback, snapshot_at=None
     ):
@@ -380,10 +422,23 @@ class BaseCalcatFriend:
         thread.start()
         return thread
 
+    def get_specific_constant_version_and_call_me_back(self, constant, callback):
+        """Blindly load whatever CalCat points to for CCV - user must be confident that
+        this CCV corresponds to correct kind of constant."""
+
+        # TODO: warn user about all the things that go wrong
+        def aux():
+            data = self.get_specific_constant_version(constant)
+            callback(constant, data)
+
+        thread = threading.Thread(target=aux)
+        thread.start()
+        return thread
+
     def flush_constants(self):
         for constant in self._constant_enum_class:
-            self._set_status(f"{constant.name}.createdAt", "")
-            self._set_status(f"{constant.name}.found", False)
+            self._set_status(constant, f"createdAt", "")
+            self._set_status(constant, f"found", False)
 
 
 class AgipdConstants(enum.Enum):