diff --git a/src/cal_tools/agipdlib.py b/src/cal_tools/agipdlib.py
index 357486fdfbe12883f6a934f92dab8e384457378a..ec6b4e25c7fe896a6e5b7fa2b99408ccb12ae627 100644
--- a/src/cal_tools/agipdlib.py
+++ b/src/cal_tools/agipdlib.py
@@ -9,6 +9,7 @@ from typing import Any, Dict, List, Optional, Tuple
import h5py
import numpy as np
import sharedmem
+from extra_data import DataCollection
from iCalibrationDB import Conditions, Constants
from cal_tools import agipdalgs as calgs
@@ -26,8 +27,6 @@ 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
-from extra_data import DataCollection
-
def get_num_cells(fname, loc, module):
with h5py.File(fname, "r") as f:
@@ -196,7 +195,9 @@ class CellSelection:
CM_PRESEL = 1
CM_FINSEL = 2
- def get_cells_on_trains(self, trains_sel: List[int], cm: int = 0) -> np.array:
+ def get_cells_on_trains(
+ self, trains_sel: List[int], cm: int = 0
+ ) -> np.array:
"""Returns mask of cells selected for processing
:param train_sel: list of a train ids selected for processing
@@ -538,6 +539,8 @@ class AgipdCorrections:
fraction = self.cm_dark_fraction
n_itr = self.cm_n_itr
n_img = self.shared_dict[i_proc]['nImg'][0]
+ if n_img == 0:
+ return
cell_id = self.shared_dict[i_proc]['cellId'][:n_img]
train_id = self.shared_dict[i_proc]['trainId'][:n_img]
cell_ids = cell_id[train_id == train_id[0]]
@@ -954,7 +957,7 @@ class AgipdCorrections:
:param allcells: array of image.cellsIds of raw data
:param allpulses: array of image.pulseIds of raw data
:param valid_indices: validated indices of image.data
- :param img_selected: mask of selected cells for given
+ :param img_selected: mask of selected cells for given
range of trains
:return firange: An array of validated image.data
indices to correct
@@ -969,12 +972,12 @@ class AgipdCorrections:
can_calibrate = (allcells < max_cells)
- if not np.any(can_calibrate):
- return
-
if img_selected is not None:
can_calibrate &= img_selected
+ if not np.any(can_calibrate):
+ return
+
if valid_indices is None:
firange = np.arange(first_index, last_index)
else:
@@ -1452,6 +1455,70 @@ class AgipdCorrections:
self.shared_dict[i]["valid_trains"] = sharedmem.empty(1024, dtype="u8") # noqa
+def validate_selected_pulses(
+ max_pulses: List[int], max_cells: int
+) -> List[int]:
+ """Validate the selected pulses given from the notebook
+
+ Validate the selected range of pulses to correct raw data
+ of at least one image.
+
+ 1) A pulse indices within one train can't be greater
+ than the operating memory cells.
+ 2) Validate the order of the given raneg of pulses.
+ 3) Raise value error if generate list of pulses is empty.
+
+ :param max_pulses: a list of at most 3 elements defining the
+ range of pulses to calibrate.
+ :param max_cells: operating memory cells.
+
+ :return adjusted_range: An adjusted range of pulse indices to correct.
+ """
+
+ # Validate selected pulses range:
+ # 1) A pulseId can't be greater than the operating memory cells.
+ pulses_range = [max_cells if p > max_cells else p for p in max_pulses]
+
+ if pulses_range != max_pulses:
+ print(
+ "WARNING: \"max_pulses\" list has been modified from "
+ f"{max_pulses} to {pulses_range}. As the number of "
+ "operating memory cells are less than the selected "
+ "maximum pulse."
+ )
+
+ if len(pulses_range) == 1:
+ adjusted_range = (0, pulses_range[0], 1)
+ elif len(pulses_range) == 2:
+ adjusted_range = (pulses_range[0], pulses_range[1], 1)
+ elif len(pulses_range) == 3:
+ adjusted_range = tuple(pulses_range)
+ else:
+ raise ValueError(
+ "ERROR: Wrong length for the list of pulses indices range. "
+ "Please check the given range for \"max_pulses\":"
+ f"{max_pulses}. \"max_pulses\" needs to be a list of "
+ "3 elements, [start, last, step]")
+
+ if adjusted_range[0] > adjusted_range[1]:
+ raise ValueError(
+ "ERROR: Pulse range start is greater than range end. "
+ "Please check the given range for \"max_pulses\":"
+ f"{max_pulses}. \"max_pulses\" needs to be a list of "
+ "3 elements, [start, last, step]")
+
+ if not np.all([isinstance(p, int) for p in max_pulses]):
+ raise TypeError(
+ "ERROR: \"max_pulses\" elements needs to be integers:"
+ f" {max_pulses}.")
+
+ print(
+ "A range of pulse indices is selected to correct:"
+ f" {pulses_range}")
+
+ return adjusted_range
+
+
class CellRange(CellSelection):
"""Selection of detector memory cells given as a range"""
@@ -1463,7 +1530,7 @@ class CellRange(CellSelection):
:param max_cells: number of exposed cells
"""
self.max_cells = max_cells
- self.crange = crange
+ self.crange = validate_selected_pulses(crange, max_cells)
self.flag = np.zeros(self.max_cells, bool)
self.flag_cm = np.zeros(self.ncell_max, bool)
self.flag[slice(*crange)] = True
@@ -1478,7 +1545,9 @@ class CellRange(CellSelection):
f"Frames per train: {self.flag.sum()}"
)
- def get_cells_on_trains(self, train_sel: List[int], cm: int = 0) -> np.array:
+ def get_cells_on_trains(
+ self, train_sel: List[int], cm: int = 0
+ ) -> np.array:
return np.tile(self._sel_for_cm(self.flag, self.flag_cm, cm),
len(train_sel))
@@ -1500,7 +1569,7 @@ class LitFrameSelection(CellSelection):
"""
# read AgipdLitFrameFinder data
self.dev = dev
- self.crange = crange
+ self.crange = validate_selected_pulses(crange, self.ncell_max)
self.ethr = energy_threshold
intr_src = dev + ':output'
nfrm = dc[intr_src, 'data.nFrame'].ndarray()
@@ -1568,7 +1637,9 @@ class LitFrameSelection(CellSelection):
f"Frames per train: {srng}"
)
- def get_cells_on_trains(self, train_sel: List[int], cm: int = 0) -> np.array:
+ def get_cells_on_trains(
+ self, train_sel: List[int], cm: int = 0
+ ) -> np.array:
train_idx = np.flatnonzero(np.in1d(self.train_ids, train_sel))
i0 = self.first[train_idx]
iN = i0 + self.count[train_idx]