[AnalysisQC] Enable common args per analysis

* use for instance
  --add-common-args EMCAL-shm-segment-size 2500000000
  which will be forwarded as a common argument and appended
  to the analysis pipeline of EMCAL analysis as
  --shm-segment-size 2500000000

* Use also e.g.
  ALL-readers <value>
  to influence set --readers <values> for all analyses

* <analysis>-<key> <value>
  supersedes
  ALL-<key> <value>

Author: Benedikt Volkel

MC/analysis_testing/analysis_test.sh

@@ -24,6 +24,7 @@ NTF=$(find ./ -name "tf*" -type d | wc | awk '//{print $1}')
 include_disabled=
 testanalysis=
 aod=
+add_common_args=
 
 if [[ "${#}" == "1" ]] ; then
     # make it backward-compatible
@@ -48,6 +49,12 @@ else
                 shift
                 shift
                 ;;
+            --add-common-args)
+                add_common_args=" ${2} ${3} "
+                shift
+                shift
+                shift
+                ;;
             *)
                 echo "ERROR: Unknown argument ${1}"
                 exit 1
@@ -59,6 +66,7 @@ fi
 # basic checks
 [[ "${testanalysis}" == "" ]] && { echo "ERROR: No analysis specified to be run" ; exit 1 ; }
 [[ "${aod}" == "" ]] && { echo "ERROR: No AOD found to be analysed" ; exit 1 ; }
+[[ "${add_common_args}" != "" ]] && add_common_args="--add-common-args ${add_common_args}"
 
 # check if enabled
 enabled=$($O2DPG_ROOT/MC/analysis_testing/o2dpg_analysis_test_config.py check -t ${testanalysis} --status)
@@ -68,7 +76,9 @@ enabled=$($O2DPG_ROOT/MC/analysis_testing/o2dpg_analysis_test_config.py check -t
 mkdir Analysis 2>/dev/null
 include_disabled=${include_disabled:+--include-disabled}
 workflow_path="Analysis/workflow_analysis_test_${testanalysis}.json"
-$O2DPG_ROOT/MC/analysis_testing/o2dpg_analysis_test_workflow.py --is-mc -f ${aod} -o ${workflow_path} --only-analyses ${testanalysis} ${include_disabled}
+rm ${workflow_path} 2>/dev/null
+$O2DPG_ROOT/MC/analysis_testing/o2dpg_analysis_test_workflow.py --is-mc -f ${aod} -o ${workflow_path} --only-analyses ${testanalysis} ${include_disabled} ${add_common_args}
+[[ ! -f "${workflow_path}" ]] && { echo "Could not construct workflow for analysis ${testanalysis}" ; exit 1 ; }
 $O2DPG_ROOT/MC/bin/o2_dpg_workflow_runner.py -f ${workflow_path} -tt Analysis_${testanalysis}$ --rerun-from Analysis_${testanalysis}$
 
 RC=$?

MC/analysis_testing/o2dpg_analysis_test_utils.py

@@ -67,3 +67,50 @@ def get_collision_system(collision_system=None):
 def full_ana_name(raw_ana_name):
     """Make the standard name of the analysis how it should appear in the workflow"""
     return f"{ANALYSIS_LABEL}_{raw_ana_name}"
+
+
+def get_common_args_as_string(analysis_name, all_common_args):
+    """
+    all_common_args is of the form
+    [<ana_name1>-shm-segment-size <value>, <ana_name2>-readers <value>, ...]
+
+    Find common arguments for this specific analysis
+    """
+
+    def make_args_string(args_map_in):
+        out_string = ""
+        for key, value in args_map_in.items():
+            out_string += f" --{key} {value}"
+        return out_string
+
+    # default arguments for all analyses
+    args_map = {"shm-segment-size": 2000000000,
+                "readers": 1,
+                "aod-memory-rate-limit": 500000000}
+
+    # arguments dedicated for this analysis
+    args_map_overwrite = {}
+
+    if not all_common_args:
+        return make_args_string(args_map)
+
+    if len(all_common_args) % 2:
+        print("ERROR: Cannot digest common args.")
+        return None
+
+    for i in range(0, len(all_common_args), 2):
+        tokens = all_common_args[i].split("-")
+        key = "-".join(tokens[1:])
+        if tokens[0] == analysis_name:
+            # for this analysis, add to dedicated dict
+            args_map_overwrite[key] = all_common_args[i+1]
+            continue
+        if tokens[0] == "ALL":
+            # otherwise add to default dict
+            args_map[key] = all_common_args[i+1]
+
+    # overwrite default dict with dedicated arguments
+    for key, value in args_map_overwrite.items():
+        args_map[key] = value
+
+    return make_args_string(args_map)

MC/analysis_testing/o2dpg_analysis_test_workflow.py

@@ -90,9 +90,7 @@
 from o2dpg_analysis_test_utils import *
 
 
-def create_ana_task(name, cmd, output_dir, *, needs=None, shmsegmentsize="--shm-segment-size 2000000000",
-                    aodmemoryratelimit="--aod-memory-rate-limit 500000000",
-                    readers="--readers 1", extraarguments="-b", is_mc=False):
+def create_ana_task(name, cmd, output_dir, *, needs=None, extraarguments="-b", is_mc=False):
     """Quick helper to create analysis task
 
     This creates an analysis task from various arguments
@@ -107,12 +105,6 @@ def create_ana_task(name, cmd, output_dir, *, needs=None, shmsegmentsize="--shm-
     Keyword args (optional):
         needs: tuple, list
             list of other tasks to be run before
-        shmsegmentsize: str
-            O2/DPL argument string for shared mem size
-        aodmemoryratelimit: str
-            O2/DPL argument string for AOD memory rate limit
-        readers: O2/DPL argument string
-            number of readers
         extraarguments: str
             O2/DPL argument string for any other desired arguments to be added to the executed cmd
     Return:
@@ -125,7 +117,7 @@ def create_ana_task(name, cmd, output_dir, *, needs=None, shmsegmentsize="--shm-
     task = createTask(name=full_ana_name(name), cwd=join(output_dir, name), lab=[ANALYSIS_LABEL, name], cpu=1, mem='2000', needs=needs)
     if is_mc:
         task["labels"].append(ANALYSIS_LABEL_ON_MC)
-    task['cmd'] = f"{cmd} {shmsegmentsize} {aodmemoryratelimit} {readers} {extraarguments}"
+    task['cmd'] = f"{cmd} {extraarguments}"
     return task
 
 
@@ -220,7 +212,7 @@ def get_additional_workflows(input_aod):
     return additional_workflows
 
 
-def add_analysis_tasks(workflow, input_aod="./AO2D.root", output_dir="./Analysis", *, analyses_only=None, is_mc=True, collision_system=None, needs=None, autoset_converters=False, include_disabled_analyses=False, timeout=None):
+def add_analysis_tasks(workflow, input_aod="./AO2D.root", output_dir="./Analysis", *, analyses_only=None, is_mc=True, collision_system=None, needs=None, autoset_converters=False, include_disabled_analyses=False, timeout=None, add_common_args=None):
     """Add default analyses to user workflow
 
     Args:
@@ -265,12 +257,18 @@ def add_analysis_tasks(workflow, input_aod="./AO2D.root", output_dir="./Analysis
             continue
         print(f"INFO: Analysis {ana['name']} uses configuration {configuration}")
 
+        add_common_args_ana = get_common_args_as_string(ana["name"], add_common_args)
+        if not add_common_args_ana:
+            print(f"ERROR: Cannot parse common args for analysis {ana['name']}")
+            continue
+
         for i in additional_workflows:
             if i not in ana["tasks"]:
                 # print("Appending extra task", i, "to analysis", ana["name"], "as it is not there yet and needed for conversion")
                 ana["tasks"].append(i)
         piped_analysis = f" --configuration {configuration} | ".join(ana["tasks"])
         piped_analysis += f" --configuration {configuration} --aod-file {input_aod}"
+        piped_analysis += add_common_args_ana
         if timeout is not None:
             piped_analysis += f" --time-limit {timeout}"
         workflow.append(create_ana_task(ana["name"], piped_analysis, output_dir, needs=needs, is_mc=is_mc))
@@ -330,7 +328,7 @@ def run(args):
         return 1
 
     workflow = []
-    add_analysis_tasks(workflow, args.input_file, expanduser(args.analysis_dir), is_mc=args.is_mc, analyses_only=args.only_analyses, autoset_converters=args.autoset_converters, include_disabled_analyses=args.include_disabled, timeout=args.timeout, collision_system=args.collision_system)
+    add_analysis_tasks(workflow, args.input_file, expanduser(args.analysis_dir), is_mc=args.is_mc, analyses_only=args.only_analyses, autoset_converters=args.autoset_converters, include_disabled_analyses=args.include_disabled, timeout=args.timeout, collision_system=args.collision_system, add_common_args=args.add_common_args)
     if args.with_qc_upload:
         add_analysis_qc_upload_tasks(workflow, args.period_name, args.run_number, args.pass_name)
     if not workflow:
@@ -356,6 +354,7 @@ def main():
     parser.add_argument("--autoset-converters", dest="autoset_converters", action="store_true", help="Compatibility mode to automatically set the converters for the analysis")
     parser.add_argument("--timeout", type=int, default=None, help="Timeout for analysis tasks in seconds.")
     parser.add_argument("--collision-system", dest="collision_system", help="Set the collision system. If not set, tried to be derived from ALIEN_JDL_LPMInterationType. Fallback to pp")
+    parser.add_argument("--add-common-args", dest="add_common_args", nargs="*", help="Pass additional common arguments per analysis, for instance --add-common-args EMCAL-shm-segment-size 2500000000 will add --shm-segment-size 2500000000 to the EMCAL analysis")
     parser.set_defaults(func=run)
     args = parser.parse_args()
     return(args.func(args))