Projects (#390)

* local server: look for pipeline availability

* local server: look for pipeline availability

* small fix get task type

* remove karolina project from main tasks

* get pipeline

* point to personal project

* register task as personal project

* bugfix

* gnagna

* ⚰️

* 🩹

* 🐛

* refactor wheel extraction with function outputing dictionaries

* Implements signature attribute for Task that is a dict with in /out files

* add assertion method on the Taks to test run completion

* add no cache option in the querying of the tasks

* rename the laser datasets to avoid underscores

* re-label the EphysAudio task properly

* fix optogenetics datasets extraction

Author: oliche

ibllib/io/extractors/base.py

@@ -167,6 +167,18 @@ def _get_task_types_json_config():
     return task_types
 
 
+def get_task_protocol(session_path):
+    try:
+        settings = load_settings(get_session_path(session_path))
+    except json.decoder.JSONDecodeError:
+        _logger.error(f"Can't read settings for {session_path}")
+        return
+    if settings:
+        return settings.get('PYBPOD_PROTOCOL', None)
+    else:
+        return
+
+
 def get_task_extractor_type(task_name):
     """
     Returns the task type string from the full pybpod task name:
@@ -176,13 +188,8 @@ def get_task_extractor_type(task_name):
     :return: one of ['biased', 'habituation', 'training', 'ephys', 'mock_ephys', 'sync_ephys']
     """
     if isinstance(task_name, Path):
-        try:
-            settings = load_settings(get_session_path(task_name))
-        except json.decoder.JSONDecodeError:
-            return
-        if settings:
-            task_name = settings.get('PYBPOD_PROTOCOL', None)
-        else:
+        task_name = get_task_protocol(task_name)
+        if task_name is None:
             return
     task_types = _get_task_types_json_config()
     task_type = next((task_types[tt] for tt in task_types if tt in task_name), None)
@@ -225,7 +232,7 @@ def _get_pipeline_from_task_type(stype):
     :param stype: session_type or task extractor type
     :return:
     """
-    if 'ephys' in stype:
+    if stype in ['ephys_biased_opto', 'ephys', 'ephys_training', 'mock_ephys', 'sync_ephys']:
         return 'ephys'
     elif stype in ['habituation', 'training', 'biased', 'biased_opto']:
         return 'training'

ibllib/io/extractors/ephys_fpga.py

@@ -544,6 +544,20 @@ def _get_all_probes_sync(session_path, bin_exists=True):
     return ephys_files
 
 
+def get_wheel_positions(sync, chmap):
+    """
+    Gets the wheel position from synchronisation pulses
+    :param sync:
+    :param chmap:
+    :return:wheel: dictionary with keys 'timestamps' and 'position'
+            moves: dictionary with keys 'intervals' and 'peakAmplitude'
+    """
+    ts, pos = extract_wheel_sync(sync=sync, chmap=chmap)
+    moves = extract_wheel_moves(ts, pos)
+    wheel = {'timestamps': ts, 'position': pos}
+    return wheel, moves
+
+
 def get_main_probe_sync(session_path, bin_exists=False):
     """
     From 3A or 3B multiprobe session, returns the main probe (3A) or nidq sync pulses
@@ -561,7 +575,6 @@ def get_main_probe_sync(session_path, bin_exists=False):
     elif version == '3B':
         # the sync master is the nidq breakout box
         sync_box_ind = np.argmax([1 if ef.get('nidq') else 0 for ef in ephys_files])
-
     sync = ephys_files[sync_box_ind].sync
     sync_chmap = ephys_files[sync_box_ind].sync_map
     return sync, sync_chmap
@@ -684,16 +697,16 @@ def _extract(self, sync=None, chmap=None, **kwargs):
         out.update({k: self.bpod2fpga(bpod_trials[k][ibpod]) for k in bpod_rsync_fields})
         out.update({k: fpga_trials[k][ifpga] for k in sorted(fpga_trials.keys())})
         # extract the wheel data
+        wheel, moves = get_wheel_positions(sync=sync, chmap=chmap)
         from ibllib.io.extractors.training_wheel import extract_first_movement_times
-        ts, pos = extract_wheel_sync(sync=sync, chmap=chmap)
-        moves = extract_wheel_moves(ts, pos)
         settings = raw_data_loaders.load_settings(session_path=self.session_path)
         min_qt = settings.get('QUIESCENT_PERIOD', None)
         first_move_onsets, *_ = extract_first_movement_times(moves, out, min_qt=min_qt)
         out.update({'firstMovement_times': first_move_onsets})
 
         assert tuple(filter(lambda x: 'wheel' not in x, self.var_names)) == tuple(out.keys())
-        return [out[k] for k in out] + [ts, pos, moves['intervals'], moves['peakAmplitude']]
+        return [out[k] for k in out] + [wheel['timestamps'], wheel['position'],
+                                        moves['intervals'], moves['peakAmplitude']]
 
 
 def extract_all(session_path, save=True, bin_exists=False):

ibllib/io/extractors/extractor_types.json

@@ -1,5 +1,4 @@
-{
- "ksocha_ephysOptoStimulation": "ephys_passive_opto",
+{"ksocha_ephysOptoStimulation": "ephys_passive_opto",
  "ksocha_ephysOptoChoiceWorld": "ephys_biased_opto",
  "passiveChoiceWorld": "ephys_replay",
  "opto_ephysChoiceWorld": "ephys_biased_opto",

ibllib/io/extractors/opto_trials.py

@@ -11,8 +11,8 @@ class LaserBool(BaseBpodTrialsExtractor):
     """
     Extracts the laser probabilities from the bpod jsonable
     """
-    save_names = ('_ibl_trials.laser_stimulation.npy', '_ibl_trials.laser_probability.npy')
-    var_names = ('laser_stimulation', 'laser_probability')
+    save_names = ('_ibl_trials.laserStimulation.npy', '_ibl_trials.laserProbability.npy')
+    var_names = ('laserStimulation', 'laserProbability')
 
     def _extract(self, **kwargs):
         _logger.info('Extracting laser datasets')
@@ -41,11 +41,11 @@ def _extract(self, **kwargs):
 
         if np.all(np.isnan(lprob)):
             # this prevents the file from being saved when no data
-            self.save_names = ('_ibl_trials.laser_stimulation.npy', None)
+            self.save_names = ('_ibl_trials.laserStimulation.npy', None)
             _logger.warning('No laser probability found in bpod data')
         if np.all(np.isnan(lstim)):
             # this prevents the file from being saved when no data
-            self.save_names = (None, '_ibl_trials.laser_probability.npy')
+            self.save_names = (None, '_ibl_trials.laserProbability.npy')
             _logger.warning('No laser stimulation found in bpod data')
         return lstim, lprob
 

ibllib/pipes/ephys_preprocessing.py

@@ -58,7 +58,7 @@ class RawEphysQC(tasks.Task):
     io_charge = 30  # this jobs reads raw ap files
     priority = 10  # a lot of jobs depend on this one
     level = 0  # this job doesn't depend on anything
-    input_files = signatures.RAWEPHYSQC
+    signature = {'input_files': signatures.RAWEPHYSQC, 'output_files': ()}
 
     def _run(self, overwrite=False):
         eid = self.one.path2eid(self.session_path)
@@ -76,12 +76,15 @@ def _run(self, overwrite=False):
 
 class EphysAudio(tasks.Task):
     """
-    Computes raw electrophysiology QC
+    Compresses the microphone wav file in a lossless flac file
     """
 
     cpu = 2
     priority = 10  # a lot of jobs depend on this one
     level = 0  # this job doesn't depend on anything
+    signature = {'input_files': ('_iblrig_micData.raw.wav', 'raw_behavior_data', True),
+                 'output_files': ('_iblrig_micData.raw.flac', 'raw_behavior_data', True),
+                 }
 
     def _run(self, overwrite=False):
         command = "ffmpeg -i {file_in} -y -nostdin -c:a flac -nostats {file_out}"
@@ -106,7 +109,7 @@ class SpikeSorting(tasks.Task):
     )
     SPIKE_SORTER_NAME = 'pykilosort'
     PYKILOSORT_REPO = Path.home().joinpath('Documents/PYTHON/SPIKE_SORTING/pykilosort')
-    input_files = signatures.SPIKESORTING
+    signature = {'input_files': signatures.SPIKESORTING, 'output_files': ()}
 
     @staticmethod
     def _sample2v(ap_file):
@@ -285,7 +288,7 @@ def _run(self, **kwargs):
 class EphysTrials(tasks.Task):
     priority = 90
     level = 1
-    input_files = signatures.EPHYSTRIALS
+    signature = {'input_files': signatures.EPHYSTRIALS, 'output_files': ()}
 
     def _behaviour_criterion(self):
         """
@@ -454,7 +457,7 @@ class EphysPassive(tasks.Task):
     cpu = 1
     io_charge = 90
     level = 1
-    input_files = signatures.EPHYSPASSIVE
+    signature = {'input_files': signatures.EPHYSPASSIVE, 'output_files': ()}
 
     def _run(self):
         """returns a list of pathlib.Paths. """

ibllib/pipes/local_server.py

@@ -10,14 +10,29 @@
 
 from one.api import ONE
 
-from ibllib.io.extractors.base import get_session_extractor_type, get_pipeline
-from ibllib.pipes import ephys_preprocessing, training_preprocessing, tasks
+from ibllib.io.extractors.base import get_pipeline, get_task_protocol, get_session_extractor_type
+from ibllib.pipes import tasks, training_preprocessing, ephys_preprocessing
 from ibllib.time import date2isostr
 import ibllib.oneibl.registration as registration
 
 _logger = logging.getLogger('ibllib')
 
 
+def _get_pipeline_class(session_path, one):
+    pipeline = get_pipeline(session_path)
+    if pipeline == 'training':
+        PipelineClass = training_preprocessing.TrainingExtractionPipeline
+    elif pipeline == 'ephys':
+        PipelineClass = ephys_preprocessing.EphysExtractionPipeline
+    else:
+        # try and look if there is a custom extractor in the personal projects extraction class
+        import projects.base
+        task_type = get_session_extractor_type(session_path)
+        PipelineClass = projects.base.get_pipeline(task_type)
+    _logger.info(f"Using {PipelineClass} pipeline for {session_path}")
+    return PipelineClass(session_path=session_path, one=one)
+
+
 def _get_lab(one):
     with open(Path.home().joinpath(".globusonline/lta/client-id.txt"), 'r') as fid:
         globus_id = fid.read()
@@ -101,16 +116,10 @@ def job_creator(root_path, one=None, dry=False, rerun=False, max_md5_size=None):
                 session_path, one=one, max_md5_size=max_md5_size)
             if dsets is not None:
                 all_datasets.extend(dsets)
-            pipeline = get_pipeline(session_path)
-            if pipeline == 'training':
-                pipe = training_preprocessing.TrainingExtractionPipeline(session_path, one=one)
-            # only start extracting ephys on a raw_session.flag
-            elif pipeline == 'ephys' and flag_file.name == 'raw_session.flag':
-                pipe = ephys_preprocessing.EphysExtractionPipeline(session_path, one=one)
-            else:
-                _logger.info(f'Session type {get_session_extractor_type(session_path)}'
-                             f'as no matching pipeline pattern {session_path}')
-                continue
+            pipe = _get_pipeline_class(session_path, one)
+            if pipe is None:
+                task_protocol = get_task_protocol(session_path)
+                _logger.info(f'Session task protocol {task_protocol} has no matching pipeline pattern {session_path}')
             if rerun:
                 rerun__status__in = '__all__'
             else:
@@ -142,7 +151,7 @@ def job_runner(subjects_path, lab=None, dry=False, one=None, count=5):
     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}')
+                          django=f'session__lab__name__in,{lab}', no_cache=True)
     tasks_runner(subjects_path, tasks, one=one, count=count, time_out=3600, dry=dry)
 
 

ibllib/pipes/tasks.py

@@ -38,7 +38,7 @@ class Task(abc.ABC):
     time_out_secs = None
     version = version.ibllib()
     log = ''
-    input_files = None
+    signature = {'input_files': (), 'output_files': ()}  # tuple (filename, collection, required_flag)
 
     def __init__(self, session_path, parents=None, taskid=None, one=None,
                  machine=None, clobber=True, aws=None, location='server'):
@@ -254,12 +254,9 @@ def _getData(self):
         :return:
         """
         assert self.one
-
-        # This will be improved by Olivier new filters
-        session_datasets = self.one.list_datasets(self.one.path2eid(self.session_path),
-                                                  details=True)
+        session_datasets = self.one.list_datasets(self.one.path2eid(self.session_path), details=True)
         df = pd.DataFrame(columns=self.one._cache.datasets.columns)
-        for file in self.input_files:
+        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))
         return df
@@ -278,6 +275,24 @@ def _cleanUp_SDSC(self):
         assert SDSC_PATCH_PATH.parts[0:4] == self.session_path.parts[0:4]
         shutil.rmtree(self.session_path)
 
+    def assert_expected_outputs(self):
+        """
+        After a run, asserts that all signature files are present at least once in the output files
+        Mainly useful for integration tests
+        :return:
+        """
+        assert self.status == 0
+        everthing_is_fine = True
+        for expected_file in self.signature['output_files']:
+            actual_files = list(self.session_path.rglob(str(Path(expected_file[1]).joinpath(expected_file[0]))))
+            if len(actual_files) == 0:
+                everthing_is_fine = False
+                _logger.error(f"Signature file expected {expected_file} not found in the output")
+        if not everthing_is_fine:
+            for out in self.outputs:
+                _logger.error(f"{out}")
+            raise FileNotFoundError("Missing outputs after task completion")
+
 
 class Pipeline(abc.ABC):
     """

ibllib/tests/test_pipes.py

@@ -32,6 +32,7 @@ def test_task_to_pipeline(self):
             ("mock_ephys", "ephys"),
             ("sync_ephys", "ephys"),
             ("ephys", "ephys"),
+            ("ephys_passive_opto", "ephys_passive_opto")
         ]
         for typ, exp in pipe_out:
             assert ibllib.io.extractors.base._get_pipeline_from_task_type(typ) == exp