Merge pull request #110 from legend-exp/dev

Some improvements

Author: gipert

workflow/rules/filelist_gen.smk

@@ -163,8 +163,10 @@ def build_filelist(
         ignore_keys = []
     if analysis_runs is None:
         analysis_runs = {}
+
     phy_filenames = []
     other_filenames = []
+
     for key in filekeys:
         if Path(search_pattern).suffix == ".*":
             search_pattern = Path(search_pattern).with_suffix(".{ext}")
@@ -198,19 +200,15 @@ def build_filelist(
                     else:
                         other_filenames += filename
                 else:
+                    # check if period in analysis_runs dicts
+                    # check if run in analysis_runs dicts
+                    # or if runs is just specified as "all"
                     if (
                         _key.datatype in analysis_runs
-                        and _key.period
-                        in analysis_runs[
-                            _key.datatype
-                        ]  # check if period in analysis_runs dicts
+                        and _key.period in analysis_runs[_key.datatype]
                         and (
-                            _key.run
-                            in analysis_runs[_key.period][
-                                _key.datatype
-                            ]  # check if run in analysis_runs dicts
-                            or analysis_runs[_key.period][_key.datatype]
-                            == "all"  # or if runs is just specified as "all"
+                            _key.run in analysis_runs[_key.datatype][_key.period]
+                            or analysis_runs[_key.datatype][_key.period] == "all"
                         )
                     ):
                         if _key.datatype in concat_datatypes:

workflow/src/legenddataflow/log.py

@@ -5,42 +5,36 @@
 from dbetto import Props
 
 
-def build_log(config_dict: dict, log_file: str | None = None) -> logging.Logger:
-    """
-    Build a logger from a configuration dictionary.
+def build_log(
+    config_dict: dict, log_file: str | None = None, fallback: str = "prod"
+) -> logging.Logger:
+    """Build a logger from a configuration dictionary.
+
     If a log file is provided, the logger will write to that file.
 
     Parameters
     ----------
-
-    config_dict : dict
+    config_dict
         A dictionary containing the logging configuration.
-    log_file : str, optional
+    log_file
         The path to the log file.
-
-    Returns
-    -------
-    Logger
-        The logger object.
-
     """
     if "logging" in config_dict["options"]:
         log_config = config_dict["options"]["logging"]
         log_config = Props.read_from(log_config)
+
         if log_file is not None:
             Path(log_file).parent.mkdir(parents=True, exist_ok=True)
-            log_config["handlers"]["dynamic"] = {
-                "class": "logging.FileHandler",
-                "level": "DEBUG",
-                "formatter": "simple",
-                "filename": log_file,
-                "mode": "a",
-            }
+            log_config["handlers"]["dataflow"]["filename"] = log_file
+
         dictConfig(log_config)
         log = logging.getLogger(config_dict["options"].get("logger", "prod"))
+
     else:
         if log_file is not None:
             Path(log_file).parent.mkdir(parents=True, exist_ok=True)
             logging.basicConfig(level=logging.INFO, filename=log_file, filemode="w")
-        log = logging.getLogger(__name__)
+
+        log = logging.getLogger(fallback)
+
     return log

workflow/src/legenddataflow/scripts/tier/dsp.py

@@ -1,6 +1,5 @@
 import argparse
 import re
-import time
 from pathlib import Path
 
 import numpy as np
@@ -57,7 +56,7 @@ def build_tier_dsp() -> None:
         msg = f"tier {args.tier} not supported"
         raise ValueError(msg)
 
-    log = build_log(config_dict, args.log)
+    log = build_log(config_dict, args.log, fallback=__name__)
 
     settings_dict = config_dict.options.get("settings", {})
     if isinstance(settings_dict, str):
@@ -77,14 +76,21 @@ def build_tier_dsp() -> None:
         }
 
     # now construct the dictionary of DSP configs for build_dsp()
-    channel_dict = {}
+    dsp_cfg_tbl_dict = {}
     for chan, file in chan_cfg_map.items():
         if chan_map[chan].analysis.processable is False:
             msg = f"channel {chan} is set to non-processable in the channel map"
             raise RuntimeError(msg)
 
         tbl = "dsp" if args.tier in ["ann", "pan"] else "raw"
-        channel_dict[f"ch{chan_map[chan].daq.rawid:07}/{tbl}"] = Props.read_from(file)
+        input_tbl_name = f"ch{chan_map[chan].daq.rawid:07}/{tbl}"
+
+        # check if the raw tables are all existing
+        if len(lh5.ls(args.input, input_tbl_name)) > 0:
+            dsp_cfg_tbl_dict[input_tbl_name] = Props.read_from(file)
+        else:
+            msg = f"table {input_tbl_name} not found in {args.input} skipping"
+            log.warning(msg)
 
     # par files
     db_files = [
@@ -93,36 +99,27 @@ def build_tier_dsp() -> None:
         if Path(par_file).suffix in (".json", ".yaml", ".yml")
     ]
 
-    database_dic = _replace_list_with_array(
+    database_dict = _replace_list_with_array(
         Props.read_from(db_files, subst_pathvar=True)
     )
-    database_dic = {
+    database_dict = {
         (f"ch{chan_map[chan].daq.rawid:07}" if chan in chan_map else chan): dic
-        for chan, dic in database_dic.items()
+        for chan, dic in database_dict.items()
     }
 
     Path(args.output).parent.mkdir(parents=True, exist_ok=True)
 
-    rng = np.random.default_rng()
-    rand_num = f"{rng.integers(0, 99999):05d}"
-    temp_output = f"{args.output}.{rand_num}"
-
-    start = time.time()
-
     build_dsp(
         args.input,
-        temp_output,
+        args.output,
         {},
-        database=database_dic,
-        chan_config=channel_dict,
+        database=database_dict,
+        chan_config=dsp_cfg_tbl_dict,
         write_mode="r",
         buffer_len=settings_dict.get("buffer_len", 1000),
         block_width=settings_dict.get("block_width", 16),
     )
 
-    log.info(f"build_dsp finished in {time.time()-start}")
-    Path(temp_output).rename(args.output)
-
     key = Path(args.output).name.replace(f"-tier_{args.tier}.lh5", "")
 
     if args.tier in ["dsp", "psp"]:
@@ -136,7 +133,7 @@ def build_tier_dsp() -> None:
 
         outputs = {}
         channels = []
-        for channel, chan_dict in channel_dict.items():
+        for channel, chan_dict in dsp_cfg_tbl_dict.items():
             output = chan_dict["outputs"]
             in_dict = False
             for entry in outputs:
@@ -162,7 +159,7 @@ def build_tier_dsp() -> None:
     else:
         outputs = {}
         channels = []
-        for channel, chan_dict in channel_dict.items():
+        for channel, chan_dict in dsp_cfg_tbl_dict.items():
             output = chan_dict["outputs"]
             in_dict = False
             for entry in outputs:

workflow/src/legenddataflow/scripts/tier/hit.py

@@ -3,6 +3,7 @@
 
 from dbetto.catalog import Props
 from legendmeta import LegendMetadata, TextDB
+from lgdo import lh5
 from pygama.hit.build_hit import build_hit
 
 from ...log import build_log
@@ -34,7 +35,7 @@ def build_tier_hit() -> None:
         .on(args.timestamp, system=args.datatype)
         .snakemake_rules.tier_hit
     )
-    log = build_log(df_config, args.log)
+    log = build_log(df_config, args.log, fallback=__name__)
     log.info("initializing")
 
     settings_dict = df_config.options.get("settings", {})
@@ -71,7 +72,14 @@ def build_tier_hit() -> None:
         # get pars (to override hit config)
         Props.add_to(hit_cfg, pars_dict.get(chan, {}).copy())
 
-        channel_dict[f"ch{chan_map[chan].daq.rawid}/dsp"] = hit_cfg
+        input_tbl_name = f"ch{chan_map[chan].daq.rawid}/dsp"
+
+        # check if the raw tables are all existing
+        if len(lh5.ls(args.input, input_tbl_name)) > 0:
+            channel_dict[input_tbl_name] = hit_cfg
+        else:
+            msg = f"table {input_tbl_name} not found in {args.input} skipping"
+            log.warning(msg)
 
     log.info("running build_hit()...")
     Path(args.output).parent.mkdir(parents=True, exist_ok=True)