A lot of fixes in complete_run.py

Author: gipert

rules/main.smk

@@ -48,6 +48,6 @@ rule autogen_output:
         filedb_path=os.path.join(pars_path(setup), "filedb"),
         setup=lambda wildcards: setup,
         basedir=basedir,
-    threads: workflow.cores
+    threads: min(workflow.cores, 64)
     script:
         "../scripts/complete_run.py"

rules/raw.smk

@@ -21,7 +21,7 @@ raw_par_catalog = ParsKeyResolve.get_par_catalog(
 
 rule build_raw_orca:
     """
-    This rule runs build_raw, it takes in a file.{daq_ext} and outputs a raw file
+    This rule runs build_raw(), it takes in a file.fcio and outputs a raw file
     """
     input:
         get_pattern_tier_daq(setup, extension="orca"),
@@ -51,7 +51,7 @@ rule build_raw_orca:
 
 rule build_raw_fcio:
     """
-    This rule runs build_raw, it takes in a file.{daq_ext} and outputs a raw file
+    This rule runs build_raw(), it takes in a file.fcio and outputs a raw file
     """
     input:
         get_pattern_tier_daq(setup, extension="fcio"),

scripts/build_fdb.py

@@ -3,7 +3,7 @@
 from pathlib import Path
 
 import numpy as np
-from legendmeta.catalog import Props
+from dbetto.catalog import Props
 from lgdo import lh5
 from pygama.flow.file_db import FileDB
 

scripts/check_blinding.py

@@ -84,7 +84,7 @@
 # check for peaks within +- 5keV of  2614 and 583 to ensure blinding still
 # valid and if so create file else raise error.  if detector is in ac mode it
 # will always pass this check
-if np.any(np.abs(maxs - 2614) < 5) and np.any(np.abs(maxs - 583) < 5) or det_status is False:
+if (np.any(np.abs(maxs - 2614) < 5) and np.any(np.abs(maxs - 583) < 5)) or det_status is False:
     Path(args.output).parent.mkdir(parents=True, exist_ok=True)
     Props.write_to(args.output, {})
 else:

scripts/complete_run.py

@@ -3,6 +3,7 @@
 import datetime
 import json
 import os
+import subprocess
 import time
 from pathlib import Path
 
@@ -157,12 +158,12 @@ def find_gen_runs(gen_tier_path):
     # first look for non-concat tiers
     paths = gen_tier_path.glob("*/*/*/*")
     # use the directories to build a datatype/period/run string
-    runs = {"/".join(p.name.split("/")[-3:]) for p in paths}
+    runs = {"/".join(str(p).split("/")[-3:]) for p in paths}
 
     # then look for concat tiers (use filenames now)
     paths_concat = gen_tier_path.glob("*/*/*.lh5")
     # use the directories to build a datatype/period/run string
-    runs_concat = {"/".join([p.name.split("-")[3]] + p.name.split("-")[1:3]) for p in paths_concat}
+    runs_concat = {"/".join([str(p).split("-")[3]] + str(p).split("-")[1:3]) for p in paths_concat}
 
     return runs | runs_concat
 
@@ -188,30 +189,32 @@ def build_file_dbs(gen_tier_path, outdir):
         logfile = Path(ut.tmp_log_path(snakemake.params.setup)) / outfile.with_suffix(".log").name
         print(f"INFO: ......building {outfile}")
 
-        cmdline = ut.runcmd(snakemake.params.setup, aslist=True)
-        prodenv = as_ro(os.getenv("PRODENV"))
-        cmdline += [f"--env=PRODENV={prodenv}"]
+        cmdline = [
+            *ut.runcmd(snakemake.params.setup, aslist=True),
+            "--",
+            "python3",
+            "-B",
+            f"{snakemake.params.basedir}/scripts/build_fdb.py",
+            "--scan-path",
+            spec,
+            "--output",
+            str(outfile),
+            "--config",
+            str(outdir / "file_db_config.json"),
+            "--log",
+            str(logfile),
+        ]
+
+        if speck[0] == "phy":
+            cmdline += ["--assume-nonsparse"]
+
+        print(cmdline)
+        print(" ".join(cmdline))
+
+        cmdenv = {}
 
         # TODO: forward stdout to log file
-        processes.add(
-            subprocess.Popen(
-                [
-                    *cmdline,
-                    "python3",
-                    "-B",
-                    f"{snakemake.params.basedir}/scripts/build_fdb.py",
-                    "--scan-path",
-                    spec,
-                    "--output",
-                    str(outfile),
-                    "--config",
-                    str(outdir / "file_db_config.json"),
-                    "--log",
-                    str(logfile),
-                    "--assume-nonsparse" if speck[0] == "phy" else "",
-                ],
-            )
-        )
+        processes.add(subprocess.Popen(cmdline))
 
         if len(processes) >= snakemake.threads:
             os.wait()

scripts/util/__init__.py

@@ -12,18 +12,18 @@
 )
 
 __all__ = [
-    "Props",
-    "PropsStream",
-    "Catalog",
-    "ParsKeyResolve",
     "CalGrouping",
-    "FileKey",
-    "ProcessingFileKey",
+    "Catalog",
     "ChannelProcKey",
+    "FileKey",
     "ParsCatalog",
-    "unix_time",
+    "ParsKeyResolve",
+    "ProcessingFileKey",
+    "Props",
+    "PropsStream",
     "runcmd",
-    "subst_vars_impl",
     "subst_vars",
+    "subst_vars_impl",
     "subst_vars_in_snakemake_config",
+    "unix_time",
 ]

scripts/util/utils.py

@@ -110,9 +110,9 @@ def filelist_path(setup):
 
 def runcmd(setup, aslist=False):
     cmdline = shlex.split(setup["execenv"]["cmd"])
-    cmdline += ["--env=" + "'PYTHONUSERBASE=" + f"{setup['paths']['install']}" + "'"]
+    cmdline += ["--env=" + "PYTHONUSERBASE=" + f"{setup['paths']['install']}"]
     if "env" in setup["execenv"]:
-        cmdline += [f'--env="{var}={val}"' for var, val in setup["execenv"]["env"].items()]
+        cmdline += [f"--env={var}={val}" for var, val in setup["execenv"]["env"].items()]
 
     cmdline += shlex.split(setup["execenv"]["arg"])