diff --git a/src/cal_tools/agipdlib.py b/src/cal_tools/agipdlib.py
index b3198241bdf17f7338904fa09be83291e146ba6e..9405781c929558f7f6dc84e50bcfdb19399883d8 100644
--- a/src/cal_tools/agipdlib.py
+++ b/src/cal_tools/agipdlib.py
@@ -261,6 +261,7 @@ class AgipdCtrl:
raise ValueError(error)
return default_voltage
+
def get_integration_time(self) -> int:
"""Read integration time from the FPGA device.
diff --git a/tests/conftest.py b/tests/conftest.py
index fd71bd602e4b7921a05c95639f2ddce0c797f558..e6ee787debd2a8ca9e4443a68c8ef64ed6a7e14f 100644
--- a/tests/conftest.py
+++ b/tests/conftest.py
@@ -1,10 +1,36 @@
import socket
from functools import lru_cache
from pathlib import Path
+from tempfile import TemporaryDirectory
+import extra_data.tests.make_examples as make_examples
import pytest
+@pytest.fixture(scope='session')
+def mock_agipd1m_run():
+ with TemporaryDirectory() as td:
+ make_examples.make_agipd1m_run(td)
+ yield td
+
+
+@pytest.fixture(scope='session')
+def mock_agipd1m_run_old():
+ # No gain.value or bunchStructure.repetitionRate in /CONTROL/
+ with TemporaryDirectory() as td:
+ make_examples.make_agipd1m_run(
+ td, rep_rate=False, gain_setting=False, integration_time=False)
+ yield td
+
+
+@pytest.fixture(scope='session')
+def mock_agipd500k_run():
+ # No gain.value or bunchStructure.repetitionRate in /CONTROL/
+ with TemporaryDirectory() as td:
+ make_examples.make_agipd500k_run(td)
+ yield td
+
+
def pytest_addoption(parser):
parser.addoption(
"--no-gpfs",
diff --git a/tests/test_agipdlib.py b/tests/test_agipdlib.py
new file mode 100644
index 0000000000000000000000000000000000000000..dc1dd92c497bad2d2a73b08a9b486f6ccb593493
--- /dev/null
+++ b/tests/test_agipdlib.py
@@ -0,0 +1,209 @@
+from datetime import datetime
+
+from extra_data import RunDirectory
+
+from cal_tools.agipdlib import AgipdCtrl
+
+SPB_AGIPD_INST_SRC = 'SPB_DET_AGIPD1M-1/DET/0CH0:xtdf'
+CTRL_SRC = 'SPB_IRU_AGIPD1M1/MDL/FPGA_COMP'
+
+
+def test_get_acq_rate_run(mock_agipd1m_run):
+ # Current up to date data with acq_rate stored
+ # as repetition rate in slow data.
+ agipd_ctrl = AgipdCtrl(
+ run_dc=RunDirectory(mock_agipd1m_run),
+ image_src=SPB_AGIPD_INST_SRC,
+ ctrl_src=CTRL_SRC)
+
+ acq_rate = agipd_ctrl.get_acq_rate()
+ assert isinstance(acq_rate, float)
+ assert acq_rate == 4.5
+
+
+def test_get_acq_rate_fast(mock_agipd1m_run):
+ # Current up to date data with acq_rate stored
+ # as repetition rate in slow data.
+ agipd_ctrl = AgipdCtrl(
+ run_dc=RunDirectory(mock_agipd1m_run),
+ image_src=SPB_AGIPD_INST_SRC,
+ ctrl_src=CTRL_SRC)
+
+ acq_rate = agipd_ctrl._get_acq_rate_run()
+ assert isinstance(acq_rate, float)
+ assert acq_rate == 4.5
+
+
+def test_get_acq_rate_instr(mock_agipd1m_run_old):
+
+ # Old data with no stored acq_rate in slow data.
+ # Reading acq_rate from fast data.
+ agipd_ctrl = AgipdCtrl(
+ run_dc=RunDirectory(mock_agipd1m_run_old),
+ image_src=SPB_AGIPD_INST_SRC,
+ ctrl_src=None)
+ acq_rate = agipd_ctrl.get_acq_rate()
+ assert isinstance(acq_rate, float)
+ assert acq_rate == 4.5
+
+
+def test_get_acq_rate(mock_agipd1m_run, mock_agipd1m_run_old):
+
+ # Current up to date data with acq_rate stored
+ # as repetition rate in slow data.
+ agipd_ctrl = AgipdCtrl(
+ run_dc=RunDirectory(mock_agipd1m_run),
+ image_src=SPB_AGIPD_INST_SRC,
+ ctrl_src=CTRL_SRC)
+
+ acq_rate = agipd_ctrl._get_acq_rate_instr()
+ assert isinstance(acq_rate, float)
+ assert acq_rate == 4.5
+
+ # Old data with no stored acq_rate in slow data.
+ # Reading acq_rate from fast data.
+ agipd_ctrl = AgipdCtrl(
+ run_dc=RunDirectory(mock_agipd1m_run_old),
+ image_src=SPB_AGIPD_INST_SRC,
+ ctrl_src=None)
+ acq_rate = agipd_ctrl.get_acq_rate()
+ assert isinstance(acq_rate, float)
+ assert acq_rate == 4.5
+
+
+def test_get_num_cells(mock_agipd1m_run):
+
+ # Reading number of memory cells from fast data.
+ agipd_ctrl = AgipdCtrl(
+ run_dc=RunDirectory(mock_agipd1m_run),
+ image_src=SPB_AGIPD_INST_SRC,
+ ctrl_src=None)
+ mem_cells = agipd_ctrl.get_num_cells()
+ assert isinstance(mem_cells, int)
+ assert mem_cells == 64
+
+
+def test_get_gain_setting_run(mock_agipd1m_run):
+ # Reading gain setting from slow data.
+ agipd_ctrl = AgipdCtrl(
+ run_dc=RunDirectory(mock_agipd1m_run),
+ image_src=SPB_AGIPD_INST_SRC,
+ ctrl_src=CTRL_SRC)
+ gain_setting = agipd_ctrl._get_gain_setting_run()
+ assert isinstance(gain_setting, int)
+ assert gain_setting == 0
+
+
+def test_get_gain_setting_run_old(mock_agipd1m_run_old):
+
+ # Reading gain setting from setupr and .....
+ agipd_ctrl = AgipdCtrl(
+ run_dc=RunDirectory(mock_agipd1m_run_old),
+ image_src=SPB_AGIPD_INST_SRC,
+ ctrl_src=CTRL_SRC)
+ gain_setting = agipd_ctrl._get_gain_setting_run_old()
+ assert isinstance(gain_setting, int)
+ assert gain_setting == 0
+
+
+def test_get_gain_setting(mock_agipd1m_run, mock_agipd1m_run_old):
+
+ # Reading gain setting from slow data.
+ agipd_ctrl = AgipdCtrl(
+ run_dc=RunDirectory(mock_agipd1m_run),
+ image_src=SPB_AGIPD_INST_SRC,
+ ctrl_src=CTRL_SRC)
+ gain_setting = agipd_ctrl.get_gain_setting()
+ assert isinstance(gain_setting, int)
+ assert gain_setting == 0
+
+ # Reading gain setting from setupr and .....
+ agipd_ctrl = AgipdCtrl(
+ run_dc=RunDirectory(mock_agipd1m_run_old),
+ image_src=SPB_AGIPD_INST_SRC,
+ ctrl_src=CTRL_SRC)
+ gain_setting = agipd_ctrl.get_gain_setting()
+ assert isinstance(gain_setting, int)
+ assert gain_setting == 0
+
+ # Old data without gain setting.
+ agipd_ctrl = AgipdCtrl(
+ run_dc=RunDirectory(mock_agipd1m_run_old),
+ image_src=SPB_AGIPD_INST_SRC,
+ ctrl_src=CTRL_SRC)
+
+ gain_setting = agipd_ctrl.get_gain_setting(
+ creation_time=datetime.strptime(
+ "2019-12-16T08:52:25",
+ "%Y-%m-%dT%H:%M:%S"))
+
+ assert gain_setting is None
+
+
+def test_get_bias_voltage(
+ mock_agipd1m_run,
+ mock_agipd500k_run,
+ mock_agipd1m_run_old
+):
+ # Read bias voltage for HED_DET_AGIPD500K from slow data.
+ agipd_ctrl = AgipdCtrl(
+ run_dc=RunDirectory(mock_agipd500k_run),
+ image_src=SPB_AGIPD_INST_SRC,
+ ctrl_src=CTRL_SRC)
+ bias_voltage = agipd_ctrl.get_bias_voltage(
+ karabo_id_control="HED_EXP_AGIPD500K2G")
+ assert isinstance(bias_voltage, int)
+ assert bias_voltage == 200
+
+ # Read bias voltage for SPB_DET_AGIPD1M-1 and MID_DET_AGIPD1M-1
+ # from slow data.
+ agipd_ctrl = AgipdCtrl(
+ run_dc=RunDirectory(mock_agipd1m_run),
+ image_src=SPB_AGIPD_INST_SRC,
+ ctrl_src=CTRL_SRC)
+ bias_voltage = agipd_ctrl.get_bias_voltage(
+ karabo_id_control="SPB_IRU_AGIPD1M1")
+ assert isinstance(bias_voltage, int)
+ assert bias_voltage == 300
+
+ # Read bias voltage for SPB_DET_AGIPD1M-1 and MID_DET_AGIPD1M-1
+ # from slow data.
+ agipd_ctrl = AgipdCtrl(
+ run_dc=RunDirectory(mock_agipd1m_run_old),
+ image_src=SPB_AGIPD_INST_SRC,
+ ctrl_src=CTRL_SRC)
+ bias_voltage = agipd_ctrl.get_bias_voltage(
+ karabo_id_control="SPB_IRU_AGIPD1M1")
+ assert isinstance(bias_voltage, int)
+ assert bias_voltage == 300
+
+
+def test_get_integration_time(mock_agipd1m_run, mock_agipd1m_run_old):
+ # Read integration time.
+ agipd_ctrl = AgipdCtrl(
+ run_dc=RunDirectory(mock_agipd1m_run),
+ image_src=SPB_AGIPD_INST_SRC,
+ ctrl_src=CTRL_SRC)
+ integration_time = agipd_ctrl.get_integration_time()
+ assert isinstance(integration_time, int)
+ assert integration_time == 15
+
+ # Integration time is not available in slow data.
+ agipd_ctrl = AgipdCtrl(
+ run_dc=RunDirectory(mock_agipd1m_run_old),
+ image_src=SPB_AGIPD_INST_SRC,
+ ctrl_src=CTRL_SRC)
+ integration_time = agipd_ctrl.get_integration_time()
+ assert isinstance(integration_time, int)
+ assert integration_time == 12
+
+
+def test_get_gain_mode(mock_agipd1m_run):
+ # Read gain mode from slow data.
+ agipd_ctrl = AgipdCtrl(
+ run_dc=RunDirectory(mock_agipd1m_run),
+ image_src=SPB_AGIPD_INST_SRC,
+ ctrl_src=CTRL_SRC)
+ gain_mode = agipd_ctrl.get_gain_mode()
+ assert isinstance(gain_mode, int)
+ assert gain_mode == 0