Merge branch 'split_run_job' into develop

Author: juhuntenburg

ibllib/ephys/ephysqc.py

@@ -110,7 +110,8 @@ def run(self, update: bool = False, overwrite: bool = True, stream: bool = None,
         if self.data.ap_meta:
             rms_file = self.probe_path.joinpath("_iblqc_ephysChannels.apRMS.npy")
             if rms_file.exists() and not overwrite:
-                _logger.warning(f'File {rms_file} already exists and overwrite=False. Skipping RMS compute.')
+                _logger.warning(f'RMS map already exists for .ap data in {self.probe_path}, skipping. '
+                                f'Use overwrite option.')
                 median_rms = np.load(rms_file)
             else:
                 rl = self.data.ap_meta.fileTimeSecs

ibllib/pipes/ephys_preprocessing.py

@@ -69,8 +69,12 @@ def _run(self, overwrite=False):
             pids = [p['id'] for p in create_alyx_probe_insertions(self.session_path, one=self.one)]
         qc_files = []
         for pid in pids:
-            eqc = ephysqc.EphysQC(pid, session_path=self.session_path, one=self.one)
-            qc_files.extend(eqc.run(update=True, overwrite=overwrite))
+            try:
+                eqc = ephysqc.EphysQC(pid, session_path=self.session_path, one=self.one)
+                qc_files.extend(eqc.run(update=True, overwrite=overwrite))
+            except AssertionError:
+                self.status = -1
+                continue
         return qc_files
 
 

ibllib/pipes/local_server.py

@@ -16,6 +16,7 @@
 import ibllib.oneibl.registration as registration
 
 _logger = logging.getLogger('ibllib')
+LARGE_TASKS = ['EphysVideoCompress', 'TrainingVideoCompress', 'EphysDLC', 'TrainingDLC', 'SpikeSorting']
 
 
 def _get_pipeline_class(session_path, one):
@@ -134,11 +135,12 @@ def job_creator(root_path, one=None, dry=False, rerun=False, max_md5_size=None):
     return all_datasets
 
 
-def job_runner(subjects_path, lab=None, dry=False, one=None, count=5):
+def job_runner(subjects_path, mode='all', lab=None, dry=False, one=None, count=5):
     """
     Function to be used as a process to run the jobs as they are created on the database
     This will query waiting jobs from the specified Lab
     :param subjects_path: on servers: /mnt/s0/Data/Subjects. Contains sessions
+    :param mode: Whether to run all jobs, or only small or large (video compression, DLC, spike sorting) jobs
     :param lab: lab name as per Alyx
     :param dry:
     :param count:
@@ -150,8 +152,18 @@ def job_runner(subjects_path, lab=None, dry=False, one=None, count=5):
         lab = _get_lab(one)
     if lab is None:
         return  # if the lab is none, this will return empty tasks each time
-    tasks = one.alyx.rest('tasks', 'list', status='Waiting',
-                          django=f'session__lab__name__in,{lab}', no_cache=True)
+    # Filter for tasks
+    if mode == 'all':
+        tasks = one.alyx.rest('tasks', 'list', status='Waiting',
+                              django=f'session__lab__name__in,{lab}', no_cache=True)
+    elif mode == 'small':
+        tasks_all = one.alyx.rest('tasks', 'list', status='Waiting',
+                                  django=f'session__lab__name__in,{lab}', no_cache=True)
+        tasks = [t for t in tasks_all if t['name'] not in LARGE_TASKS]
+    elif mode == 'large':
+        tasks = one.alyx.rest('tasks', 'list', status='Waiting',
+                              django=f'session__lab__name__in,{lab},name__in,{LARGE_TASKS}', no_cache=True)
+
     tasks_runner(subjects_path, tasks, one=one, count=count, time_out=3600, dry=dry)