Merge branch 'release/2.11.0'

Author: juhuntenburg

brainbox/io/one.py

@@ -941,17 +941,18 @@ def _get_spike_sorting_collection(self, spike_sorter='pykilosort', revision=None
         _logger.debug(f"selecting: {collection} to load amongst candidates: {self.collections}")
         return collection
 
-    def download_spike_sorting_object(self, obj, spike_sorter='pykilosort', dataset_types=None):
+    def download_spike_sorting_object(self, obj, spike_sorter='pykilosort', dataset_types=None, collection=None):
         """
         Downloads an ALF object
         :param obj: object name, str between 'spikes', 'clusters' or 'channels'
         :param spike_sorter: (defaults to 'pykilosort')
-        :param dataset_types: list of extra dataset types
+        :param dataset_types: list of extra dataset types, for example ['spikes.samples']
+        :param collection: string specifiying the collection, for example 'alf/probe01/pykilosort'
         :return:
         """
         if len(self.collections) == 0:
             return {}, {}, {}
-        self.collection = self._get_spike_sorting_collection(spike_sorter=spike_sorter)
+        self.collection = collection or self._get_spike_sorting_collection(spike_sorter=spike_sorter)
         _logger.debug(f"loading spike sorting from {self.collection}")
         spike_attributes, cluster_attributes = self._get_attributes(dataset_types)
         attributes = {'spikes': spike_attributes, 'clusters': cluster_attributes, 'channels': None,

ibllib/__init__.py

@@ -1,4 +1,4 @@
-__version__ = "2.10.6"
+__version__ = "2.11.0"
 import warnings
 
 from ibllib.misc import logger_config

ibllib/dsp/voltage.py

@@ -218,8 +218,8 @@ def interpolate_bad_channels(data, channel_labels=None, h=None, p=1.3, kriging_d
             data[i, :] = 0
             continue
         data[i, :] = np.matmul(weights[imult], data[imult, :])
-    # from easyqc.gui import viewseis
-    # f = viewseis(data.T, si=1/30, h=h, title='interp2', taxis=0)
+    # from viewephys.gui import viewephys
+    # f = viewephys(data.T, fs=1/30, h=h, title='interp2')
     return data
 
 

ibllib/oneibl/data_handlers.py

@@ -47,10 +47,15 @@ def getData(self, one=None):
 
         one = one or self.one
         session_datasets = one.list_datasets(one.path2eid(self.session_path), details=True)
-        df = pd.DataFrame(columns=one._cache.datasets.columns)
+        dfs = []
         for file in self.signature['input_files']:
-            df = df.append(filter_datasets(session_datasets, filename=file[0], collection=file[1],
-                           wildcards=True, assert_unique=False))
+            dfs.append(filter_datasets(session_datasets, filename=file[0], collection=file[1],
+                       wildcards=True, assert_unique=False))
+        df = pd.concat(dfs)
+
+        # Some cases the eid is stored in the index. If so we drop this level
+        if 'eid' in df.index.names:
+            df = df.droplevel(level='eid')
         return df
 
     def uploadData(self, outputs, version):
@@ -228,7 +233,7 @@ def setUp(self):
         :return:
         """
         df = super().getData()
-        self.one._download_datasets(df)
+        self.one._check_filesystem(df)
 
     def uploadData(self, outputs, version, **kwargs):
         """

ibllib/pipes/ephys_preprocessing.py

@@ -1007,11 +1007,14 @@ def _run(self, cams=None, overwrite=False):
                         executable="/bin/bash",
                     )
                     info, error = process.communicate()
-                    info_str = info.decode("utf-8").strip()
-                    _logger.info(info_str)
+                    # info_str = info.decode("utf-8").strip()
+                    # _logger.info(info_str)
                     if process.returncode != 0:
                         error_str = error.decode("utf-8").strip()
-                        _logger.error(f'DLC failed for {cam}Camera\n {error_str}')
+                        _logger.error(f'DLC failed for {cam}Camera.\n\n'
+                                      f'++++++++ Output of subprocess for debugging ++++++++\n\n'
+                                      f'{error_str}\n'
+                                      f'++++++++++++++++++++++++++++++++++++++++++++\n')
                         self.status = -1
                         # We dont' run motion energy, or add any files if dlc failed to run
                         continue
@@ -1029,11 +1032,14 @@ def _run(self, cams=None, overwrite=False):
                         executable="/bin/bash",
                     )
                     info, error = process.communicate()
-                    info_str = info.decode("utf-8").strip()
-                    _logger.info(info_str)
+                    # info_str = info.decode("utf-8").strip()
+                    # _logger.info(info_str)
                     if process.returncode != 0:
                         error_str = error.decode("utf-8").strip()
-                        _logger.error(f'Motion energy failed for {cam}Camera \n {error_str}')
+                        _logger.error(f'Motion energy failed for {cam}Camera.\n\n'
+                                      f'++++++++ Output of subprocess for debugging ++++++++\n\n'
+                                      f'{error_str}\n'
+                                      f'++++++++++++++++++++++++++++++++++++++++++++\n')
                         self.status = -1
                         continue
                     actual_outputs.append(next(self.session_path.joinpath('alf').glob(

ibllib/pipes/local_server.py

@@ -16,7 +16,7 @@
 import ibllib.oneibl.registration as registration
 
 _logger = logging.getLogger('ibllib')
-LARGE_TASKS = ['EphysVideoCompress', 'TrainingVideoCompress', 'SpikeSorting', 'EphysDLC']  # 'TrainingDLC',
+LARGE_TASKS = ['EphysVideoCompress', 'TrainingVideoCompress', 'SpikeSorting', 'EphysDLC']
 
 
 def _get_pipeline_class(session_path, one):
@@ -135,39 +135,39 @@ def job_creator(root_path, one=None, dry=False, rerun=False, max_md5_size=None):
     return all_datasets
 
 
-def job_runner(subjects_path, mode='all', lab=None, dry=False, one=None, count=5):
+def task_queue(mode='all', lab=None, one=None):
     """
-    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:
-    :return:
+    Query waiting jobs from the specified Lab
+    :param mode: Whether to return all waiting tasks, or only small or large (specified in LARGE_TASKS) jobs
+    :param lab: lab name as per Alyx, otherwise try to infer from local globus install
+    :param one: ONE instance
+    -------
+
     """
     if one is None:
         one = ONE(cache_rest=None)
     if lab is None:
+        _logger.info("Trying to infer lab from globus installation")
         lab = _get_lab(one)
     if lab is None:
+        _logger.error("No lab provided or found")
         return  # if the lab is none, this will return empty tasks each time
     # Filter for tasks
     if mode == 'all':
-        tasks = one.alyx.rest('tasks', 'list', status='Waiting',
-                              django=f'session__lab__name__in,{lab}', no_cache=True)
+        waiting_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]
+        waiting_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)
+        waiting_tasks = one.alyx.rest('tasks', 'list', status='Waiting',
+                                      django=f'session__lab__name__in,{lab},name__in,{LARGE_TASKS}', no_cache=True)
 
     # Order tasks by priority
-    tasks = sorted(tasks, key=lambda d: d['priority'], reverse=True)
+    sorted_tasks = sorted(waiting_tasks, key=lambda d: d['priority'], reverse=True)
 
-    tasks_runner(subjects_path, tasks, one=one, count=count, time_out=3600, dry=dry)
+    return sorted_tasks
 
 
 def tasks_runner(subjects_path, tasks_dict, one=None, dry=False, count=5, time_out=None, **kwargs):

release_notes.md

@@ -1,23 +1,29 @@
-## Release Note 2.10
+## Release Notes 2.11
 
-## Release Note 2.10.6 2022-03-15
+### Release Notes 2.11.0 2022-03-08
+- brainbox.io.one.SpikeSortingLoader: option to load using `collection` argument
+- Restructuring of how jobs are run on local servers, run large jobs as service
+
+## Release Notes 2.10
+
+### Release Notes 2.10.6 2022-03-15
 - Allow parent tasks to be 'Incomplete' to run task on local server
 - Change one base_rul for dlc_qc_plot on cortexlab 
 
-## Release Note 2.10.5 2022-03-11
+### Release Notes 2.10.5 2022-03-11
 - Fix moot release accident
 
-## Release Note 2.10.4 2022-03-11
+### Release Notes 2.10.4 2022-03-11
 - Data handler connects to correct alyx database on cortexlab
 
-## Release Note 2.10.3 2022-03-09
+### Release Note 2.10.3 2022-03-09
 - Fixes to EphysPostDLC
 - Small change to storing in dsp.voltage.decompress_destripe_cbin function
 
-## Release Note 2.10.2 2022-02-28
+### Release Notes 2.10.2 2022-02-28
 - Small fixes to local server task queues
 
-## Release Note 2.10.1 2022-02-22
+### Release Notes 2.10.1 2022-02-22
 - Authenticate alyx user in Task class
 - Some fixes to make dlc_qc_plot in EphysPostDLC more reliable
 - SpikeGlx:
@@ -28,32 +34,32 @@
 - voltage: decompress cbin
   - add support for custom spikeglx.Reader
 
-## Release Note 2.10.0 2022-02-11
+### Release Notes 2.10.0 2022-02-11
 - Fix in EphysDLC task to not return multiple copies of outputs
 - Loading examples for different IBL data types
 - Fix for probe syncing when Nidq and probe pulses don't match
 - Account for new ONE tables in ond datahandler
 - Add bad channels plots into RawEphysQc task
 
-## Release Note 2.9
+## Release Notes 2.9
 
 ### Release Notes 2.9.1 2022-01-24
 - deprecation warnings and documentation for spike sorting loading method
 - bugfix: remove lru_cache on AllenAtlas class for iblviewer
 
-### Release Note 2.9.0 2022-01-24
+### Release Notes 2.9.0 2022-01-24
 - Adding EphysDLC task in ephys_preprocessing pipeline
 - NOTE: requires DLC environment to be set up on local servers! 
 - Fixes to EphysPostDLC dlc_qc_plot
 
-## Release Note 2.8
+## Release Notes 2.8
 
 ### Release Notes 2.8.0 2022-01-19
 - Add lfp, aprms, spike raster and behaviour report plots to task infastructure
 - Computation of apRMS in decompress_destripe_cbin before conversion to normalised units
 - Add SpikeSortingLoader class in brainbox.io.one
 
-## Release Note 2.7
+## Release Notes 2.7
 
 ### Release Notes 2.7.1 2022-01-05