Switch BadPixels{,FF} base class to enum.IntFlag

Description

Casually scrolling through Python docs, I saw enum.IntFlag (available since Python 3.6) which could make all our flag-related code nicer.

The two advantages are:

• Int: the resulting enum constants we get are also subclasses of int. This means that if we for instance want to put them into numpy arrays, we can assign them directly, dropping .value.
• Flag: enables bitwise operations between enum constants resulting in enum constants. Again, this means we can drop .value (which would throw away the "semantic" value of the enum class).

I've gone over cal_tools and dropped all .value suffixes. Given that work is ongoing with notebooks, I'd suggest they can be updated later as we refactor anyway.

Side note: what to call a merge request which is just refactoring / code style stuff? This is hardly a feature or a fix.

How Has This Been Tested?

Existing code using .value is not impacted. Let's compare the two base classes:

from enum import Enum, IntFlag
import numpy as np


class OldBadPixels(Enum):
    OFFSET_OUT_OF_THRESHOLD  = 0b000000000000000000001 # bit 1
    NOISE_OUT_OF_THRESHOLD   = 0b000000000000000000010 # bit 2
	...

class NewBadPixels(IntFlag):
    OFFSET_OUT_OF_THRESHOLD  = 0b000000000000000000001 # bit 1
    NOISE_OUT_OF_THRESHOLD   = 0b000000000000000000010 # bit 2
	...

We can use the IntFlag-based ones directly as numbers:

a1 = np.zeros(1, dtype=np.uint32)
a1[0] |= NewBadPixels.NON_LIN_RESPONSE_REGION
a1[0] |= NewBadPixels.CI_LINEAR_DEVIATION

Or use .value (so existing code doing this is not affected):

a2 = np.zeros(1, dtype=np.uint32)
a2[0] |= NewBadPixels.NON_LIN_RESPONSE_REGION.value
a2[0] |= NewBadPixels.CI_LINEAR_DEVIATION.value

So far, a1 == a2. For completeness, note that the old Enum-based flags are not numbers:

b1 = np.zeros(1, dtype=np.uint32)
b1[0] |= OldBadPixels.NON_LIN_RESPONSE_REGION
b1[0] |= OldBadPixels.CI_LINEAR_DEVIATION
Traceback (most recent call last):
  File "test-440.py", line 68, in <module>
    b1[0] |= OldBadPixels.NON_LIN_RESPONSE_REGION
TypeError: unsupported operand type(s) for |: 'int' and 'OldBadPixels'

And as expected, the .value has not changed at all:

b2 = np.zeros(1, dtype=np.uint32)
b2[0] |= OldBadPixels.NON_LIN_RESPONSE_REGION.value
b2[0] |= OldBadPixels.CI_LINEAR_DEVIATION.value

So a1 == a2 == b2.

As advertised, we can do bitwise operations directly on members and get BadPixels bad instead of untypely numbers:

> print(NewBadPixels.DATA_STD_IS_ZERO | NewBadPixels.INTERPOLATED)
NewBadPixels.INTERPOLATED|DATA_STD_IS_ZERO

Bonus: we can recover a BadPixels instance:

> print(NewBadPixels(int(a1[0])))
NewBadPixels.NON_LIN_RESPONSE_REGION|CI_LINEAR_DEVIATION

And we can still use is between the enumerable things:

> print(NewBadPixels(int(a1[0])) is NewBadPixels.NON_LIN_RESPONSE_REGION | NewBadPixels.CI_LINEAR_DEVIATION)
True

Types of changes

Code improvement (non-breaking).

Checklist:

• My code follows the code style of this project.

Reviewers

@danilevc @ahmedk @roscar

cal_tools/cal_tools/agipdlib.py

@@ -5,7 +5,16 @@ from typing import Any, Dict, Optional, Tuple
 import h5py
 import numpy as np
 import sharedmem
-from iCalibrationDB import Conditions, Constants, Detectors
+from cal_tools.agipdutils import (assemble_constant_dict,
+                                  baseline_correct_via_noise,
+                                  baseline_correct_via_stripe,
+                                  correct_baseline_via_hist,
+                                  correct_baseline_via_hist_asic,
+                                  make_noisy_adc_mask, match_asic_borders,
+                                  melt_snowy_pixels)
+from cal_tools.enums import BadPixels, SnowResolution
+from cal_tools.tools import get_constant_from_db_and_time
+from iCalibrationDB import Conditions, Constants
 
 from cal_tools.agipdutils import *
 from cal_tools.cython import agipdalgs as calgs
@@ -80,7 +89,8 @@ def get_acq_rate(fast_paths: Tuple[str, str, int],
 
 
 def get_gain_setting(fname: str, h5path_ctrl: str) -> int:
-    """
+    """Retrieve Gain setting.
+
     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`
@@ -160,8 +170,8 @@ class AgipdCorrections:
             image/data section
         :param h5_index_path: path in HDF5 file which is prefixed to the
             index section
-        :param corr_bools: A dict with all of the correction booleans requested or
-                           available
+        :param corr_bools: A dict with all of the correction booleans requested
+                           or available
 
         The following example shows a typical use case:
         .. code-block:: python
@@ -203,7 +213,7 @@ class AgipdCorrections:
         self.rng_pulses = max_pulses
         # avoid list(range(*[0]]))
         self.pulses_lst = list(range(*max_pulses)) \
-            if not (len(max_pulses) == 1 and max_pulses[0] == 0) else max_pulses  #noqa
+            if not (len(max_pulses) == 1 and max_pulses[0] == 0) else max_pulses  # noqa
         self.max_cells = max_cells
 
         # Correction parameters
@@ -498,11 +508,11 @@ class AgipdCorrections:
         # force into high or medium gain if requested
         if self.corr_bools.get('force_mg_if_below'):
             gain[(gain == 2) & (
-                        (data - offsetb[1]) < self.mg_hard_threshold)] = 1
+                (data - offsetb[1]) < self.mg_hard_threshold)] = 1
 
         if self.corr_bools.get('force_hg_if_below'):
             gain[(gain > 0) & (
-                        (data - offsetb[0]) < self.hg_hard_threshold)] = 0
+                (data - offsetb[0]) < self.hg_hard_threshold)] = 0
 
         # choose constants according to gain setting
         off = calgs.gain_choose(gain, offsetb)
@@ -512,7 +522,7 @@ class AgipdCorrections:
         data -= off
         del off
 
-    def baseline_correction(self, i_proc:int, first:int, last:int):
+    def baseline_correction(self, i_proc: int, first: int, last: int):
         """
         Perform image-wise base-line shift correction for
         data in shared memory via histogram or stripe
@@ -536,14 +546,12 @@ class AgipdCorrections:
         # output is saved in sharedmem to pass for correct_agipd()
         # as this function takes about 3 seconds.
         self.shared_dict[i_proc]['msk'][first:last] = \
-                            calgs.gain_choose_int(gain,
-                                                  self.mask[module_idx][:, cellid])  # noqa
+            calgs.gain_choose_int(gain, self.mask[module_idx][:, cellid])
 
         if hasattr(self, "rel_gain"):
             # Get the correct rel_gain depending on cell-id
             self.shared_dict[i_proc]['rel_corr'][first:last] = \
-                                calgs.gain_choose(gain,
-                                                  self.rel_gain[module_idx][:, cellid])  # noqa
+                calgs.gain_choose(gain, self.rel_gain[module_idx][:, cellid])
 
         # do this image wise, as the shift is per image
         for i in range(data.shape[0]):
@@ -647,9 +655,9 @@ class AgipdCorrections:
         # after calculating it while offset correcting.
         if self.corr_bools.get('melt_snow'):
             _ = melt_snowy_pixels(self.shared_dict[i_proc]['raw_data'][first:last],  # noqa
-                                        data, gain,
-                                        self.shared_dict[i_proc]['t0_rgain'][first:last],  # noqa
-                                        self.snow_resolution)
+                                  data, gain,
+                                  self.shared_dict[i_proc]['t0_rgain'][first:last],  # noqa
+                                  self.snow_resolution)
 
         # Inner ASIC borders are matched to the same signal level
         if self.corr_bools.get("match_asics"):
@@ -712,7 +720,7 @@ class AgipdCorrections:
             valid_indices = np.concatenate([np.arange(validf[i],
                                                       validf[i]+validc[i])
                                             for i in range(validf.size)],
-                                            axis=0)
+                                           axis=0)
             valid_indices = np.squeeze(valid_indices).astype(np.int32)
 
         elif index_v == 1:
@@ -753,8 +761,8 @@ class AgipdCorrections:
         allpulses = data_dict['pulseId'][:n_img]
 
         # Initializing can_calibrate array
-        can_calibrate = self.choose_selected_pulses(allpulses,
-                                can_calibrate=[True]*len(allpulses))
+        can_calibrate = self.choose_selected_pulses(
+            allpulses, can_calibrate=[True]*len(allpulses))
 
         # Only select data corresponding to selected pulses
         # and overwrite data in shared-memory leaving
@@ -779,7 +787,7 @@ class AgipdCorrections:
         return n_img
 
     def validate_selected_pulses(self, allpulses: np.array
-                                ) -> Tuple[int, int, int]:
+                                 ) -> Tuple[int, int, int]:
         """Validate the selected pulses given from the notebook
 
         Validate that the given range of pulses to correct
@@ -816,7 +824,6 @@ class AgipdCorrections:
 
     def choose_selected_pulses(self, allpulses: np.array,
                                can_calibrate: np.array) -> np.array:
-
         """
         Choose given selected pulse from pulseId array of
         raw data. The selected pulses range is validated then
@@ -831,7 +838,7 @@ class AgipdCorrections:
         """
 
         (first_pulse, last_pulse,
-        pulse_step) = self.validate_selected_pulses(allpulses)
+         pulse_step) = self.validate_selected_pulses(allpulses)
 
         # collect the pulses to be calibrated
         cal_pulses = allpulses[first_pulse: last_pulse: pulse_step]
@@ -853,7 +860,8 @@ class AgipdCorrections:
         return can_calibrate
 
     def gen_valid_range(self, first_index: int, last_index: int,
-                        max_cells: int, allcells: np.array, allpulses: np.array,
+                        max_cells: int, allcells: np.array,
+                        allpulses: np.array,
                         valid_indices: Optional[np.array] = None,
                         apply_sel_pulses: Optional[bool] = True
                         ) -> np.array:
@@ -890,8 +898,8 @@ class AgipdCorrections:
             return
 
         if apply_sel_pulses:
-            can_calibrate = self.choose_selected_pulses(allpulses,
-                                            can_calibrate=can_calibrate)
+            can_calibrate = self.choose_selected_pulses(
+                allpulses, can_calibrate=can_calibrate)
         if valid_indices is None:
             firange = np.arange(first_index, last_index)
         else:
@@ -1075,7 +1083,7 @@ class AgipdCorrections:
 
         self.offset[module_idx][...] = cons_data["Offset"].transpose()[...]
         self.noise[module_idx][...] = cons_data["Noise"].transpose()[...]
-        self.thresholds[module_idx][...] = cons_data["ThresholdsDark"].transpose()[:3,...]  # noqa
+        self.thresholds[module_idx][...] = cons_data["ThresholdsDark"].transpose()[:3, ...]  # noqa
 
         if self.corr_bools.get("low_medium_gap"):
             t0 = self.thresholds[module_idx][0]
@@ -1090,7 +1098,7 @@ class AgipdCorrections:
                                                                       :bpixels.shape[2],  # noqa
                                                                       None]
 
-            if when["SlopesFF"]: # Checking if constant was retrieved
+            if when["SlopesFF"]:  # Checking if constant was retrieved
 
                 slopesFF = cons_data["SlopesFF"]
                 # This could be used for backward compatibility
@@ -1100,18 +1108,20 @@ class AgipdCorrections:
                 # This is for backward compatability for old FF constants
                 # (128, 512, mem_cells)
                 if slopesFF.shape[-1] == 2:
-                    xray_cor = np.squeeze(slopesFF[...,0])
+                    xray_cor = np.squeeze(slopesFF[..., 0])
                     xray_cor_med = np.nanmedian(xray_cor)
-                    xray_cor[np.isnan(xray_cor)]= xray_cor_med
-                    xray_cor[(xray_cor<0.8) | (xray_cor>1.2)] = xray_cor_med
+                    xray_cor[np.isnan(xray_cor)] = xray_cor_med
+                    xray_cor[(xray_cor < 0.8) | (
+                        xray_cor > 1.2)] = xray_cor_med
                     xray_cor = np.dstack([xray_cor]*self.max_cells)
                 else:
                     # Memory cell resolved xray_cor correction
                     xray_cor = slopesFF  # (128, 512, mem_cells)
                     if xray_cor.shape[-1] < self.max_cells:
-                        # In case of having new constant with less memory cells,
-                        # due to lack of enough FF data or during development.
-                        # xray_cor should be expanded by last memory cell.
+                        # When working with new constant with fewer memory
+                        # cells, eg. lacking enough FF data or during
+                        # development, xray_cor must be expand its last memory
+                        # cell to maintain a consistent shape.
                         xray_cor = np.dstack(xray_cor,
                                              np.dstack([xray_cor[..., -1]]
                                              * (self.max_cells - xray_cor.shape[-1])))  # noqa
@@ -1151,11 +1161,11 @@ class AgipdCorrections:
                 pc_med_l = slopesPC[..., :self.max_cells, 4]
 
                 # calculate median for slopes
-                pc_high_med = np.nanmedian(pc_high_m, axis=(0,1))
-                pc_med_med = np.nanmedian(pc_med_m, axis=(0,1))
+                pc_high_med = np.nanmedian(pc_high_m, axis=(0, 1))
+                pc_med_med = np.nanmedian(pc_med_m, axis=(0, 1))
                 # calculate median for intercepts:
-                pc_high_l_med = np.nanmedian(pc_high_l, axis=(0,1))
-                pc_med_l_med = np.nanmedian(pc_med_l, axis=(0,1))
+                pc_high_l_med = np.nanmedian(pc_high_l, axis=(0, 1))
+                pc_med_l_med = np.nanmedian(pc_med_l, axis=(0, 1))
 
                 # sanitize PC data
                 # (it should be done already on the level of constants)