Merge branch 'calcat-api-2' into 'master'

Revised CalCat API

See merge request !885

.pre-commit-config.yaml

@@ -17,7 +17,7 @@ repos:
       # If `CI_MERGE_REQUEST_TARGET_BRANCH_SHA` env var is set then this will
       # run flake8 on the diff of the merge request, otherwise it will run
       # flake8 as it would usually execute via the pre-commit hook
-      entry: bash -c 'if [ -z ${CI_MERGE_REQUEST_TARGET_BRANCH_SHA} ]; then (flake8 "$@"); else (git diff $CI_MERGE_REQUEST_TARGET_BRANCH_SHA...$CI_MERGE_REQUEST_SOURCE_BRANCH_SHA | flake8 --diff); fi' --
+      entry: bash -c 'if [ -z ${CI_MERGE_REQUEST_TARGET_BRANCH_SHA} ]; then (flake8 "$@" --max-line-length 88); else (git diff $CI_MERGE_REQUEST_TARGET_BRANCH_SHA...$CI_MERGE_REQUEST_SOURCE_BRANCH_SHA | flake8 --diff --max-line-length 88); fi' --
   - repo: https://github.com/myint/rstcheck
     rev: 3f92957478422df87bd730abde66f089cc1ee19b  # commit where pre-commit support was added
     hooks:

setup.py

@@ -58,7 +58,7 @@ install_requires = [
         "markupsafe==2.0.1",
         "astcheck==0.3.0",
         "cfelpyutils==2.0.6",
-        "calibration_client==11.2.0",
+        "calibration_client==11.3.0",
         "dill==0.3.0",
         "docutils==0.17.1",
         "dynaconf==3.1.4",

src/cal_tools/calcat_interface.py

@@ -2,6 +2,7 @@
 from datetime import date, datetime, time, timezone
 from functools import lru_cache
 from pathlib import Path
+from typing import Optional, Sequence
 from weakref import WeakKeyDictionary
 
 import h5py
@@ -113,8 +114,7 @@ class CalCatApi(metaclass=ClientWrapper):
         """Encode operating condition to CalCat API format.
 
         Args:
-            caldata (CalibrationData): Calibration data instance used to
-                interface with database.
+            condition (dict): Mapping of parameter DB name to value
 
         Returns:
             (dict) Operating condition for use in CalCat API.
@@ -192,6 +192,19 @@ class CalCatApi(metaclass=ClientWrapper):
 
         return resp_calibration["data"]["id"]
 
+    @lru_cache()
+    def calibration_name(self, calibration_id):
+        """Name for a calibration in CalCat."""
+
+        resp_calibration = Calibration.get_by_id(
+            self.client, calibration_id
+        )
+
+        if not resp_calibration["success"]:
+            raise CalCatError(resp_calibration)
+
+        return resp_calibration["data"]["name"]
+
     @lru_cache()
     def parameter_id(self, param_name):
         """ID for an operating condition parameter in CalCat."""
@@ -203,6 +216,33 @@ class CalCatApi(metaclass=ClientWrapper):
 
         return resp_parameter["data"]["id"]
 
+    def _closest_ccv_by_time_by_condition(
+            self,
+            detector_name: str,
+            calibration_ids: Sequence[int],
+            condition: dict,
+            karabo_da: Optional[str] = None,
+            event_at=None,
+            pdu_snapshot_at=None,
+    ):
+        resp = CalibrationConstantVersion.get_closest_by_time_by_detector_conditions(
+            self.client,
+            detector_name,
+            calibration_ids,
+            self.format_cond(condition),
+            karabo_da=karabo_da or "",
+            event_at=self.format_time(event_at),
+            pdu_snapshot_at=self.format_time(pdu_snapshot_at),
+        )
+
+        if not resp["success"]:
+            if resp["status_code"] == 200:
+                # calibration_client turns empty response into an error
+                return []
+            raise CalCatError(resp)
+
+        return resp["data"]
+
     def closest_ccv_by_time_by_condition(
         self,
         detector_name,
@@ -284,20 +324,16 @@ class CalCatApi(metaclass=ClientWrapper):
         # afterwards, if necessary.
         karabo_da = next(iter(da_to_modname)) if len(da_to_modname) == 1 else '',
 
-        resp_versions = CalibrationConstantVersion.get_closest_by_time_by_detector_conditions(  # noqa
-            self.client,
+        resp_data = self._closest_ccv_by_time_by_condition(
             detector_name,
             calibration_ids,
-            self.format_cond(condition),
+            condition,
             karabo_da=karabo_da,
             event_at=event_at,
             pdu_snapshot_at=pdu_snapshot_at,
         )
 
-        if not resp_versions["success"]:
-            raise CalCatError(resp_versions)
-
-        for ccv in resp_versions["data"]:
+        for ccv in resp_data:
             try:
                 mod = da_to_modname[ccv['physical_detector_unit']['karabo_da']]
             except KeyError:

tests/test_calcat_interface2.py

+import numpy as np
+import pytest
+import xarray as xr
+
+from cal_tools.calcat_interface2 import (
+    AGIPDConditions,
+    CalibrationData,
+    DSSCConditions,
+    LPDConditions,
+    SingleConstant,
+)
+
+
+@pytest.mark.requires_gpfs
+def test_AGIPD_CalibrationData_metadata():
+    """Test CalibrationData with AGIPD condition"""
+    cond = AGIPDConditions(
+        # From: https://in.xfel.eu/calibration/calibration_constants/5754#condition
+        sensor_bias_voltage=300,  # V
+        memory_cells=352,
+        acquisition_rate=2.2,  # MHz
+        gain_mode=0,
+        gain_setting=1,
+        integration_time=12,
+        source_energy=9.2,
+    )
+    agipd_cd = CalibrationData.from_condition(
+        cond,
+        "MID_DET_AGIPD1M-1",
+        event_at="2022-09-01 13:26:48.00",
+        calibrations=["Offset", "SlopesFF"],
+    )
+    assert agipd_cd.detector_name == "MID_DET_AGIPD1M-1"
+    assert "Offset" in agipd_cd
+    assert set(agipd_cd["Offset"].constants) == {f"AGIPD{m:02}" for m in range(16)}
+    assert isinstance(agipd_cd["Offset", "AGIPD00"], SingleConstant)
+    assert agipd_cd["Offset", "Q1M2"] == agipd_cd["Offset", "AGIPD01"]
+
+
+@pytest.mark.requires_gpfs
+def test_AGIPD_merge():
+    cond = AGIPDConditions(
+        # From: https://in.xfel.eu/calibration/calibration_constants/5754#condition
+        sensor_bias_voltage=300,  # V
+        memory_cells=352,
+        acquisition_rate=2.2,  # MHz
+        gain_mode=0,
+        gain_setting=1,
+        integration_time=12,
+        source_energy=9.2,
+    )
+    agipd_cd = CalibrationData.from_condition(
+        cond,
+        "MID_DET_AGIPD1M-1",
+        event_at="2022-09-01 13:26:48.00",
+        calibrations=["Offset", "SlopesFF"],
+    )
+
+    modnos_q1 = list(range(0, 4))
+    modnos_q4 = list(range(12, 16))
+    merged = agipd_cd.select_modules(modnos_q1).merge(
+        agipd_cd.select_modules(modnos_q4)
+    )
+    assert merged.module_nums == modnos_q1 + modnos_q4
+
+    offset_only = agipd_cd.select_calibrations(["Offset"])
+    slopes_only = agipd_cd.select_calibrations(["SlopesFF"])
+    assert set(offset_only) == {"Offset"}
+    assert set(slopes_only) == {"SlopesFF"}
+    merged_cals = offset_only.merge(slopes_only)
+    assert set(merged_cals) == {"Offset", "SlopesFF"}
+    assert merged_cals.module_nums == list(range(16))
+
+
+@pytest.mark.requires_gpfs
+def test_AGIPD_CalibrationData_metadata_SPB():
+    """Test CalibrationData with AGIPD condition"""
+    cond = AGIPDConditions(
+        sensor_bias_voltage=300,
+        memory_cells=352,
+        acquisition_rate=1.1,
+        integration_time=12,
+        source_energy=9.2,
+        gain_mode=0,
+        gain_setting=0,
+    )
+    agipd_cd = CalibrationData.from_condition(
+        cond,
+        "SPB_DET_AGIPD1M-1",
+        event_at="2020-01-07 13:26:48.00",
+    )
+    assert "Offset" in agipd_cd
+    assert set(agipd_cd["Offset"].constants) == {f"AGIPD{m:02}" for m in range(16)}
+    assert agipd_cd["Offset"].module_nums == list(range(16))
+    assert agipd_cd["Offset"].qm_names == [
+        f"Q{(m // 4) + 1}M{(m % 4) + 1}" for m in range(16)
+    ]
+    assert isinstance(agipd_cd["Offset", 0], SingleConstant)
+
+
+@pytest.mark.requires_gpfs
+def test_AGIPD_load_data():
+    cond = AGIPDConditions(
+        sensor_bias_voltage=300,
+        memory_cells=352,
+        acquisition_rate=1.1,
+        integration_time=12,
+        source_energy=9.2,
+        gain_mode=0,
+        gain_setting=0,
+    )
+    agipd_cd = CalibrationData.from_condition(
+        cond,
+        "SPB_DET_AGIPD1M-1",
+        event_at="2020-01-07 13:26:48.00",
+    )
+    arr = agipd_cd["Offset"].select_modules(list(range(4))).xarray()
+    assert arr.shape == (4, 128, 512, 352, 3)
+    assert arr.dims[0] == "module"
+    np.testing.assert_array_equal(arr.coords["module"], np.arange(0, 4))
+    assert arr.dtype == np.float64
+
+    # Load parallel
+    arr_p = agipd_cd["Offset"].select_modules(list(range(4))).xarray(parallel=4)
+    xr.testing.assert_identical(arr_p, arr)
+
+
+@pytest.mark.requires_gpfs
+def test_DSSC_modules_missing():
+    dssc_cd = CalibrationData.from_condition(
+        DSSCConditions(sensor_bias_voltage=100, memory_cells=600),
+        "SQS_DET_DSSC1M-1",
+        event_at="2023-11-29 00:00:00",
+    )
+    # DSSC was used with only 3 quadrants at this point
+    modnos = list(range(4)) + list(range(8, 16))
+    assert dssc_cd.aggregator_names == [f"DSSC{m:02}" for m in modnos]
+    assert dssc_cd.module_nums == modnos
+    assert dssc_cd.qm_names == [f"Q{(m // 4) + 1}M{(m % 4) + 1}" for m in modnos]
+
+    offset = dssc_cd["Offset"]
+    assert offset.module_nums == modnos
+
+    # test ModulesConstantVersions.select_modules()
+    modnos_q3 = list(range(8, 12))
+    aggs_q3 = [f"DSSC{m:02}" for m in modnos_q3]
+    qm_q3 = [f"Q3M{i}" for i in range(1, 5)]
+    assert offset.select_modules(modnos_q3).module_nums == modnos_q3
+    assert offset.select_modules(aggregator_names=aggs_q3).module_nums == modnos_q3
+    assert offset.select_modules(qm_names=qm_q3).module_nums == modnos_q3
+
+    # test CalibrationData.select_modules()
+    assert dssc_cd.select_modules(modnos_q3).module_nums == modnos_q3
+    assert dssc_cd.select_modules(aggregator_names=aggs_q3).module_nums == modnos_q3
+    assert dssc_cd.select_modules(qm_names=qm_q3).module_nums == modnos_q3
+
+
+@pytest.mark.requires_gpfs
+def test_LPD_constant_missing():
+    lpd_cd = CalibrationData.from_condition(
+        LPDConditions(memory_cells=200, sensor_bias_voltage=250),
+        "FXE_DET_LPD1M-1",
+        event_at="2022-05-22T02:00:00",
+    )
+    # Constants are missing for 1 module (LPD05), but it was still included in
+    # the PDUs for the detector, so it should still appear in the lists.
+    assert lpd_cd.aggregator_names == [f"LPD{m:02}" for m in range(16)]
+    assert lpd_cd.module_nums == list(range(16))
+    assert lpd_cd.qm_names == [f"Q{(m // 4) + 1}M{(m % 4) + 1}" for m in range(16)]
+
+    # When we look at a specific constant, module LPD05 is missing
+    modnos_w_constant = list(range(0, 5)) + list(range(6, 16))
+    assert lpd_cd["Offset"].module_nums == modnos_w_constant
+
+    # Test CalibrationData.require_constant()
+    assert lpd_cd.require_calibrations(["Offset"]).module_nums == modnos_w_constant
+
+
+@pytest.mark.requires_gpfs
+def test_AGIPD_CalibrationData_report():
+    """Test CalibrationData with data from report"""
+    # Report ID: https://in.xfel.eu/calibration/reports/3757
+    agipd_cd = CalibrationData.from_report(3757)
+    assert agipd_cd.detector_name == "SPB_DET_AGIPD1M-1"
+    assert set(agipd_cd) == {"Offset", "Noise", "ThresholdsDark", "BadPixelsDark"}
+    assert agipd_cd.aggregator_names == [f"AGIPD{n:02}" for n in range(16)]
+    assert isinstance(agipd_cd["Offset", "AGIPD00"], SingleConstant)