From 285ea0744cc60be88ef9f01b1a0c6b23532f591a Mon Sep 17 00:00:00 2001
From: Thomas Kluyver <thomas@kluyver.me.uk>
Date: Tue, 2 Jan 2024 16:31:25 +0000
Subject: [PATCH] Convert functions to CalCatAPIClient methods
---
src/cal_tools/calcat_interface2.py | 85 +++++++++++++++---------------
1 file changed, 42 insertions(+), 43 deletions(-)
diff --git a/src/cal_tools/calcat_interface2.py b/src/cal_tools/calcat_interface2.py
index bb1cf1f7f..80746f58b 100644
--- a/src/cal_tools/calcat_interface2.py
+++ b/src/cal_tools/calcat_interface2.py
@@ -111,6 +111,41 @@ class CalCatAPIClient:
}
return content, pagination_info
+ # ------------------
+ # Cached wrappers for simple ID lookups of fixed-ish info
+ #
+ # N.B. lru_cache behaves oddly with instance methods (it's a global cache,
+ # with the instance as part of the key), but in this case it should be OK.
+ @lru_cache()
+ def calibration_by_id(self, cal_id):
+ return self.get(f"calibrations/{cal_id}")
+
+ @lru_cache()
+ def detector_by_id(self, det_id):
+ return self.get(f"detectors/{det_id}")
+
+ # --------------------
+ # Shortcuts to find 1 of something by an ID-like field (e.g. name) other
+ # than CalCat's own integer IDs. Error on no match or >1 matches.
+ @lru_cache()
+ def detector_by_identifier(self, identifier):
+ # The "identifier", "name" & "karabo_name" fields seem to have the same names
+ res = self.get("detectors", {"identifier": identifier})
+ if not res:
+ raise KeyError(f"No detector with identifier {identifier}")
+ elif len(res) > 1:
+ raise ValueError(f"Multiple detectors found with identifier {identifier}")
+ return res[0]
+
+ @lru_cache()
+ def calibration_by_name(self, name):
+ res = self.get("calibrations", {"name": name})
+ if not res:
+ raise KeyError(f"No calibration with name {name}")
+ elif len(res) > 1:
+ raise ValueError(f"Multiple calibrations found with name {name}")
+ return res[0]
+
global_client = None
@@ -329,45 +364,6 @@ class MultiModuleConstant:
return xarray.DataArray(ndarr, dims=dims, coords=coords, name=name)
-@lru_cache()
-def detector_by_name(identifier, client=None):
- client = client or get_client()
- res = client.get("detectors", {"identifier": identifier})
- if not res:
- raise KeyError(f"No detector with identifier {identifier}")
- elif len(res) > 1:
- raise ValueError(f"Multiple detectors found with identifier {identifier}")
- return res[0]
-
-
-@lru_cache()
-def detector_id_to_name(det_id: int, client=None):
- """Convert a numeric detector ID to a name like 'FXE_DET_LPD1M-1'"""
- client = client or get_client()
- res = client.get(f"detectors/{det_id}")
- return res["identifier"] # "name" & "karabo_name" appear to be equivalent
-
-
-@lru_cache()
-def calibration_id(name, client=None):
- """ID for a calibration in CalCat."""
- client = client or get_client()
- res = client.get("calibrations", {"name": name})
- if not res:
- raise KeyError(f"No calibration with name {name}")
- elif len(res) > 1:
- raise ValueError(f"Multiple calibrations found with name {name}")
- return res[0]["id"]
-
-
-@lru_cache()
-def calibration_name(cal_id, client=None):
- """Name for a calibration in CalCat."""
- client = client or get_client()
- res = client.get(f"calibrations/{cal_id}")
- return res["name"]
-
-
class CalibrationData(Mapping):
"""Collected constants for a given detector"""
@@ -416,7 +412,7 @@ class CalibrationData(Mapping):
client = client or get_client()
- detector_id = detector_by_name(detector_name, client)["id"]
+ detector_id = client.detector_by_identifier(detector_name)["id"]
pdus = client.get(
"physical_detector_units/get_all_by_detector",
{
@@ -435,7 +431,7 @@ class CalibrationData(Mapping):
condition_dict = condition.make_dict(params)
cal_id_map = {
- calibration_id(calibration): calibration for calibration in cal_types
+ client.calibration_by_name(name)["id"]: name for name in cal_types
}
calibration_ids = list(cal_id_map.keys())
@@ -498,13 +494,16 @@ class CalibrationData(Mapping):
else:
pdus[kda] = pdu
- cal_type = calibration_name(ccv["calibration_constant"]["calibration_id"])
+ cal_type = client.calibration_by_id(
+ ccv["calibration_constant"]["calibration_id"]
+ )["name"]
const_group = constant_groups.setdefault(cal_type, {})
const_group[kda] = SingleConstant.from_response(ccv)
if len(det_ids) > 1:
raise Exception(f"Found multiple detector IDs in report: {det_ids}")
- det_name = detector_id_to_name(det_ids.pop(), client)
+ # The "identifier", "name" & "karabo_name" fields seem to have the same names
+ det_name = client.detector_by_id(det_ids.pop())["identifier"]
module_details = sorted(pdus.values(), key=lambda d: d["karabo_da"])
return cls(constant_groups, module_details, det_name)
--
GitLab