Add corrections eff

cax/main.py

@@ -187,6 +187,9 @@ def massive():
                         help="Select the cluster partition")
     parser.add_argument('--reservation', type=str,
                         help="Select the reservation")
+    parser.add_argument('--no-batch-mode', dest='no_batch_mode', action='store_true',
+                        help="Use massive-cax without batch queue submission.")
+
 
     args = parser.parse_args()
 
@@ -195,6 +198,7 @@ def massive():
         exit()
 
     run_once = args.once
+    no_batch_mode = args.no_batch_mode
 
     config_arg = ''
     if args.config_file:
@@ -288,10 +292,48 @@ def massive():
         docs = list(collection.find(query,
                                     sort=sort_key,
                                     projection=['start', 'number','name',
-                                                'detector', '_id']))
+                                                'detector', '_id', 'processor']))
 
         for doc in docs:
+
+            #If no_batch_mode is chosen, no jobs will submitted to batch queues
+            if no_batch_mode == True:
+
+                #Define basic command:
+                basic_command = """#!/bin/bash
+cax --once {config} --name {name}
+"""
+
+                print("no-batch-mode")
+
+                #Pre-read the config file for the no-batch-mode modus:
+                config.set_json(os.path.abspath(args.config_file))
+                task_list = config.get_config(config.get_hostname())['task_list']
+
+                #This is a list of tasks which refer to apply corrections to a run:
+                task_list_correction = ['AddElectronLifetime', 'AddGains', 'AddDriftVelocity', 'SetS2xyMap', 'SetLightCollectionEfficiency', 'SetFieldDistortion', 'SetNeuralNetwork']
+
+                #--------------------
+                #Start with single activities: Apply corrections only to data sets which do not have any correction yet:
+                if bool(set(task_list) & set(task_list_correction)) or task_list_correction == task_list:
+
+                    this_run_version = doc.get('processor', {}).get('correction_versions', {})
+                    if len(this_run_version) == 0:
+                        command = basic_command.format(config=config_arg, name=doc['name'])
+                        logging.info(command)
+                        stdout_value = qsub.command_submission( command )
+
+                        #Return command output:
+                        for i in stdout_value:
+                           logging.info('AddCorrections: %s', i)
+
+                #--------------------
+                #Start with single activities: Add another activity:
+
+                #Nothing else
+                continue
 
+            #Start with processing via batch (if no 'add corrections' was requested)
             job_name = ''
 
             if doc['detector'] == 'tpc':
@@ -336,10 +378,10 @@ def massive():
 
         if run_once:
             break
-        #else:
-        #    pace = 5
-        #    logging.info("Done, waiting %d minutes" % pace)
-        #    time.sleep(60*pace) # Pace 5 minutes
+        else:
+           pace = 5
+           logging.info("Done, waiting %d minutes" % pace)
+           time.sleep(60*pace) # Pace 5 minutes
 
 
 def move():

cax/qsub.py

@@ -124,3 +124,21 @@ def get_queue(host=config.get_hostname(), partition=''):
     if len(queue_list) > 1:
         return queue_list[1:]
     return []
+
+
+def command_submission(command):
+
+    #Submit the command
+    sc = create_script(command)
+    execute = subprocess.Popen( ['sh', sc.name] , \
+                                stdin=subprocess.PIPE,\
+                                stdout=subprocess.PIPE,\
+                                stderr=subprocess.STDOUT, shell=False )
+    stdout_value, stderr_value = execute.communicate()
+    stdout_value = stdout_value.decode("utf-8")
+    stdout_value = stdout_value.split("\n")
+    #delete script:
+    delete_script( sc )
+
+    return stdout_value
+

cax/tasks/corrections.py

@@ -56,11 +56,9 @@ def each_run(self):
         # We can't do this in init: cax is a long-running application, and corrections may change while it is running.
         self.correction_doc = cdoc = self.correction_collection.find_one(sort=(('calculation_time', -1), ))
         self.version = cdoc.get('version', str(cdoc['calculation_time']))
-
         # Get the correction sympy function, if one is set
         if 'function' in cdoc:
             self.function = parse_expr(cdoc['function'])
-
         # Check if this correction's version correction has already been applied. If so, skip this run.
         classname = self.__class__.__name__
         this_run_version = self.run_doc.get('processor', {}).get('correction_versions', {}).get(classname, 'not_set')