diff --git a/cal_tools/cal_tools/agipdlib.py b/cal_tools/cal_tools/agipdlib.py index 266babbe99a2fecc1311b8cd1bd2072c1c66430b..18c6560bc06f74743c41557c1a36b2067baa2054 100644 --- a/cal_tools/cal_tools/agipdlib.py +++ b/cal_tools/cal_tools/agipdlib.py @@ -70,9 +70,11 @@ def get_acq_rate(slow_paths: Tuple[str, str], return options.get(diff, None) -def get_gain_setting(fname, h5path_ctrl): +def get_gain_setting(fname: str, h5path_ctrl: str) -> int: """ - Return gain setting base on setupr and patternTypeIndex + 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` gain-setting 1: setupr@dark=8, setupr@slopespc=40 gain-setting 0: setupr@dark=0, setupr@slopespc=32 @@ -86,11 +88,18 @@ def get_gain_setting(fname, h5path_ctrl): :param h5path_ctrl: path to control information inside the file :return: gain setting """ - with h5py.File(fname, "r") as f: - train_id = f["INDEX/trainId"][()] + gain_path = f'{h5path_ctrl}/gain/value' + with h5py.File(fname, "r") as fin: + if gain_path in fin: + return fin[gain_path][0] + + # Get the index at which the train is not zero. + train_id = fin["INDEX/trainId"][()] idx = np.nonzero(train_id)[0][0] - setupr = f[f'{h5path_ctrl}/setupr/value'][idx] - pattern_type_idx = f[f'{h5path_ctrl}/patternTypeIndex/value'][idx] + + setupr = fin[f'{h5path_ctrl}/setupr/value'][idx] + pattern_type_idx = fin[f'{h5path_ctrl}/patternTypeIndex/value'][idx] + if (setupr == 0 and pattern_type_idx < 4) or ( setupr == 32 and pattern_type_idx == 4): return 0 diff --git a/cal_tools/cal_tools/tools.py b/cal_tools/cal_tools/tools.py index d6364ba6b3ef82e32ab32400d53f23f170b7322f..dcb1ea0598965cb94f5d6601e6eaa8e3dcbe247e 100644 --- a/cal_tools/cal_tools/tools.py +++ b/cal_tools/cal_tools/tools.py @@ -8,6 +8,7 @@ from queue import Queue import re import textwrap from time import sleep +from typing import Optional from urllib.parse import urljoin import dateutil.parser @@ -228,7 +229,9 @@ def get_run_info(proposal, run): return resp.json() -def get_dir_creation_date(directory, run, tsdir=False, verbosity=0): +def get_dir_creation_date(directory: str, run: int, + tsdir: Optional[bool] = False, + verbosity: Optional[int] = 0): """ Return run starting time from the MDC. If not succeeded, return modification time of oldest file.h5