general clean up and documentation

Author: isehle

README.md

@@ -103,7 +103,7 @@ The `PU0` files above were merged and are stored under `/data_CMS/cms/alves/L1HG
 <a id="job-submission"></a>
 ## Job Submission
 
-Job submission to HT Condor is handled through `bye_splits/production/submit_scripts/job_submit.py` using the `job` section of `config.yaml` for its configuration. The configuration should include usual condor variables, i.e `user`, `proxy`, `queue`, and `local`, as well as a path to the `script` you would like to run on condor and a list of `arguments` that the script accepts. It then contains a section for each particle type which should contain a `submit_dir`, i.e. the directory in which to read and write submission related files, `files` which should be a `.txt` file containing the values you would like to iterate over, and `files_per_batch` which can be any number between 1 and the total number of values you would like to run.
+Job submission to HT Condor is handled through `bye_splits/production/submit_scripts/job_submit.py` using the `job` section of `config.yaml` for its configuration. The configuration should include usual condor variables, i.e `user`, `proxy`, `queue`, and `local`, as well as a path to the `script` you would like to run on condor. The `arguments` sub-section should contain `key/value` pairs matching the expected arguments that `script` accepts. The variable that you would like to iterate over should be set in `iterOver` and its value should correspond to a `key` in the `arguments` sub-section whose value is a list containing the values the script should iterate over. It then contains a section for each particle type which should contain a `submit_dir`, i.e. the directory in which to read and write submission related files, and `args_per_batch` which can be any number between 1 and `len(arguments[<iterOver>])`.
 
 
 <a id="org0bc224d"></a>

bye_splits/production/produce.cc

@@ -7,51 +7,105 @@
 parent_dir = os.path.abspath(__file__ + 5 * "../")
 sys.path.insert(0, parent_dir)
 
-from bye_splits.utils import params, common
+from bye_splits.utils import params, common, job_helpers
 
 import subprocess
 import yaml
 
-# Read particle specific variables from the YAML file
-particle_var = lambda part, var: config["job"][part][var]
-
 class JobBatches:
+    """Class for setting up job batches and setting configuration
+    variables. The function setup_batches() will take the list in
+    config[arguments[<iterOver>]] and return a list of lists containing
+    <args_per_batch> values in each sublist. Example for five total values
+    with <args_per_batch> = 2:
+    [0.01, 0.02, 0.03, 0.04, 0.05] --> [[0.01, 0.02], [0.03, 0.04], [0.05]]"""
+
     def __init__(self, particle, config):
         self.particle = particle
-        self.config = config
+        self.config   = config
         self.iterOver = config["job"]["iterOver"]
+        self.particle_var = lambda part, var: config["job"][part][var]
 
     def setup_batches(self):
         total = self.config["job"]["arguments"][self.iterOver]
 
-        vals_per_batch = particle_var(self.particle, "files_per_batch")    
+        vals_per_batch = self.particle_var(self.particle, "args_per_batch")    
 
         batches = [total[i: i + vals_per_batch] for i in range(0, len(total), vals_per_batch)]
 
         return batches
 
 class CondJobBase:
     def __init__(self, particle, config):
-        self.particle = particle
-        self.script = config["job"]["script"]
-        self.iterOver = config["job"]["iterOver"]
-        self.args = config["job"]["arguments"]
-        self.queue = config["job"]["queue"]
-        self.proxy = config["job"]["proxy"]
-        self.local = config["job"]["local"]
-        self.user = config["job"]["user"]
-        self.particle_dir = particle_var(self.particle, "submit_dir")
-        self.batches = JobBatches(particle, config).setup_batches()
+        self.particle     = particle
+        self.script       = config["job"]["script"]
+        self.iterOver     = config["job"]["iterOver"]
+        self.args         = config["job"]["arguments"]
+        self.queue        = config["job"]["queue"]
+        self.proxy        = config["job"]["proxy"]
+        self.local        = config["job"]["local"]
+        self.user         = config["job"]["user"]
+        self.batch        = JobBatches(particle, config)
+        self.particle_dir = self.batch.particle_var(self.particle, "submit_dir")
+        self.batches      = self.batch.setup_batches()
+
+    def _write_arg_keys(self, current_version):
+        """Writes the line containing the argument
+        names to the buffer list."""
+        
+        arg_keys = ["Arguments ="]
+        for arg in self.args.keys():
+            arg_keys.append("$({})".format(arg))
+        arg_keys = " ".join(arg_keys) + "\n"
+
+        current_version.append(arg_keys)
+
+    def _write_arg_values(self, current_version):
+        """Adds the argument values, where the batch lists are converted
+        to strings as [val_1, val_2, ...] --> "[val_1;val_2]".
+        The choice of a semicolon as the delimiter is arbitrary but it
+        cannot be a comma because this is the delimeter condor itself uses.
+
+        Example:
+
+        queue radius, particle from (
+        [0.01, 0.02], photon
+        )
+        incorrectly assigns radius="[0.01", particle="0.02]"
+
+        queue radius, particle from (
+            [0.01;0.02], photon
+        )
+        correctly assigns radius="[0.01, 0.02]", particle="photon"
+        """   
+        arg_keys = ", ".join(self.args.keys())
+        arg_keys = "queue " + arg_keys + " from (\n"
+        current_version.append(arg_keys)
+        for batch in self.batches:
+            sub_args = list(self.args.keys())[1:]
+            arg_vals = [self.args[key] for key in sub_args]
+            all_vals = ["{}".format(batch).replace(", ", ";")]+arg_vals
+            all_vals = ", ".join(all_vals) + "\n"
+            current_version.append(all_vals)
+
+        current_version.append(")")
 
     def prepare_batch_submission(self):
+        """Writes the .sh script that constitutes the executable
+        in the .sub script. The basename will be the same as the fundamental
+        script, followed by a version number. Stores the contents in a list that's
+        used as a buffer and checked against the content in previous
+        versions, only writing the file if an identical file doesn't
+        already exist. The version number will be incrimented in this case."""
+
         sub_dir = "{}subs/".format(self.particle_dir)
 
-        if not os.path.exists(sub_dir):
-            os.makedirs(sub_dir)
+        common.create_dir(sub_dir)
+
         script_basename = os.path.basename(self.script).replace(".sh", "").replace(".py", "")
 
         submit_file_name_template = "{}{}_submit.sh".format(sub_dir, script_basename)
-        submit_file_versions = common.grab_most_recent(submit_file_name_template, return_all=True)
+        submit_file_versions = job_helpers.grab_most_recent(submit_file_name_template, return_all=True)
 
         current_version = []
         current_version.append("#!/usr/bin/env bash\n")
@@ -71,26 +125,27 @@ def prepare_batch_submission(self):
             current_version.append("bash {}".format(self.script))
 
         # Write the file only if an identical file doesn't already exist
-        self.sub_file = common.conditional_write(submit_file_versions, submit_file_name_template, current_version)
+        self.sub_file = job_helpers.conditional_write(submit_file_versions, submit_file_name_template, current_version)
 
     def prepare_multi_job_condor(self):
+        """Writes the .sub script that is submitted to HT Condor.
+        Follows the same naming convention and conditional_write()
+        procedure as the previous function."""
+
         log_dir = "{}logs/".format(self.particle_dir)
 
         script_basename = os.path.basename(self.script).replace(".sh", "").replace(".py", "")
 
         job_file_name_template = "{}jobs/{}.sub".format(self.particle_dir, script_basename)
 
-        job_file_versions = common.grab_most_recent(job_file_name_template, return_all=True)
+        job_file_versions = job_helpers.grab_most_recent(job_file_name_template, return_all=True)
 
         current_version = []
         current_version.append("executable = {}\n".format(self.sub_file))
         current_version.append("Universe              = vanilla\n")
-        if len(self.args) > 0:
-            current_version.append("Arguments =")
 
-            for arg in self.args.keys():
-                current_version.append(" $({}) ".format(arg))
-            current_version.append("\n")
+        if len(self.args) > 0:
+            self._write_arg_keys(current_version)
 
         current_version.append("output = {}{}_C$(Cluster)P$(Process).out\n".format(log_dir, script_basename))
         current_version.append("error = {}{}_C$(Cluster)P$(Process).err\n".format(log_dir, script_basename))
@@ -100,52 +155,36 @@ def prepare_multi_job_condor(self):
         current_version.append("WNTag                 = el7\n")
         current_version.append('+SingularityCmd       = ""\n')
         current_version.append("include: /opt/exp_soft/cms/t3/t3queue |\n")
-
+        
         if len(self.args.keys()) > 0:
-            arg_keys = ", ".join(self.args.keys())
-            arg_keys = "queue " + arg_keys + " from (\n"
-            current_version.append(arg_keys)
-            for batch in self.batches:
-                sub_args = list(self.args.keys())[1:]
-                arg_vals = [self.args[key] for key in sub_args]
-                all_vals = ["{}".format(batch).replace(", ", ";")]+arg_vals
-                all_vals = ", ".join(all_vals) + "\n"
-                current_version.append(all_vals)
-
-            current_version.append(")")
+            self._write_arg_values(current_version)
 
         # Write the file only if an identical file doesn't already exist
-        self.submission_file = common.conditional_write(job_file_versions, job_file_name_template, current_version) # Save to launch later
+        self.submission_file = job_helpers.conditional_write(job_file_versions, job_file_name_template, current_version) # Save to launch later
 
 class CondJob:
+    """Creates the job directories and files
+    with prepare_jobs() and runs the jobs with
+    launch_jobs()."""
+
     def __init__(self, particle, config):
         self.base = CondJobBase(particle=particle, config=config)
 
-
     def prepare_jobs(self):
 
-        configs = lambda dir: dir + "configs"
-        jobs = lambda dir: dir + "jobs"
-        logs = lambda dir: dir + "logs"
-
-        config_dir = configs(self.base.particle_dir)
-        job_dir = jobs(self.base.particle_dir)
-        log_dir = logs(self.base.particle_dir)
+        config_dir = self.base.particle_dir + "configs"
+        job_dir    = self.base.particle_dir + "jobs"
+        log_dir    = self.base.particle_dir + "logs"
 
-        if not os.path.exists(config_dir):
-            os.makedirs(config_dir)
-        if not os.path.exists(job_dir):
-            os.makedirs(job_dir)
-        if not os.path.exists(log_dir):
-            os.makedirs(log_dir)
+        for d in (config_dir, job_dir, log_dir):
+            common.create_dir(d)
 
         self.base.prepare_batch_submission()
         self.base.prepare_multi_job_condor()
 
-
     def launch_jobs(self):
 
-        if self.base.local == True:
+        if self.base.local:
             machine = "local"
         else:
             machine = "llrt3.in2p3.fr"

bye_splits/production/submit_scripts/job_submit.py

@@ -1,6 +1,6 @@
 # coding: utf-8
 
-_all_ = []
+_all_ = ['cluster_radius']
 
 import os
 import sys
@@ -21,6 +21,11 @@
 import yaml
 
 def cluster_radius(pars, cfg):
+    """Runs the default clustering algorithm using the
+    specified radius. Runs on both negative and positive
+    eta files, and adds layer weights to the cluster
+    kwargs if specified."""
+    
     cluster_d = params.read_task_params("cluster")
 
     particles = pars["particles"]
@@ -37,19 +42,20 @@ def cluster_radius(pars, cfg):
     if "weights" in cfg:
         cluster_d["weights"] = cfg["weights"]
 
-    for key in ("ClusterInTC", "ClusterInSeeds", "ClusterOutPlot", "ClusterOutValidation"):
-        name = cluster_d[key]
+    for eta_tag in ("negEta", "posEta"):
+        for key in ("ClusterInTC", "ClusterInSeeds", "ClusterOutPlot", "ClusterOutValidation"):
+            name = cluster_d[key]
 
-        cluster_d[key] =  "{}_{}_{}_posEta_9oct".format(particles, pileup, name)
-    
-    nevents_end = tasks.cluster.cluster_default(pars, **cluster_d)
+            cluster_d[key] =  "{}_{}_{}_{}".format(particles, pileup, name, eta_tag)
+        
+        nevents_end = tasks.cluster.cluster_default(pars, **cluster_d)
 
 if __name__ == "__main__":
     parser = argparse.ArgumentParser(description="")
     parser.add_argument("--radius", help="Coefficient to use as the max cluster radius", required=True, type=float)
     parser.add_argument("--particles", choices=("photons", "electrons", "pions"), required=True)
     parser.add_argument("--pileup", help="tag for PU200 vs PU0", choices=("PU0", "PU200"), required=True)
-    parser.add_argument("--weighted", help="Apply pre-calculated layer weights", default=False)
+    parser.add_argument("--weighted", help="Apply pre-caluclated layer weights", action="store_true")
     parsing.add_parameters(parser)
 
     FLAGS = parser.parse_args()