Skip to content
Snippets Groups Projects

Compare revisions

Changes are shown as if the source revision was being merged into the target revision. Learn more about comparing revisions.

Source

Select target project
No results found

Target

Select target project
  • calibration/pycalibration
1 result
Show changes
Commits on Source (2)
Hide whitespace changes
Inline Side-by-side
src/cal_tools/agipdlib.py
View file @ 5dceb0c9
import os
import posixpath
import zlib
from datetime import datetime
from multiprocessing.pool import ThreadPool
from typing import Any, Dict, List, Optional, Tuple
......@@ -26,6 +27,7 @@ from cal_tools.enums import AgipdGainMode, BadPixels, SnowResolution
from cal_tools.h5_copy_except import h5_copy_except_paths
from cal_tools.tools import get_constant_from_db_and_time
class AgipdCtrl:
def __init__(
self,
......@@ -34,22 +36,26 @@ class AgipdCtrl:
ctrl_src: str,
raise_error: bool = True,
):
"""
Initialize AgipdCondition class to read all required AGIPD parameters.
""" Initialize AgipdCondition class to read
all required AGIPD parameters.
:param run_dc: Run data collection with expected sources
to read needed parameters.
:param image_src: H5 source for image data.
:param ctrl_src: H5 source for control (slow) data.
:param raise_error: Boolean to raise errors for missing
sources and keys.
"""
self.run_dc = run_dc
self.image_src = image_src
self.ctrl_src = ctrl_src
self.raise_error = raise_error
def get_num_cells(self) -> Optional[int]:
"""
:return mem_cells: Number of memory cells.
return None, if no data available.
"""Read number of memory cells from fast data.
:return mem_cells: Number of memory cells
return None, if no data available.
"""
cells = np.squeeze(
self.run_dc[
......@@ -62,38 +68,27 @@ class AgipdCtrl:
dists = [abs(o - maxcell) for o in options]
return options[np.argmin(dists)]
def get_acq_rate(self) -> Optional[float]:
"""Get the acquisition rate from said detector module.
If the data is available from the middlelayer FPGA_COMP device,
then it is retrieved from there.
If not, the rate is calculated from two different pulses time.
The first entry is deliberately not used, as the detector just began
operating, and it might have skipped a train.
:return acq_rate: the acquisition rate.
return None, if not available.
"""
def _get_acq_rate_run(self) -> Optional[float]:
"""Get acquisition (repetition) rate from RUN source."""
# Attempt to look for acquisition rate in slow data
rep_rate_src = (
self.ctrl_src, "bunchStructure.repetitionRate.value")
if (
rep_rate_src[0] in self.run_dc.all_sources and
rep_rate_src[1] in self.run_dc.keys_for_source(rep_rate_src[0])
):
# The acquisition rate value is stored in a 1D array of type
# float.
# It is desired to loose precision here because the usage is
# about bucketing the rate for managing meta-data.
return round(float(self.run_dc[rep_rate_src].as_single_value()), 1)
return round(float(self.run_dc.get_run_value(*rep_rate_src)), 1)
def _get_acq_rate_instr(self) -> Optional[float]:
"""Get acquisition (repetition rate) from INSTRUMENT source."""
train_pulses = np.squeeze(
self.run_dc[
self.image_src, "image.pulseId"
].drop_empty_trains().train_from_index(0)[1]
self.image_src,
"image.pulseId"].drop_empty_trains().train_from_index(0)[1]
)
# Compute acquisition rate from fast data
......@@ -101,15 +96,35 @@ class AgipdCtrl:
options = {8: 0.5, 4: 1.1, 2: 2.2, 1: 4.5}
return options.get(diff, None)
def get_gain_setting(
self,
creation_time: "datetime.datetime",
) -> Optional[int]:
"""Retrieve Gain setting.
def get_acq_rate(self) -> Optional[float]:
"""Read the acquisition rate for the selected detector module.
If the data is available from the middlelayer FPGA_COMP device,
then it is retrieved from there.
If not, the setting is calculated off `setupr` and `patternTypeIndex`
The value is read from CONTROL source e.g.`CONTROL/../MDL/FPGA_COMP`,
If key is not available, the rate is calculated from
two consecutive pulses of the same trainId.
:return acq_rate: the acquisition rate.
return None, if not available.
"""
acq_rate = self._get_acq_rate_run()
if acq_rate is not None:
return acq_rate
# For AGIPD500K this function would produce wrong value (always 4.5)
# before W10/2022.
# TODO: Confirm leaving this as it is.
return self._get_acq_rate_instr()
def _get_gain_setting_run(self) -> Optional[int]:
"""Read gain_settings from RUN source and gain.value key."""
return int(self.run_dc.get_run_value(self.ctrl_src, "gain.value"))
def _get_gain_setting_run_old(self) -> Optional[int]:
"""Read gain_settings from setupr and patterTypeIndex
for old RAW data.
If `gain.value` isn't available in MDL source,
the setting is calculated from `setupr` and `patternTypeIndex`
gain-setting 1: setupr@dark=8, setupr@slopespc=40
gain-setting 0: setupr@dark=0, setupr@slopespc=32
......@@ -118,21 +133,12 @@ class AgipdCtrl:
patternTypeIndex 2: Medium-gain
patternTypeIndex 3: Low-gain
patternTypeIndex 4: SlopesPC
:return: gain setting.
return 0, if not available.
Returns:
int: gain_setting value.
"""
# TODO: remove after fixing get_possible_conditions
if creation_time and creation_time.replace(tzinfo=None) < parser.parse('2020-01-31'):
print("Set gain-setting to None for runs taken before 2020-01-31")
return
if "gain.value" in self.run_dc.keys_for_source(self.ctrl_src):
return self.run_dc[self.ctrl_src, "gain.value"].as_single_value()
setupr = self.run_dc[self.ctrl_src, "setupr.value"].as_single_value()
pattern_type_idx = self.run_dc[
self.ctrl_src, "patternTypeIndex.value"].as_single_value()
setupr = self.run_dc.get_run_value(self.ctrl_src, "setupr.value")
pattern_type_idx = self.run_dc.get_run_value(
self.ctrl_src, "patternTypeIndex.value")
if (setupr == 0 and pattern_type_idx < 4) or (
setupr == 32 and pattern_type_idx == 4):
......@@ -141,20 +147,50 @@ class AgipdCtrl:
setupr == 40 and pattern_type_idx == 4):
return 1
else:
# TODO: Confirm that this can be removed.
if self.raise_error:
raise ValueError(
"Could not derive gain setting from"
" setupr and patternTypeIndex"
)
return
def get_gain_setting(
self,
creation_time: Optional[datetime] = None,
) -> Optional[int]:
"""Read Gain setting from RUN sources.
if key `gain.value` is not available, calculate gain_setting from
setupr and patterTypeIndex. If it failed raise ValueError.
:param creation_time: datetime object for the data creation time.
:return: gain setting.
return 0, if not available.
"""
# TODO: remove after fixing get_possible_conditions
if (
creation_time and
creation_time.replace(tzinfo=None) < parser.parse('2020-01-31')
):
print("Set gain-setting to None for runs taken before 2020-01-31")
return
if "gain.value" in self.run_dc.keys_for_source(self.ctrl_src):
return self._get_gain_setting_run()
gain_setting = self._get_gain_setting_run_old()
if gain_setting is not None:
return gain_setting
else:
# TODO: confirm that this can be removed.
print(
"WARNING: gain_setting is not available "
"ERROR: gain_setting is not available "
f"at source {self.ctrl_src}.\nSet gain_setting to 0.")
# TODO: why return 0 and not None?
return 0
def get_gain_mode(self) -> AgipdGainMode:
"""Returns the gain mode (adaptive or fixed) from slow data"""
def get_gain_mode(self) -> int:
"""Returns the gain mode (adaptive or fixed) from slow data."""
if (
self.ctrl_src in self.run_dc.all_sources and
......@@ -194,7 +230,7 @@ class AgipdCtrl:
voltage_src[1] in self.run_dc.keys_for_source(voltage_src[0])
):
return self.run_dc[voltage_src].as_single_value(atol=1, reduce_by='max')
return self.run_dc.get_run_value(*voltage_src)
else:
print(
"WARNING: Unable to read bias_voltage from"
......@@ -1257,7 +1293,7 @@ class AgipdCorrections:
def initialize_from_db(self, karabo_id: str, karabo_da: str,
cal_db_interface: str,
creation_time: 'datetime.datetime',
creation_time: datetime,
memory_cells: float, bias_voltage: int,
photon_energy: float, gain_setting: float,
acquisition_rate: float, integration_time: int,
......

Legal Notice | Contacts