Decide to start ipcluster per-notebook, not per calibration type

xfel_calibrate/calibrate.py

@@ -611,7 +611,7 @@ def remove_duplications(l):
 def concurrent_run(temp_path, nb, nbname, args, cparm=None, cval=None,
                    final_job=False, job_list=[], fmt_args={}, cluster_cores=8,
                    sequential=False, dep_jids=[],
-                   show_title=True, cluster_profile='NO_CLUSTER'):
+                   show_title=True):
     """ Launch a concurrent job on the cluster via SLURM
     """
 
@@ -619,12 +619,18 @@ def concurrent_run(temp_path, nb, nbname, args, cparm=None, cval=None,
         args[cparm] = cval
 
     suffix = flatten_list(cval)
-    if "cluster_profile" in args:
-        args["cluster_profile"] = "{}_{}".format(cluster_profile, suffix)
 
     # first convert the notebook
     parms = extract_parameters(nb)
+
+    if has_parm(parms, "cluster_profile"):
+        cluster_profile = f"{args['cluster_profile']}_{suffix}"
+    else:
+        # Don't start ipcluster if there's no cluster_profile parameter
+        cluster_profile = 'NO_CLUSTER'
+
     params = parameter_values(parms, **args)
+    params = parameter_values(params, cluster_profile=cluster_profile)
     new_nb = replace_definitions(nb, params, execute=False)
     if not show_title:
         first_markdown_cell(new_nb).source = ''
@@ -651,7 +657,7 @@ def concurrent_run(temp_path, nb, nbname, args, cparm=None, cval=None,
     srun_base += [os.path.join(PKG_DIR, "bin", "slurm_calibrate.sh"),  # path to helper sh
                   os.path.abspath(nbpath),  # path to notebook
                   python_path,  # path to python
-                  args.get("cluster_profile", "NO_CLUSTER"),
+                  cluster_profile,
                   '"{}"'.format(base_name.upper()),
                   '"{}"'.format(args["detector"].upper()),
                   '"{}"'.format(args["type"].upper()),
@@ -789,12 +795,9 @@ def run():
             concurrency["parameter"])
         warnings.warn(msg, RuntimeWarning)
 
-    cluster_profile = "NO_CLUSTER"
-    if not has_parm(parms, "cluster_profile"):
-        warnings.warn("Notebook has no cluster_profile parameter, " +
-                      "running on cluster will likely fail!", RuntimeWarning)
-    elif "cluster_profile" not in args or args["cluster_profile"] == parser.get_default('cluster_profile'):
-        cluster_profile = "slurm_prof_{}".format(run_uuid)
+    # If not explicitly specified, use a new profile for ipcluster
+    if args.get("cluster_profile") in {None, parser.get_default("cluster_profile")}:
+        args['cluster_profile'] = "slurm_prof_{}".format(run_uuid)
 
     # create a temporary output directory to work in
     run_tmp_path = os.path.join(temp_path, f"slurm_out_{detector}_{caltype}_{run_uuid}")
@@ -854,7 +857,7 @@ def run():
                                job_list=joblist, fmt_args=fmt_args,
                                cluster_cores=cluster_cores,
                                sequential=sequential,
-                               cluster_profile=cluster_profile)
+                              )
         joblist.append(jobid)
 
     if concurrency.get("parameter", None) is None:
@@ -865,7 +868,7 @@ def run():
                                cluster_cores=cluster_cores,
                                sequential=sequential,
                                dep_jids=joblist,
-                               cluster_profile=cluster_profile)
+                              )
         joblist.append(jobid)
     else:
         cvar = concurrency["parameter"]
@@ -948,7 +951,7 @@ def run():
                                job_list=joblist, fmt_args=fmt_args,
                                cluster_cores=cluster_cores,
                                sequential=sequential,
-                               cluster_profile=cluster_profile)
+                              )
         joblist.append(jobid)
 
     if not all([j is None for j in joblist]):