Skip to content
Snippets Groups Projects
Commit 285ea074 authored by Thomas Kluyver's avatar Thomas Kluyver
Browse files

Convert functions to CalCatAPIClient methods

parent b3066fb1
No related branches found
No related tags found
1 merge request!885Revised CalCat API
......@@ -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)
......
0% Loading or .
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment