diff --git a/cal_tools/cal_tools/agipdlib.py b/cal_tools/cal_tools/agipdlib.py
index a098b07b83a7af795b8000c0aced7f4eca62c6bb..1b8da40f52e87768da0b880d9a7da41f7d75e6a5 100644
--- a/cal_tools/cal_tools/agipdlib.py
+++ b/cal_tools/cal_tools/agipdlib.py
@@ -17,7 +17,7 @@ from cal_tools.agipdutils import (
)
from cal_tools.enums import AgipdGainMode, BadPixels, SnowResolution
from cal_tools.tools import get_constant_from_db_and_time
-from iCalibrationDB import Conditions
+from iCalibrationDB import Conditions, Constants
from cal_tools.cython import agipdalgs as calgs
@@ -1272,7 +1272,8 @@ class AgipdCorrections:
return when
def initialize_from_db(self, karabo_id: str, karabo_da: str,
- cal_db_interface: str, creation_time: str,
+ cal_db_interface: str,
+ creation_time: 'datetime.datetime',
memory_cells: float, bias_voltage: int,
photon_energy: float, gain_setting: float,
acquisition_rate: float, module_idx: int,
diff --git a/cal_tools/cal_tools/tools.py b/cal_tools/cal_tools/tools.py
index 7572e7d18911d92abfba522c31dcf8f453745461..587bb8970a0a645d8bfbe6113669e4c42f1e4953 100644
--- a/cal_tools/cal_tools/tools.py
+++ b/cal_tools/cal_tools/tools.py
@@ -398,6 +398,10 @@ def get_pdu_from_db(karabo_id: str, karabo_da: Union[str, list],
:param karabo_id: Karabo identifier.
:param karabo_da: Karabo data aggregator.
+ :param constant: Calibration constant object to
+ intialize CalibrationConstantMetadata class.
+ :param condition: Detector condition object to
+ intialize CalibrationConstantMetadata class.
:param cal_db_interface: Interface string, e.g. "tcp://max-exfl016:8015".
:param snapshot_at: Database snapshot.
:param timeout: Calibration Database timeout.
@@ -570,6 +574,8 @@ def send_to_db(db_module: str, karabo_id: str, constant, condition,
:param constant: Calibration constant known for given detector
:param condition: Calibration condition
:param file_loc: Location of raw data.
+ :param report_path: xfel-calbrate report path to inject along with
+ the calibration constant versions to the database.
:param cal_db_interface: Interface string, e.g. "tcp://max-exfl016:8015"
:param creation_time: Latest time for constant to be created
:param timeout: Timeout for zmq request
diff --git a/tests/test_cal_tools.py b/tests/test_cal_tools.py
index c7656d1770a4bec28f935eae2c282306940b1b60..7ee13b577056da2017de34bd4a684d01d0fa120b 100644
--- a/tests/test_cal_tools.py
+++ b/tests/test_cal_tools.py
@@ -2,8 +2,14 @@ from datetime import datetime
from pathlib import Path
import pytest
+from cal_tools.agipdlib import AgipdCorrections
from cal_tools.plotting import show_processed_modules
-from cal_tools.tools import get_dir_creation_date, module_index_to_qm
+from cal_tools.tools import (
+ get_dir_creation_date,
+ get_pdu_from_db,
+ module_index_to_qm,
+)
+from iCalibrationDB import Conditions, Constants
def test_show_processed_modules():
@@ -32,6 +38,108 @@ def test_dir_creation_date():
assert str(date) == '2019-12-16 08:52:25.196603'
+# AGIPD dark offset metadata
+constant = Constants.AGIPD.Offset()
+mem_cells = 352
+bias_voltage = 300
+acq_rate = 1.1
+gain_setting = 0
+photon_energy = 9.2
+condition = Conditions.Dark.AGIPD(memory_cells=mem_cells,
+ bias_voltage=bias_voltage,
+ acquisition_rate=acq_rate,
+ gain_setting=gain_setting)
+cal_db_interface = "tcp://max-exfl017:8020"
+
+
+def test_get_pdu_from_db():
+
+ # A karabo_da str returns a list of one element.
+ pdu_dict = get_pdu_from_db(karabo_id="TEST_DET_CI-2",
+ karabo_da="TEST_DAQ_DA_01",
+ constant=constant,
+ condition=condition,
+ cal_db_interface=cal_db_interface,
+ snapshot_at="2021-03-01 09:44:00+00:00",
+ timeout=30000)
+ assert len(pdu_dict) == 1
+ assert pdu_dict[0] == 'PHYSICAL_DETECTOR_UNIT-1_DO_NOT_DELETE'
+
+ # A list of karabo_das to return thier PDUs, if available.
+ pdu_dict = get_pdu_from_db(karabo_id="TEST_DET_CI-2",
+ karabo_da=["TEST_DAQ_DA_01", "TEST_DAQ_DA_02",
+ "UNAVAILABLE_DA"],
+ constant=constant,
+ condition=condition,
+ cal_db_interface=cal_db_interface,
+ snapshot_at="2021-03-01 09:44:00+00:00",
+ timeout=30000)
+
+ assert pdu_dict == ['PHYSICAL_DETECTOR_UNIT-1_DO_NOT_DELETE',
+ 'PHYSICAL_DETECTOR_UNIT-2_DO_NOT_DELETE',
+ None]
+
+ # "all" is used to return all corresponding units for a karabo_id.
+ pdu_dict = get_pdu_from_db(karabo_id="TEST_DET_CI-2",
+ karabo_da="all",
+ constant=constant,
+ condition=condition,
+ cal_db_interface="tcp://max-exfl017:8020",
+ snapshot_at="2021-03-01 09:44:00+00:00",
+ timeout=30000)
+
+ assert len(pdu_dict) == 3
+ assert pdu_dict == ['PHYSICAL_DETECTOR_UNIT-1_DO_NOT_DELETE',
+ 'PHYSICAL_DETECTOR_UNIT-2_DO_NOT_DELETE',
+ 'PHYSICAL_DETECTOR_UNIT-3_DO_NOT_DELETE']
+
+
+def test_initialize_from_db():
+ creation_time = datetime.strptime("2020-01-07 13:26:48.00",
+ "%Y-%m-%d %H:%M:%S.%f")
+
+ agipd_corr = AgipdCorrections(max_cells=mem_cells,
+ max_pulses=[0, 500, 1])
+
+ agipd_corr.allocate_constants(modules=[0],
+ constant_shape=(3, mem_cells, 512, 128))
+
+ dark_const_time_dict = agipd_corr.initialize_from_db(
+ karabo_id="TEST_DET_CI-2",
+ karabo_da="TEST_DAQ_DA_01",
+ cal_db_interface=cal_db_interface,
+ creation_time=creation_time,
+ memory_cells=mem_cells,
+ bias_voltage=bias_voltage,
+ photon_energy=photon_energy,
+ gain_setting=gain_setting,
+ acquisition_rate=acq_rate,
+ module_idx=0,
+ only_dark=False,
+ )
+
+ assert dark_const_time_dict == {
+ "Offset": None,
+ "Noise": None,
+ "ThresholdsDark": None,
+ "BadPixelsDark": None,
+ }
+
+ dark_const_time_dict = agipd_corr.initialize_from_db(
+ karabo_id="SPB_DET_AGIPD1M-1",
+ karabo_da="AGIPD00",
+ cal_db_interface=cal_db_interface,
+ creation_time=creation_time,
+ memory_cells=mem_cells, bias_voltage=bias_voltage,
+ photon_energy=photon_energy, gain_setting=gain_setting,
+ acquisition_rate=acq_rate, module_idx=0,
+ only_dark=False)
+ # A retrieved constant has a value of datetime creation_time
+ assert isinstance(dark_const_time_dict["Offset"], datetime)
+ assert list(dark_const_time_dict.keys()) == [
+ "Offset", "Noise", "ThresholdsDark", "BadPixelsDark"]
+
+
def test_module_index_to_qm():
assert module_index_to_qm(0) == 'Q1M1'