Merge branch 'feat/new_calcat_interface' into 'master'

New calcat interface

See merge request detectors/pycalibration!738

docs/source/calcat_interface.rst

+CALCAT Interface
+================
+
+.. module:: cal_tools.calcat_interface
+
+.. class:: CalCatError
+
+.. class:: CalibrationData
+
+    .. attribute:: metadata
+
+    .. attribute:: ndarray
+
+    .. attribute:: ndarray_map
+
+.. class:: SplitConditionCalibrationData
+
+.. class:: LPD_CalibrationData
+
+.. class:: DSSC_CalibrationData
+
+.. class:: JUNGFRAU_CalibrationData
+
+.. class:: PNCCD_CalibrationData
+
+.. class:: EPIX100_CalibrationData
+
+.. class:: GOTTHARD2_CalibrationData
+

docs/source/index.rst

@@ -57,6 +57,7 @@ Documentation contents:
 
    xfel_calibrate_conf
    cal_tools_algorithms
+   calcat_interface
 
 .. toctree::
    :caption: Development

src/cal_tools/restful_config.py

@@ -13,3 +13,20 @@ restful_config = Dynaconf(
     ],
     merge_enabled=True,
 )
+
+
+def calibration_client():
+    from calibration_client import CalibrationClient
+
+    # Create client for CalCat.
+    calcat_config = restful_config.get('calcat')
+    return CalibrationClient(
+        base_api_url=calcat_config['base-api-url'],
+        use_oauth2=calcat_config['use-oauth2'],
+        client_id=calcat_config['user-id'],
+        client_secret=calcat_config['user-secret'],
+        user_email=calcat_config['user-email'],
+        token_url=calcat_config['token-url'],
+        refresh_url=calcat_config['refresh-url'],
+        auth_url=calcat_config['auth-url'],
+        scope='')

tests/test_calcat_interface.py

+"""
+TEST CALCAT INTERFACE:
+
+All the tests run with client = None
+"""
+
+import numpy as np
+import pytest
+
+from cal_tools.calcat_interface import (
+    AGIPD_CalibrationData,
+    DSSC_CalibrationData,
+    JUNGFRAU_CalibrationData,
+    LPD_CalibrationData,
+)
+
+
+@pytest.mark.requires_gpfs
+@pytest.mark.parametrize(
+    "mod,mod_naming",
+    [
+        ("AGIPD00", "da"),
+        ("Q1M1", "qm"),
+        (0, "modno"),
+
+    ]
+)
+def test_AGIPD_CalibrationData_metadata(mod, mod_naming):
+    """Test CalibrationData through AGIPD_CalibrationData
+    and test metadata() method.
+
+    Args:
+        mod (str): Module name
+        mod_naming (str): Selected module naming
+    """
+    agipd_md = AGIPD_CalibrationData(
+        detector_name="SPB_DET_AGIPD1M-1",
+        sensor_bias_voltage=300,
+        memory_cells=352,
+        acquisition_rate=1.1,
+        integration_time=12,
+        source_energy=9.2,
+        gain_mode=0,
+        gain_setting=0,
+        event_at="2020-01-07 13:26:48.00",
+        modules=[mod],
+        client=None,
+        module_naming=mod_naming
+    ).metadata()
+    assert isinstance(agipd_md, dict)
+    # Check correct module naming (da, qm, modno)
+    assert list(agipd_md.keys())[0] == mod
+    # closest_ccv_by_time_by_condition
+    assert [
+        "begin_validity_at",
+        "cc_id", "cc_name",
+        "ccv_id", "ccv_name",
+        "condition_id", "dataset",
+        "end_idx", "end_validity_at",
+        "path", "physical_name",
+        "raw_data_location", "start_idx",
+    ] == sorted(list(agipd_md[mod]["Offset"].keys()))
+
+
+@pytest.mark.parametrize(
+    "mod,mod_naming",
+    [
+        ("DSSC00", "da"),
+        ("Q1M1", "qm"),
+        (0, "modno"),
+
+    ]
+)
+def test_DSSC_physical_detector_units(mod, mod_naming):
+    """Test physical_detector_units property and DSSC_CalibrationData.
+
+    Args:
+        mod (str): Module name
+        mod_naming (str): Selected module naming
+    """
+    dssc_cal = DSSC_CalibrationData(
+        detector_name="SCS_DET_DSSC1M-1",
+        sensor_bias_voltage=300,
+        memory_cells=400,
+        pulse_id_checksum=None,
+        acquisition_rate=None,
+        target_gain=None,
+        encoded_gain=None,
+        event_at="2020-01-07 13:26:48.00",
+        modules=[mod],
+        module_naming=mod_naming
+    )
+    pdus = dssc_cal.physical_detector_units
+    assert isinstance(pdus, dict)
+    assert len(pdus) == 16
+    assert list(pdus.keys())[0] == mod
+
+
+@pytest.mark.requires_gpfs
+@pytest.mark.parametrize(
+    "mod,mod_naming",
+    [
+        ("JNGFR01", "da"),
+        (1, "modno"),
+
+    ]
+)
+def test_JUNGFRAU_CalibrationData_ndarray(mod, mod_naming):
+    """Test ndarray() through JUNGFRAU_CalibrationData.
+
+    Args:
+        mod (str): Module name
+        mod_naming (str): Selected module naming
+    """
+
+    jf_cal = JUNGFRAU_CalibrationData(
+        detector_name="SPB_IRDA_JF4M",
+        sensor_bias_voltage=180,
+        memory_cells=1,
+        integration_time=10,
+        gain_setting=0,
+        gain_mode=0,
+        modules=[mod],
+        module_naming=mod_naming,
+        event_at="2022-07-11 13:26:48.00",
+    )
+    assert isinstance(jf_cal.ndarray(mod, "Offset10Hz"), np.ndarray)
+
+
+@pytest.mark.requires_gpfs
+def test_LPD_CalibrationData_ndarray():
+    """Test ndarray_map() through LPD_CalibrationData.
+
+    Args:
+        mod (str): Module name
+        mod_naming (str): Selected module naming
+    """
+
+    lpd_cal = LPD_CalibrationData(
+        detector_name="FXE_DET_LPD1M-1",
+        sensor_bias_voltage=250,
+        memory_cells=512,
+        feedback_capacitor=5.0,
+        event_at="2022-07-11 13:26:48.00",
+    )
+    const_map = lpd_cal.ndarray_map()
+    assert isinstance(const_map, dict)
+    assert "LPD00" in const_map.keys()
+    assert "Offset" in const_map["LPD00"].keys()
+    assert isinstance(const_map["LPD00"]["Offset"], np.ndarray)