diff --git a/notebooks/AGIPD/AGIPD_Correct_and_Verify.ipynb b/notebooks/AGIPD/AGIPD_Correct_and_Verify.ipynb
index b47dea5b89b8b740f3567980c56d4d139e5d0bc1..613b264ad2f2fe0541ac49b543c5f2f33b5e80a3 100644
--- a/notebooks/AGIPD/AGIPD_Correct_and_Verify.ipynb
+++ b/notebooks/AGIPD/AGIPD_Correct_and_Verify.ipynb
@@ -46,7 +46,7 @@
"gain_setting = 0.1 # the gain setting, use 0.1 to try to auto-determine\n",
"photon_energy = 9.2 # photon energy in keV\n",
"overwrite = True # set to True if existing data should be overwritten\n",
- "max_pulses = [0, 352, 1] # range list [st, end, step] of maximum pulse indices within a train. 3 allowed maximum list input elements.\n",
+ "max_pulses = [0, 352, 1] # range list [st, end, step] of memory cell indices to be processed within a train. 3 allowed maximum list input elements.\n",
"mem_cells_db = 0 # set to a value different than 0 to use this value for DB queries\n",
"cell_id_preview = 1 # cell Id used for preview in single-shot plots\n",
"integration_time = -1 # integration time, negative values for auto-detection.\n",
@@ -270,26 +270,6 @@
"print(f\"Detector instance {dinstance}\")"
]
},
- {
- "cell_type": "code",
- "execution_count": null,
- "metadata": {},
- "outputs": [],
- "source": [
- "# Display Information about the selected pulses indices for correction.\n",
- "pulses_lst = list(range(*max_pulses)) if not (len(max_pulses)==1 and max_pulses[0]==0) else max_pulses\n",
- "\n",
- "try:\n",
- " if len(pulses_lst) > 1:\n",
- " print(\"A range of {} pulse indices is selected: from {} to {} with a step of {}\"\n",
- " .format(len(pulses_lst), pulses_lst[0] , pulses_lst[-1] + (pulses_lst[1] - pulses_lst[0]),\n",
- " pulses_lst[1] - pulses_lst[0]))\n",
- " else:\n",
- " print(f\"one pulse is selected: a pulse of idx {pulses_lst[0]}\")\n",
- "except Exception as e:\n",
- " raise ValueError(f\"max_pulses input Error: {e}\")"
- ]
- },
{
"cell_type": "code",
"execution_count": null,
diff --git a/src/cal_tools/agipdlib.py b/src/cal_tools/agipdlib.py
index 2881c9589828ae8b1c6dbebeb4eea3b457bb3250..121c120c81d12c171c6c0be2679cc1a7ade8415f 100644
--- a/src/cal_tools/agipdlib.py
+++ b/src/cal_tools/agipdlib.py
@@ -3,7 +3,7 @@ import traceback
import zlib
from multiprocessing.pool import ThreadPool
from pathlib import Path
-from typing import Any, Dict, Optional, Tuple
+from typing import Any, Dict, List, Optional, Tuple
import h5py
import numpy as np
@@ -247,18 +247,14 @@ class AgipdCorrections:
# Data description
self.h5_data_path = h5_data_path
self.h5_index_path = h5_index_path
- self.rng_pulses = max_pulses
- # avoid list(range(*[0]]))
- self.pulses_lst = (
- list(range(*max_pulses))
- if max_pulses != [0]
- else max_pulses
- )
self.max_cells = max_cells
self.gain_mode = gain_mode
self.comp_threads = comp_threads
+ self.start, self.last, self.step = self._validate_selected_pulses(
+ max_pulses, max_cells)
+
# Correction parameters
self.baseline_corr_noise_threshold = -1000
self.snow_resolution = SnowResolution.INTERPOLATE
@@ -835,7 +831,8 @@ class AgipdCorrections:
# Initializing can_calibrate array
can_calibrate = self.choose_selected_pulses(
- allpulses, can_calibrate=[True]*len(allpulses))
+ allpulses,
+ can_calibrate=np.ones(shape=(len(allpulses),), dtype=np.bool))
# Only select data corresponding to selected pulses
# and overwrite data in shared-memory leaving
@@ -859,37 +856,69 @@ class AgipdCorrections:
return n_img
- def validate_selected_pulses(self, allpulses: np.array
- ) -> Tuple[int, int, int]:
+ def _validate_selected_pulses(
+ self, max_pulses: List[int],
+ max_cells: int,
+ ) -> List[int]:
"""Validate the selected pulses given from the notebook
- Validate that the given range of pulses to correct
- are within the cells of raw data.
+ 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 allpulses:
- :return :
- - first_pulse: first pulse index
- - last_pulse: last pulse index
- - pulse_step: step range for the selected pulse indices
+ :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.
"""
- # Calculate the pulse step from the chosen max_pulse range
- pulse_step = self.rng_pulses[2] if len(self.rng_pulses) == 3 else 1
# Validate selected pulses range:
- # 1) Make sure the range max doesn't have non-valid idx.
- if self.pulses_lst[-1] + pulse_step > int(allpulses[-1]):
- last_pulse = int(allpulses[-1]) + pulse_step
- else:
- last_pulse = int(self.pulses_lst[-1]) + pulse_step
+ # 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."
+ )
- # 2) Check if 1st pulse index was out of valid range.
- # If that is the case only calibrate the last step pulse.
- if self.pulses_lst[0] >= last_pulse:
- first_pulse = last_pulse - pulse_step
+ 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:
- first_pulse = self.pulses_lst[0]
-
- return first_pulse, last_pulse, pulse_step
+ 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
def choose_selected_pulses(self, allpulses: np.array,
can_calibrate: np.array) -> np.array:
@@ -906,26 +935,13 @@ class AgipdCorrections:
selected pulses
"""
- (first_pulse, last_pulse,
- pulse_step) = self.validate_selected_pulses(allpulses)
-
- # collect the pulses to be calibrated
- cal_pulses = allpulses[first_pulse: last_pulse: pulse_step]
-
- # Check if a specific pulses need to be calibrated
- # or with only simple pulse ranging.
- if pulse_step > 1 or \
- allpulses[first_pulse] >= allpulses[last_pulse]:
- can_calibrate = np.logical_and(can_calibrate,
- np.isin(allpulses, cal_pulses))
- else:
- # Check interesection between array of booleans and
- # array of pulses to calibrate.
- can_calibrate = np.logical_and(can_calibrate,
- (allpulses <= np.max(cal_pulses)),
- (allpulses >= np.min(cal_pulses))
- )
-
+ # Check interesection between array of booleans and
+ # array of pulses to calibrate.
+ can_calibrate &= (
+ (allpulses >= allpulses[self.start]) &
+ (allpulses <= allpulses[self.last-1]) &
+ (((allpulses - allpulses[self.start]) % allpulses[self.step]) == 0) # noqa
+ )
return can_calibrate
def gen_valid_range(self, first_index: int, last_index: int,