single job for spms

Author: ggmarshall

pyproject.toml

@@ -118,6 +118,7 @@ par-geds-raw-blindcal   = "legenddataflow.scripts.par.geds.raw.blindcal:par_geds
 par-geds-raw-blindcheck = "legenddataflow.scripts.par.geds.raw.blindcheck:par_geds_raw_blindcheck"
 par-geds-tcm-pulser     = "legenddataflow.scripts.par.geds.tcm.pulser:par_geds_tcm_pulser"
 par-spms-dsp-trg-thr    = "legenddataflow.scripts.par.spms.dsp.trigger_threshold:par_spms_dsp_trg_thr"
+par-spms-dsp-trg-thr-multi    = "legenddataflow.scripts.par.spms.dsp.trigger_threshold:par_spms_dsp_trg_thr_multi"
 
 [tool.uv.workspace]
 exclude = ["generated", "inputs", "software", "workflow"]

workflow/rules/chanlist_gen.smk

@@ -13,13 +13,10 @@ from legenddataflow import execenv_pyexe
 from legenddataflow.utils import filelist_path
 
 
-# FIXME: the system argument should always be explicitly supplied
-def get_chanlist(
-    setup, keypart, workflow, config, det_status, chan_maps, system="geds"
-):
+def get_chanlist(config, keypart, workflow, det_status, chan_maps, system):
     key = ChannelProcKey.parse_keypart(keypart)
 
-    flist_path = filelist_path(setup)
+    flist_path = filelist_path(config)
     os.makedirs(flist_path, exist_ok=True)
     output_file = os.path.join(
         flist_path,
@@ -45,15 +42,13 @@ def get_par_chanlist(
     basedir,
     det_status,
     chan_maps,
+    system,
     datatype="cal",
-    system="geds",
     name=None,
     extension="yaml",
 ):
 
-    chan_list = get_chanlist(
-        setup, keypart, workflow, config, det_status, chan_maps, system
-    )
+    chan_list = get_chanlist(setup, keypart, workflow, det_status, chan_maps, system)
 
     par_pattern = get_pattern_pars_tmp_channel(
         setup, tier, name, datatype=datatype, extension=extension
@@ -64,9 +59,18 @@ def get_par_chanlist(
     return filenames
 
 
-def get_plt_chanlist(setup, keypart, tier, basedir, det_status, chan_maps, name=None):
+def get_plt_chanlist(
+    setup,
+    keypart,
+    tier,
+    basedir,
+    det_status,
+    chan_maps,
+    system,
+    name=None,
+):
 
-    chan_list = get_chanlist(setup, keypart, workflow, config, det_status, chan_maps)
+    chan_list = get_chanlist(setup, keypart, workflow, det_status, chan_maps, system)
 
     par_pattern = get_pattern_plts_tmp_channel(setup, tier, name)
 

workflow/rules/channel_merge.smk

@@ -5,7 +5,7 @@ from legenddataflow.utils import set_last_rule_name
 from legenddataflow.execenv import execenv_pyexe
 
 
-def build_merge_rules(tier, lh5_merge=False, lh5_tier=None):
+def build_merge_rules(tier, lh5_merge=False, lh5_tier=None, system="geds"):
     if lh5_tier is None:
         lh5_tier = tier
     rule:
@@ -17,6 +17,7 @@ def build_merge_rules(tier, lh5_merge=False, lh5_tier=None):
                 basedir,
                 det_status,
                 chan_maps,
+                system=system,
             ),
         output:
             patterns.get_pattern_plts(config, tier),
@@ -38,6 +39,7 @@ def build_merge_rules(tier, lh5_merge=False, lh5_tier=None):
                 basedir,
                 det_status,
                 chan_maps,
+                system=system,
                 name="objects",
                 extension="pkl",
             ),
@@ -68,6 +70,7 @@ def build_merge_rules(tier, lh5_merge=False, lh5_tier=None):
                     basedir,
                     det_status,
                     chan_maps,
+                    system=system,
                 ),
             output:
                 temp(
@@ -86,34 +89,6 @@ def build_merge_rules(tier, lh5_merge=False, lh5_tier=None):
 
         set_last_rule_name(workflow, f"build_pars_{tier}_db")
 
-        rule:
-            """Merge pars for SiPM channels in a single pars file."""
-            input:
-                lambda wildcards: get_par_chanlist(
-                    config,
-                    f"all-{wildcards.experiment}-{wildcards.period}-{wildcards.run}-{wildcards.datatype}-{wildcards.timestamp}-channels",
-                    tier,
-                    basedir,
-                    det_status,
-                    chan_maps,
-                    datatype=wildcards.datatype,
-                    system="spms"
-                ),
-            output:
-                patterns.get_pattern_pars(
-                    config,
-                    tier,
-                    name="spms",
-                    datatype="{datatype}",
-                ),
-            group:
-                f"merge-{tier}"
-            shell:
-                execenv_pyexe(config, "merge-channels") + \
-                "--input {input} "
-                "--output {output} "
-
-        set_last_rule_name(workflow, f"build_pars_spms_{tier}_db")
 
     rule:
         input:
@@ -124,6 +99,7 @@ def build_merge_rules(tier, lh5_merge=False, lh5_tier=None):
                 basedir,
                 det_status,
                 chan_maps,
+                system=system,
                 extension="lh5" if lh5_merge is True else inspect.signature(get_par_chanlist).parameters['extension'].default,
             ),
             in_db=patterns.get_pattern_pars_tmp(

workflow/rules/dsp_pars_spms.smk

@@ -13,30 +13,47 @@ rule build_pars_dsp_tau_spms:
     params:
         timestamp="{timestamp}",
         datatype="{datatype}",
-        channel="{channel}",
-        raw_table_name=lambda wildcards: get_table_name(
-            metadata,
+        channels=lambda wildcards: get_chanlist(
             config,
-            wildcards.datatype,
-            wildcards.timestamp,
-            wildcards.channel,
-            "raw",
+            f"all-{wildcards.experiment}-{wildcards.period}-{wildcards.run}-{wildcards.datatype}-{wildcards.timestamp}-channels",
+            workflow,
+            det_status,
+            chan_maps,
+            system="spms",
         ),
+        raw_table_names=lambda wildcards: [
+            get_table_name(
+                metadata,
+                config,
+                wildcards.datatype,
+                wildcards.timestamp,
+                channel,
+                "raw",
+            )
+            for channel in get_chanlist(
+                config,
+                f"all-{wildcards.experiment}-{wildcards.period}-{wildcards.run}-{wildcards.datatype}-{wildcards.timestamp}-channels",
+                workflow,
+                det_status,
+                chan_maps,
+                system="spms",
+            )
+        ],
     wildcard_constraints:
         datatype=r"\b(?!cal\b|xtc\b)\w+\b",
     output:
-        temp(patt.get_pattern_pars_tmp_channel(config, "dsp", datatype="{datatype}")),
+        patt.get_pattern_pars(config, "dsp", name="spms", datatype="{datatype}"),
     log:
-        patt.get_pattern_log_channel(config, "pars_spms", time, datatype="{datatype}"),
+        patt.get_pattern_log(config, "pars_spms", time, datatype="{datatype}"),
     group:
         "par-dsp"
     shell:
-        execenv_pyexe(config, "par-spms-dsp-trg-thr") + "--config-path {configs} "
+        execenv_pyexe(config, "par-spms-dsp-trg-thr-multi") + "--config-path {configs} "
         "--raw-file {input.raw_file} "
         "--dsp-db {input.pardb} "
         "--datatype {params.datatype} "
         "--timestamp {params.timestamp} "
-        "--sipm-name {params.channel} "
-        "--raw-table-name {params.raw_table_name} "
+        "--sipm-names {params.channels} "
+        "--raw-table-names {params.raw_table_names} "
         "--output-file {output} "
         "--logfile {log} "

workflow/src/legenddataflow/scripts/par/spms/dsp/trigger_threshold.py

@@ -11,71 +11,48 @@
 from .....log import build_log
 
 
-def par_spms_dsp_trg_thr() -> None:
-    # CLI interface
-    argparser = argparse.ArgumentParser()
-    argparser.add_argument("--raw-file", required=True)
-    argparser.add_argument("--raw-table-name", required=True)
-    argparser.add_argument("--output-file", required=True)
-    argparser.add_argument("--config-path", required=True)
-    argparser.add_argument("--datatype", required=True)
-    argparser.add_argument("--timestamp", required=True)
-    argparser.add_argument("--sipm-name", required=True)
-    argparser.add_argument("--dsp-db", nargs="*", default=[])
-    argparser.add_argument("--logfile")
-    args = argparser.parse_args()
-
-    # dataflow configs
-    df_configs = (
-        TextDB(args.config_path, lazy=True)
-        .on(args.timestamp, system=args.datatype)
-        .snakemake_rules.pars_spms_dsp_trg_thr
-    )
-
-    # setup logging
-    log = build_log(df_configs, args.logfile)
-
+def get_channel_trg_thr(df_configs, sipm_name, dsp_db, raw_file, raw_table_name, log):
     log.debug("reading in the configuration files")
     config = df_configs.inputs
     dsp_config = utils.load_dict(
-        cfgtools.get_channel_config(config.processing_chain, args.sipm_name)
+        cfgtools.get_channel_config(config.processing_chain, sipm_name)
     )
     settings = AttrsDict(
-        utils.load_dict(cfgtools.get_channel_config(config.settings, args.sipm_name))
+        utils.load_dict(cfgtools.get_channel_config(config.settings, sipm_name))
     )
 
     # get DSP database from overrides
-    _db_dict = Props.read_from(args.dsp_db).get(args.sipm_name, {})
+    _db_dict = Props.read_from(dsp_db).get(sipm_name, {})
 
     fwhm = None
 
     # read raw file list
     log.debug("reading in the raw waveforms")
-    if len(lh5.ls(args.raw_file, f"{args.raw_table_name}/waveform_bit_drop")) == 0:
+    if len(lh5.ls(raw_file, f"{raw_table_name}/waveform_bit_drop")) == 0:
         msg = (
-            f"could not find waveform '{args.raw_table_name}/waveform_bit_drop' "
+            f"could not find waveform '{raw_table_name}/waveform_bit_drop' "
             "in {args.raw_file}, returning null pars"
         )
         log.warning(msg)
 
     else:
         data = lh5.read(
-            args.raw_table_name,
-            args.raw_file,
+            raw_table_name,
+            raw_file,
             field_mask=["waveform_bit_drop"],
             n_rows=settings.n_events,
         )
 
         if len(data) == 0:
             msg = (
-                f"could not find any waveforms '{args.raw_table_name}/waveform_bit_drop' "
+                f"could not find any waveforms '{raw_table_name}/waveform_bit_drop' "
                 "in {args.raw_file}, returning null pars"
             )
             log.warning(msg)
 
         elif len(data) < settings.n_events:
             msg = (
-                f"number of waveforms '{args.raw_table_name}/waveform_bit_drop' < {settings.n_events}"
+                f"number of waveforms '{raw_table_name}/waveform_bit_drop' < {settings.n_events}"
                 "in {args.raw_file}, can't build histogram"
             )
             raise RuntimeError(msg)
@@ -112,9 +89,92 @@ def par_spms_dsp_trg_thr() -> None:
                 msg = f"determined FWHM of baseline derivative distribution is so <= 0: {fwhm:.3f}"
                 raise RuntimeError(msg)
 
+            return fwhm
+    return None
+
+
+def par_spms_dsp_trg_thr() -> None:
+    # CLI interface
+    argparser = argparse.ArgumentParser()
+    argparser.add_argument("--raw-file", required=True)
+    argparser.add_argument("--raw-table-name", required=True)
+    argparser.add_argument("--output-file", required=True)
+    argparser.add_argument("--config-path", required=True)
+    argparser.add_argument("--datatype", required=True)
+    argparser.add_argument("--timestamp", required=True)
+    argparser.add_argument("--sipm-name", required=True)
+    argparser.add_argument("--dsp-db", nargs="*", default=[])
+    argparser.add_argument("--logfile")
+    args = argparser.parse_args()
+
+    # dataflow configs
+    df_configs = (
+        TextDB(args.config_path, lazy=True)
+        .on(args.timestamp, system=args.datatype)
+        .snakemake_rules.pars_spms_dsp_trg_thr
+    )
+
+    # setup logging
+    log = build_log(df_configs, args.logfile)
+
+    fwhm = get_channel_trg_thr(
+        df_configs,
+        args.sipm_name,
+        args.dsp_db,
+        args.raw_file,
+        args.raw_table_name,
+        log,
+    )
+
     log.debug(f"writing out baseline_curr_fwhm = {fwhm}")
     Path(args.output_file).parent.mkdir(parents=True, exist_ok=True)
     Props.write_to(
         args.output_file,
         {"baseline_curr_fwhm": float(fwhm) if fwhm is not None else fwhm},
     )
+
+
+def par_spms_dsp_trg_thr_multi() -> None:
+    # CLI interface
+    argparser = argparse.ArgumentParser()
+    argparser.add_argument("--raw-file", required=True)
+    argparser.add_argument("--raw-table-names", required=True, nargs="*")
+    argparser.add_argument("--output-file", required=True)
+    argparser.add_argument("--config-path", required=True)
+    argparser.add_argument("--datatype", required=True)
+    argparser.add_argument("--timestamp", required=True)
+    argparser.add_argument("--sipm-names", required=True, nargs="*")
+    argparser.add_argument("--dsp-db", nargs="*", default=[])
+    argparser.add_argument("--logfile")
+    args = argparser.parse_args()
+
+    # dataflow configs
+    df_configs = (
+        TextDB(args.config_path, lazy=True)
+        .on(args.timestamp, system=args.datatype)
+        .snakemake_rules.pars_spms_dsp_trg_thr
+    )
+
+    # setup logging
+    log = build_log(df_configs, args.logfile)
+
+    out_dict = {}
+    for sipm_name, raw_table_name in zip(args.sipm_names, args.raw_table_names):
+        fwhm = get_channel_trg_thr(
+            df_configs,
+            sipm_name,
+            args.dsp_db,
+            args.raw_file,
+            raw_table_name,
+            log,
+        )
+
+        log.debug(f"baseline_curr_fwhm for {sipm_name} = {fwhm}")
+        out_dict[sipm_name] = {
+            "baseline_curr_fwhm": float(fwhm) if fwhm is not None else fwhm
+        }
+    Path(args.output_file).parent.mkdir(parents=True, exist_ok=True)
+    Props.write_to(
+        args.output_file,
+        out_dict,
+    )