Error-handling sweep: crash/logic bug fixes, ERROR-token crash logging, shared helpers

pyproject.toml

@@ -162,7 +162,7 @@ isort.required-imports = ["from __future__ import annotations"]
 
 
 [tool.pylint]
-py-version = "3.9"
+py-version = "3.11"
 ignore-paths = [".*/_version.py"]
 reports.output-format = "colorized"
 similarities.ignore-imports = "yes"

src/legenddataflowscripts/par/geds/dsp/dplms.py

@@ -14,7 +14,7 @@
 from pygama.pargen.dplms_ge_dict import dplms_ge_dict
 from pygama.pargen.dsp_optimize import run_one_dsp
 
-from ....utils import build_log, convert_dict_np_to_float
+from ....utils import build_log, convert_dict_np_to_float, require_config_keys
 
 
 def par_geds_dsp_dplms() -> None:
@@ -107,6 +107,7 @@ def par_geds_dsp_dplms() -> None:
 
     dplms_dict = Props.read_from(args.config_file)
     db_dict = Props.read_from(args.database)
+    require_config_keys(dplms_dict, ["run_dplms"], f"dplms config ({args.config_file})")
 
     if dplms_dict["run_dplms"] is True:
         with Path(args.fft_raw_filelist).open() as f:

src/legenddataflowscripts/par/geds/dsp/eopt.py

@@ -19,7 +19,7 @@
     run_bayesian_optimisation,
 )
 
-from ....utils import build_log
+from ....utils import build_log, require_config_keys
 
 warnings.filterwarnings(action="ignore", category=RuntimeWarning)
 try:
@@ -107,7 +107,9 @@ def par_geds_dsp_eopt() -> None:
     argparser.add_argument(
         "--final-dsp-pars", help="final_dsp_pars", type=str, required=True
     )
-    argparser.add_argument("--qbb-grid-path", help="qbb_grid_path", type=str)
+    argparser.add_argument(
+        "--qbb-grid-path", help="qbb_grid_path", type=str, required=True
+    )
     argparser.add_argument("--plot-path", help="plot_path", type=str)
 
     argparser.add_argument(
@@ -122,8 +124,10 @@ def par_geds_dsp_eopt() -> None:
 
     opt_dict = Props.read_from(args.config_file)
     db_dict = Props.read_from(args.decay_const)
+    require_config_keys(opt_dict, ["run_eopt"], f"eopt config ({args.config_file})")
 
-    if opt_dict.pop("run_eopt") is True:
+    run_eopt = opt_dict.pop("run_eopt") is True
+    if run_eopt:
         peaks_kev = np.array(opt_dict["peaks"])
         kev_widths = [tuple(kev_width) for kev_width in opt_dict["kev_widths"]]
 
@@ -431,20 +435,22 @@ def par_geds_dsp_eopt() -> None:
         else:
             plot_dict = {}
 
-        plot_dict["trap_optimisation"] = {
-            "kernel_space": bopt_trap.plot(init_samples=sample_x),
-            "acq_space": bopt_trap.plot_acq(init_samples=sample_x),
-        }
-
-        plot_dict["cusp_optimisation"] = {
-            "kernel_space": bopt_cusp.plot(init_samples=sample_x),
-            "acq_space": bopt_cusp.plot_acq(init_samples=sample_x),
-        }
-
-        plot_dict["zac_optimisation"] = {
-            "kernel_space": bopt_zac.plot(init_samples=sample_x),
-            "acq_space": bopt_zac.plot_acq(init_samples=sample_x),
-        }
+        # the optimiser plots only exist when the optimisation actually ran
+        if run_eopt:
+            plot_dict["trap_optimisation"] = {
+                "kernel_space": bopt_trap.plot(init_samples=sample_x),
+                "acq_space": bopt_trap.plot_acq(init_samples=sample_x),
+            }
+
+            plot_dict["cusp_optimisation"] = {
+                "kernel_space": bopt_cusp.plot(init_samples=sample_x),
+                "acq_space": bopt_cusp.plot_acq(init_samples=sample_x),
+            }
+
+            plot_dict["zac_optimisation"] = {
+                "kernel_space": bopt_zac.plot(init_samples=sample_x),
+                "acq_space": bopt_zac.plot_acq(init_samples=sample_x),
+            }
 
         Path(args.plot_path).parent.mkdir(parents=True, exist_ok=True)
         with Path(args.plot_path).open("wb") as w:

src/legenddataflowscripts/par/geds/dsp/evtsel.py

@@ -17,7 +17,13 @@
 from dspeed import build_dsp
 from pygama.pargen.data_cleaning import generate_cuts, get_keys
 
-from ....utils import build_log, get_pulser_mask
+from ....utils import (
+    build_log,
+    check_pulser_mask,
+    expand_filelist,
+    get_pulser_mask,
+    require_config_keys,
+)
 
 warnings.filterwarnings(action="ignore", category=RuntimeWarning)
 
@@ -225,22 +231,20 @@ def par_geds_dsp_evtsel() -> None:
 
     peak_dict = Props.read_from(args.config_file)
     db_dict = Props.read_from(args.decay_const)
+    require_config_keys(
+        peak_dict, ["run_selection"], f"evtsel config ({args.config_file})"
+    )
 
     Path(args.peak_file).parent.mkdir(parents=True, exist_ok=True)
     if peak_dict.pop("run_selection") is True:
         log.debug("Starting peak selection")
+        require_config_keys(
+            peak_dict,
+            ["peaks", "kev_widths", "cut_parameters", "n_events", "final_cut_field"],
+            f"evtsel config ({args.config_file})",
+        )
 
-        if (
-            isinstance(args.raw_filelist, list)
-            and args.raw_filelist[0].split(".")[-1] == "filelist"
-        ):
-            files = args.raw_filelist[0]
-            with Path(files).open() as f:
-                files = f.read().splitlines()
-        else:
-            files = args.raw_filelist
-
-        raw_files = sorted(files)
+        raw_files = expand_filelist(args.raw_filelist, "--raw-filelist")
 
         peaks_kev = peak_dict["peaks"]
         kev_widths = peak_dict["kev_widths"]
@@ -264,9 +268,13 @@ def par_geds_dsp_evtsel() -> None:
         )
 
         if args.no_pulse is False:
+            if args.pulser_file is None:
+                msg = "either --pulser-file or --no-pulse is required"
+                raise ValueError(msg)
             mask = get_pulser_mask(
                 args.pulser_file,
             )
+            check_pulser_mask(mask, tb, args.raw_table_name)
         else:
             mask = np.full(len(tb), False)
 

src/legenddataflowscripts/par/geds/dsp/nopt.py

@@ -12,7 +12,7 @@
 from dspeed import build_dsp
 from pygama.pargen.data_cleaning import generate_cuts, get_cut_indexes
 
-from ....utils import build_log
+from ....utils import build_log, require_config_keys
 
 
 def par_geds_dsp_nopt() -> None:
@@ -91,6 +91,7 @@ def par_geds_dsp_nopt() -> None:
 
     opt_dict = Props.read_from(args.config_file)
     db_dict = Props.read_from(args.database)
+    require_config_keys(opt_dict, ["run_nopt"], f"nopt config ({args.config_file})")
 
     if opt_dict.pop("run_nopt") is True:
         with Path(args.raw_filelist).open() as f:
@@ -162,4 +163,4 @@ def par_geds_dsp_nopt() -> None:
             pkl.dump(plot_dict, f, protocol=pkl.HIGHEST_PROTOCOL)
 
     Path(args.dsp_pars).parent.mkdir(parents=True, exist_ok=True)
-    Props.write_to(args.dsp_pars, dict(nopt_pars=out_dict, **db_dict))
+    Props.write_to(args.dsp_pars, {**db_dict, "nopt_pars": out_dict})

src/legenddataflowscripts/par/geds/dsp/pz.py

@@ -14,8 +14,11 @@
 
 from ....utils import (
     build_log,
+    check_pulser_mask,
     convert_dict_np_to_float,
+    expand_filelist,
     get_pulser_mask,
+    require_config_keys,
 )
 
 
@@ -99,31 +102,25 @@ def par_geds_dsp_pz() -> None:
 
     log = build_log(args.log_config, args.log)
     kwarg_dict = Props.read_from(args.config_file)
+    require_config_keys(kwarg_dict, ["run_tau"], f"pz config ({args.config_file})")
 
     if kwarg_dict["run_tau"] is True:
+        require_config_keys(
+            kwarg_dict, ["threshold"], f"pz config ({args.config_file})"
+        )
         dsp_config = Props.read_from(args.processing_chain)
         kwarg_dict.pop("run_tau")
+        # prefer dedicated pz files when given; an empty pz filelist falls
+        # back to the raw files
         input_file = []
-        if args.pz_files is not None and len(args.pz_files) > 0:
-            if (
-                isinstance(args.pz_files, list)
-                and args.pz_files[0].split(".")[-1] == "filelist"
-            ):
-                input_file = args.pz_files[0]
-                with Path(input_file).open() as f:
-                    input_file = f.read().splitlines()
+        if args.pz_files:
+            if len(args.pz_files) == 1 and Path(args.pz_files[0]).suffix == ".filelist":
+                with Path(args.pz_files[0]).open() as f:
+                    input_file = [line for line in map(str.strip, f) if line]
             else:
                 input_file = args.pz_files
         if len(input_file) == 0:
-            if (
-                isinstance(args.raw_files, list)
-                and args.raw_files[0].split(".")[-1] == "filelist"
-            ):
-                input_file = args.raw_files[0]
-                with Path(input_file).open() as f:
-                    input_file = f.read().splitlines()
-            else:
-                input_file = args.raw_files
+            input_file = expand_filelist(args.raw_files, "--raw-files")
 
         msg = f"Reading Data for {args.raw_table_name} from:"
         log.debug(msg)
@@ -139,7 +136,11 @@ def par_geds_dsp_pz() -> None:
         if args.no_pulse is False and (
             args.pz_files is None or len(args.pz_files) == 0
         ):
+            if args.pulser_file is None:
+                msg = "either --pulser-file or --no-pulse is required"
+                raise ValueError(msg)
             mask = get_pulser_mask(args.pulser_file)
+            check_pulser_mask(mask, data, args.raw_table_name)
         else:
             mask = np.full(len(data), False)
 

src/legenddataflowscripts/par/geds/hit/aoe.py

@@ -17,9 +17,12 @@
 
 from ....utils import (
     build_log,
+    check_pulser_mask,
     convert_dict_np_to_float,
+    expand_filelist,
     fill_plot_dict,
     get_pulser_mask,
+    require_config_keys,
 )
 
 log = logging.getLogger(__name__)
@@ -338,6 +341,7 @@ def par_geds_hit_aoe() -> None:
 
     log = build_log(args.log_config, args.log)
     kwarg_dict = Props.read_from(args.config_file)
+    require_config_keys(kwarg_dict, ["run_aoe"], f"aoe config ({args.config_file})")
 
     ecal_dict = Props.read_from(args.ecal_file)
     cal_dict = ecal_dict["pars"]
@@ -352,10 +356,14 @@ def par_geds_hit_aoe() -> None:
     else:
         out_plot_dict = {}
 
-    with Path(args.files[0]).open() as f:
-        files = sorted(f.read().splitlines())
+    files = expand_filelist(args.files)
 
     if kwarg_dict["run_aoe"] is True:
+        require_config_keys(
+            kwarg_dict,
+            ["current_param", "energy_param", "cal_energy_param", "cut_field"],
+            f"aoe config ({args.config_file})",
+        )
         params = [
             kwarg_dict["current_param"],
             "tp_0_est",
@@ -393,17 +401,15 @@ def par_geds_hit_aoe() -> None:
         else:
             mask = np.zeros(len(threshold_mask), dtype=bool)
 
+        check_pulser_mask(mask, threshold_mask, args.table_name)
         data["is_pulser"] = mask[threshold_mask]
 
         msg = f"{len(data.query('~is_pulser'))}  non pulser events"
         log.info(msg)
 
-        with Path(args.files[0]).open() as f:
-            files = f.read().splitlines()
-        files = sorted(files)
-
         data["run_timestamp"] = args.timestamp
 
+        override_dict = None
         if args.override_files:
             override_dict = Props.read_from(args.override_files)
             override_dict = override_dict.get(args.detector, None)

src/legenddataflowscripts/par/geds/hit/ecal.py

@@ -23,8 +23,11 @@
 
 from ....utils import (
     build_log,
+    check_pulser_mask,
     convert_dict_np_to_float,
+    expand_filelist,
     get_pulser_mask,
+    require_config_keys,
 )
 
 mpl.use("agg")
@@ -196,13 +199,13 @@ def plot_pulser_timemap(
 
 
 def get_median(x):
-    if len(x[~np.isnan(x)]) >= 10:
+    if len(x[~np.isnan(x)]) < 10:
         return np.nan
     return np.nanpercentile(x, 50)
 
 
 def get_err(x):
-    if len(x[~np.isnan(x)]) >= 10:
+    if len(x[~np.isnan(x)]) < 10:
         return np.nan
     return np.nanvar(x) / np.sqrt(len(x))
 
@@ -809,6 +812,8 @@ def par_geds_hit_ecal() -> None:
 
     log = build_log(args.log_config, args.log)
 
+    hit_dict = {}
+    in_results_dict = {}
     if args.in_hit_dict:
         hit_dict = Props.read_from(args.in_hit_dict)
         in_results_dict = hit_dict.get("results", {})
@@ -828,6 +833,18 @@ def par_geds_hit_ecal() -> None:
     hit_dict.update(database_dic["ctc_params"])
 
     kwarg_dict = Props.read_from(args.config_file)
+    require_config_keys(
+        kwarg_dict,
+        [
+            "plot_options",
+            "bl_plot_options",
+            "common_plots",
+            "energy_params",
+            "cut_param",
+            "threshold",
+        ],
+        f"ecal config ({args.config_file})",
+    )
 
     # convert plot functions from strings to functions and split off baseline and common plots
     for field, item in kwarg_dict["plot_options"].items():
@@ -838,9 +855,7 @@ def par_geds_hit_ecal() -> None:
         bl_plots[field]["function"] = eval(item["function"])
     common_plots = kwarg_dict.pop("common_plots")
 
-    with Path(args.files[0]).open() as f:
-        files = f.read().splitlines()
-    files = sorted(files)
+    files = expand_filelist(args.files)
 
     # load data in
     data, threshold_mask = load_data(
@@ -865,6 +880,7 @@ def par_geds_hit_ecal() -> None:
     else:
         mask = np.zeros(len(threshold_mask), dtype=bool)
 
+    check_pulser_mask(mask, threshold_mask, args.table_name)
     data["is_pulser"] = mask[threshold_mask]
 
     pk_pars = [
@@ -920,9 +936,13 @@ def par_geds_hit_ecal() -> None:
 
         if kwarg_dict.get("guess_offset", False) is True:
             guess = np.array([0, guess])
-            guess[0] = hit_dict.get(
+            # "guess_offset_param" is a fallback key into hit_dict, not a value
+            offset_cfg = hit_dict.get(
                 "guess_offset_param", "is_valid_cuspEmax_classifier"
-            )["parameters"]["a"]
+            )
+            if isinstance(offset_cfg, str):
+                offset_cfg = hit_dict[offset_cfg]
+            guess[0] = offset_cfg["parameters"]["a"]
 
         full_object_dict[cal_energy_param] = HPGeCalibration(
             energy_param,