Iblrigv8 (#641)

* small atlas fix

* Prepare experiment takes in an optional device id parameter

* typo

* session_params: session creation creates stub in local and remote

* Create stubs logging

* copy ephys: update prints to logger and option to transfer single sessions

* transfer ephys data logging and compatibility dynamic pipeline

* Actually we don't want a status flag

* fixing some lies in the type hints

* bugfix get collection: yaml can interpret some keys as datestrings

* un bon patch bien dégueulasse

* Fix training CW extraction v8

* add atlas test and add temporary requirement for widefield package

* add optional task_collection arg to compute training status

* skimage deprecated in favour of scikit-image

* typo

* training status computation compatible with chained protocols

* Dynamic pipeline training status

* finish merge

* add some logging to the jobs on server - make sure each run is separate

* fix training_status task failing in training pipeline integration

* dynamic pipeline lookups for specific subtask with sync suffix if not found

* TaskQC no longer saves to disk; use get_bpod_extractor

* training status task: do not raise on extraction error

* Auto stash before checking out "origin/iblrigv8"

stop training regression when two protocols on same day

* fix for bpod camera extraction get correct collection

* don't pass into qc

* pop the task collection

* fix tests

* flake after merge

---------

Co-authored-by: Miles Wells <[email protected]>
Co-authored-by: Mayo Faulkner <[email protected]>

Author: 3 people

ibllib/atlas/atlas.py

@@ -1162,16 +1162,19 @@ def from_dict(d, brain_atlas=None):
         >>> tri = {'x': 544.0, 'y': 1285.0, 'z': 0.0, 'phi': 0.0, 'theta': 5.0, 'depth': 4501.0}
         >>> ins = Insertion.from_dict(tri)
         """
+        assert brain_atlas, 'Input argument brain_atlas must be defined'
         z = d['z'] / 1e6
-        if brain_atlas:
-            iy = brain_atlas.bc.y2i(d['y'] / 1e6)
-            ix = brain_atlas.bc.x2i(d['x'] / 1e6)
-            # Only use the brain surface value as z if it isn't NaN (this happens when the surface
-            # touches the edges of the atlas volume
-            if not np.isnan(brain_atlas.top[iy, ix]):
-                z = brain_atlas.top[iy, ix]
-        return Insertion(x=d['x'] / 1e6, y=d['y'] / 1e6, z=z, phi=d['phi'], theta=d['theta'],
-                         depth=d['depth'] / 1e6, beta=d.get('beta', 0), label=d.get('label', ''))
+        if not hasattr(brain_atlas, 'top'):
+            brain_atlas.compute_surface()
+        iy = brain_atlas.bc.y2i(d['y'] / 1e6)
+        ix = brain_atlas.bc.x2i(d['x'] / 1e6)
+        # Only use the brain surface value as z if it isn't NaN (this happens when the surface touches the edges
+        # of the atlas volume
+        if not np.isnan(brain_atlas.top[iy, ix]):
+            z = brain_atlas.top[iy, ix]
+        return Insertion(x=d['x'] / 1e6, y=d['y'] / 1e6, z=z,
+                         phi=d['phi'], theta=d['theta'], depth=d['depth'] / 1e6,
+                         beta=d.get('beta', 0), label=d.get('label', ''))
 
     @property
     def trajectory(self):

ibllib/graphic.py

@@ -128,7 +128,7 @@ class BaseBpodTrialsExtractor(BaseExtractor):
     settings = None
     task_collection = None
 
-    def extract(self, task_collection='raw_behavior_data', bpod_trials=None, settings=None, **kwargs):
+    def extract(self, bpod_trials=None, settings=None, **kwargs):
         """
         :param: bpod_trials (optional) bpod trials from jsonable in a dictionary
         :param: settings (optional) bpod iblrig settings json file in a dictionary
@@ -139,7 +139,7 @@ def extract(self, task_collection='raw_behavior_data', bpod_trials=None, setting
         """
         self.bpod_trials = bpod_trials
         self.settings = settings
-        self.task_collection = task_collection
+        self.task_collection = kwargs.pop('task_collection', 'raw_behavior_data')
         if self.bpod_trials is None:
             self.bpod_trials = raw.load_data(self.session_path, task_collection=self.task_collection)
         if not self.settings:
@@ -252,13 +252,13 @@ def get_session_extractor_type(session_path, task_collection='raw_behavior_data'
         return False
 
 
-def get_pipeline(session_path):
+def get_pipeline(session_path, task_collection='raw_behavior_data'):
     """
     Get the pre-processing pipeline name from a session path
     :param session_path:
     :return:
     """
-    stype = get_session_extractor_type(session_path)
+    stype = get_session_extractor_type(session_path, task_collection=task_collection)
     return _get_pipeline_from_task_type(stype)
 
 

ibllib/io/extractors/base.py

@@ -5,6 +5,7 @@
 import logging
 import importlib
 from collections import OrderedDict
+import warnings
 
 from pkg_resources import parse_version
 from ibllib.io.extractors import habituation_trials, training_trials, biased_trials, opto_trials
@@ -54,6 +55,7 @@ def extract_all(session_path, save=True, bpod_trials=None, settings=None,
     list of pathlib.Path
         The output files if save is true.
     """
+    warnings.warn('`extract_all` functions soon to be deprecated, use `bpod_trials.get_bpod_extractor` instead', FutureWarning)
     if not extractor_type:
         extractor_type = get_session_extractor_type(session_path, task_collection=task_collection)
     _logger.info(f'Extracting {session_path} as {extractor_type}')
@@ -101,6 +103,23 @@ def extract_all(session_path, save=True, bpod_trials=None, settings=None,
 
 
 def get_bpod_extractor(session_path, protocol=None, task_collection='raw_behavior_data') -> BaseBpodTrialsExtractor:
+    """
+    Returns an extractor for a given session.
+
+    Parameters
+    ----------
+    session_path : str, pathlib.Path
+        The path to the session to be extracted.
+    protocol : str, optional
+        The protocol name, otherwise uses the PYBPOD_PROTOCOL key in iblrig task settings files.
+    task_collection : str
+        The folder within the session that contains the raw task data.
+
+    Returns
+    -------
+    BaseBpodTrialsExtractor
+        An instance of the task extractor class, instantiated with the session path.
+    """
     builtins = {
         'HabituationTrials': HabituationTrials,
         'TrainingTrials': TrainingTrials,

ibllib/io/extractors/bpod_trials.py

@@ -238,9 +238,9 @@ def _extract(self, video_path=None, display=False, extrapolate_missing=True, **k
         # is empty, or contains only one value (i.e. doesn't change)
         if gpio is not None and gpio['indices'].size > 1:
             _logger.info('Aligning to sync TTLs')
-            task_collection = kwargs.get('task_collection', 'raw_behavior_data')
             # Extract audio TTLs
-            _, audio = raw.load_bpod_fronts(self.session_path, data=self.bpod_trials, task_collection=task_collection)
+            _, audio = raw.load_bpod_fronts(self.session_path, data=self.bpod_trials,
+                                            task_collection=self.task_collection)
             _, ts = raw.load_camera_ssv_times(self.session_path, 'left')
             """
             There are many sync TTLs that are for some reason missed by the GPIO.  Conversely

ibllib/io/extractors/camera.py

@@ -52,10 +52,18 @@ class ContrastLR(BaseBpodTrialsExtractor):
     var_names = ('contrastLeft', 'contrastRight')
 
     def _extract(self):
-        contrastLeft = np.array([t['contrast']['value'] if np.sign(
-            t['position']) < 0 else np.nan for t in self.bpod_trials])
-        contrastRight = np.array([t['contrast']['value'] if np.sign(
-            t['position']) > 0 else np.nan for t in self.bpod_trials])
+        # iblrigv8 has only flat values in the trial table so we can switch to parquet table when times come
+        # and all the clutter here would fit in ~30 lines
+        if isinstance(self.bpod_trials[0]['contrast'], float):
+            contrastLeft = np.array([t['contrast'] if np.sign(
+                t['position']) < 0 else np.nan for t in self.bpod_trials])
+            contrastRight = np.array([t['contrast'] if np.sign(
+                t['position']) > 0 else np.nan for t in self.bpod_trials])
+        else:
+            contrastLeft = np.array([t['contrast']['value'] if np.sign(
+                t['position']) < 0 else np.nan for t in self.bpod_trials])
+            contrastRight = np.array([t['contrast']['value'] if np.sign(
+                t['position']) > 0 else np.nan for t in self.bpod_trials])
 
         return contrastLeft, contrastRight
 
@@ -112,9 +120,13 @@ class RepNum(BaseBpodTrialsExtractor):
     var_names = 'repNum'
 
     def _extract(self):
-        trial_repeated = np.array(
-            [t['contrast']['type'] == 'RepeatContrast' for t in self.bpod_trials])
-        trial_repeated = trial_repeated.astype(int)
+        def get_trial_repeat(trial):
+            if 'debias_trial' in trial:
+                return trial['debias_trial']
+            else:
+                return trial['contrast']['type'] == 'RepeatContrast'
+
+        trial_repeated = np.array(list(map(get_trial_repeat, self.bpod_trials))).astype(int)
         repNum = trial_repeated.copy()
         c = 0
         for i in range(len(trial_repeated)):

ibllib/io/extractors/training_trials.py

@@ -328,6 +328,12 @@ def _read_settings_json_compatibility_enforced(settings):
     # 2018-12-05 Version 3.2.3 fixes (permanent fixes in IBL_RIG from 3.2.4 on)
     if md['IBLRIG_VERSION_TAG'] == '':
         pass
+    elif parse_version(md.get('IBLRIG_VERSION_TAG')) >= parse_version('8.0.0'):
+        md['SESSION_NUMBER'] = str(md['SESSION_NUMBER']).zfill(3)
+        md['PYBPOD_BOARD'] = md['RIG_NAME']
+        md['PYBPOD_CREATOR'] = (md['ALYX_USER'], '')
+        md['SESSION_DATE'] = md['SESSION_START_TIME'][:10]
+        md['SESSION_DATETIME'] = md['SESSION_START_TIME']
     elif parse_version(md.get('IBLRIG_VERSION_TAG')) <= parse_version('3.2.3'):
         if 'LAST_TRIAL_DATA' in md.keys():
             md.pop('LAST_TRIAL_DATA')

ibllib/io/raw_data_loaders.py

@@ -27,7 +27,6 @@
 from datetime import datetime
 import logging
 from pathlib import Path
-import warnings
 from copy import deepcopy
 
 from one.converters import ConversionMixin
@@ -405,14 +404,12 @@ def get_collections(sess_params):
 
     def iter_dict(d):
         for k, v in d.items():
-            if not v or isinstance(v, str):
-                continue
             if isinstance(v, list):
                 for d in filter(lambda x: isinstance(x, dict), v):
                     iter_dict(d)
-            elif 'collection' in v:
+            elif isinstance(v, dict) and 'collection' in v:
                 collection_map[k] = v['collection']
-            else:
+            elif isinstance(v, dict):
                 iter_dict(v)
 
     iter_dict(sess_params)
@@ -459,7 +456,7 @@ def get_remote_stub_name(session_path, device_id=None):
     return session_path / '_devices' / remote_filename
 
 
-def prepare_experiment(session_path, acquisition_description=None, local=None, remote=None, overwrite=False):
+def prepare_experiment(session_path, acquisition_description=None, local=None, remote=None, device_id=None, overwrite=False):
     """
     Copy acquisition description yaml to the server and local transfers folder.
 
@@ -471,30 +468,46 @@ def prepare_experiment(session_path, acquisition_description=None, local=None, r
         The data to write to the experiment.description.yaml file.
     local : str, pathlib.Path
         The path to the local session folders.
+        >>> C:\iblrigv8_data\cortexlab\Subjects  # noqa
     remote : str, pathlib.Path
         The path to the remote server session folders.
+        >>> Y:\Subjects  # noqa
+    device_id : str, optional
+        A device name, if None the TRANSFER_LABEL parameter is used (defaults to this device's
+        hostname with a unique numeric ID)
     overwrite : bool
         If true, overwrite any existing file with the new one, otherwise, update the existing file.
     """
     if not acquisition_description:
         return
+
     # Determine if user passed in arg for local/remote subject folder locations or pull in from
-    # local param file or prompt user if missing
-    params = misc.create_basic_transfer_params(local_data_path=local, remote_data_path=remote)
+    # local param file or prompt user if missing data.
+    if local is None or remote is None or device_id is None:
+        params = misc.create_basic_transfer_params(local_data_path=local, remote_data_path=remote, TRANSFER_LABEL=device_id)
+        local, device_id = (params['DATA_FOLDER_PATH'], params['TRANSFER_LABEL'])
+        # if the user provides False as an argument, it means the intent is to not copy anything, this
+        # won't be preserved by create_basic_transfer_params by default
+        remote = False if remote is False else params['REMOTE_DATA_FOLDER_PATH']
+
+    # THis is in the docstring but still, if the session Path is absolute, we need to make it relative
+    if Path(session_path).is_absolute():
+        session_path = Path(*session_path.parts[-3:])
 
     # First attempt to copy to server
-    local_only = remote is False or params.get('REMOTE_DATA_FOLDER_PATH', False) is False
-    if not local_only:
-        remote_device_path = get_remote_stub_name(session_path, params['TRANSFER_LABEL'])
+    if remote is not False:
+        remote_session_path = Path(remote).joinpath(session_path)
+        remote_device_path = get_remote_stub_name(remote_session_path, device_id=device_id)
         previous_description = read_params(remote_device_path) if remote_device_path.exists() and not overwrite else {}
         try:
             write_yaml(remote_device_path, merge_params(previous_description, acquisition_description))
+            _logger.info(f'Written data to remote device at: {remote_device_path}.')
         except Exception as ex:
-            warnings.warn(f'Failed to write data to {remote_device_path}: {ex}')
+            _logger.warning(f'Failed to write data to remote device at: {remote_device_path}. \n {ex}')
 
-    # Now copy to local directory
-    local = params.get('TRANSFERS_PATH', params['DATA_FOLDER_PATH'])
-    filename = f'_ibl_experiment.description_{params["TRANSFER_LABEL"]}.yaml'
+    # then create on the local machine
+    filename = f'_ibl_experiment.description_{device_id}.yaml'
     local_device_path = Path(local).joinpath(session_path, filename)
     previous_description = read_params(local_device_path) if local_device_path.exists() and not overwrite else {}
     write_yaml(local_device_path, merge_params(previous_description, acquisition_description))
+    _logger.info(f'Written data to local session at : {local_device_path}.')

ibllib/io/session_params.py

@@ -77,9 +77,8 @@ def _run(self, update=True):
         """
         Extracts an iblrig training session
         """
-        save_path = self.session_path.joinpath(self.output_collection)
-        trials, wheel, output_files = bpod_trials.extract_all(
-            self.session_path, save=True, task_collection=self.collection, save_path=save_path)
+        extractor = bpod_trials.get_bpod_extractor(self.session_path, task_collection=self.collection)
+        trials, output_files = extractor.extract(task_collection=self.collection, save=True)
 
         if trials is None:
             return None
@@ -88,8 +87,8 @@ def _run(self, update=True):
         # Run the task QC
         # Compile task data for QC
         qc = HabituationQC(self.session_path, one=self.one)
-        qc.extractor = TaskQCExtractor(self.session_path, one=self.one, sync_collection=self.sync_collection, sync_type=self.sync,
-                                       task_collection=self.collection, save_path=save_path)
+        qc.extractor = TaskQCExtractor(self.session_path, sync_collection=self.sync_collection,
+                                       one=self.one, sync_type=self.sync, task_collection=self.collection)
         namespace = 'task' if self.protocol_number is None else f'task_{self.protocol_number:02}'
         qc.run(update=update, namespace=namespace)
         return output_files
@@ -239,24 +238,25 @@ def _run(self, update=True):
         """
         Extracts an iblrig training session
         """
-        save_path = self.session_path.joinpath(self.output_collection)
-        trials, wheel, output_files = bpod_trials.extract_all(
-            self.session_path, save=True, task_collection=self.collection, save_path=save_path)
+        extractor = bpod_trials.get_bpod_extractor(self.session_path, task_collection=self.collection)
+        extractor.default_path = self.output_collection
+        trials, output_files = extractor.extract(task_collection=self.collection, save=True)
         if trials is None:
             return None
         if self.one is None or self.one.offline:
             return output_files
         # Run the task QC
         # Compile task data for QC
         type = get_session_extractor_type(self.session_path, task_collection=self.collection)
+        # FIXME Task data should not need re-extracting
         if type == 'habituation':
             qc = HabituationQC(self.session_path, one=self.one)
             qc.extractor = TaskQCExtractor(self.session_path, one=self.one, sync_collection=self.sync_collection,
-                                           sync_type=self.sync, task_collection=self.collection, save_path=save_path)
+                                           sync_type=self.sync, task_collection=self.collection)
         else:  # Update wheel data
             qc = TaskQC(self.session_path, one=self.one)
             qc.extractor = TaskQCExtractor(self.session_path, one=self.one, sync_collection=self.sync_collection,
-                                           sync_type=self.sync, task_collection=self.collection, save_path=save_path)
+                                           sync_type=self.sync, task_collection=self.collection)
             qc.extractor.wheel_encoding = 'X1'
         # Aggregate and update Alyx QC fields
         namespace = 'task' if self.protocol_number is None else f'task_{self.protocol_number:02}'
@@ -323,8 +323,7 @@ def _run_qc(self, trials_data, update=True, plot_qc=True):
         # Run the task QC
         qc = TaskQC(self.session_path, one=self.one, log=_logger)
         qc.extractor = TaskQCExtractor(self.session_path, lazy=True, one=qc.one, sync_collection=self.sync_collection,
-                                       sync_type=self.sync, task_collection=self.collection,
-                                       save_path=self.session_path.joinpath(self.output_collection))
+                                       sync_type=self.sync, task_collection=self.collection)
         # Extract extra datasets required for QC
         qc.extractor.data = trials_data  # FIXME This line is pointless
         qc.extractor.extract_data()
@@ -423,8 +422,7 @@ def _run_qc(self, trials_data, update=True, **kwargs):
         # TODO Task QC extractor for Timeline
         qc = TaskQC(self.session_path, one=self.one, log=_logger)
         qc.extractor = TaskQCExtractor(self.session_path, lazy=True, one=qc.one, sync_collection=self.sync_collection,
-                                       sync_type=self.sync, task_collection=self.collection,
-                                       save_path=self.session_path.joinpath(self.output_collection))
+                                       sync_type=self.sync, task_collection=self.collection)
         # Extract extra datasets required for QC
         qc.extractor.data = TaskQCExtractor.rename_data(trials_data.copy())
         qc.extractor.load_raw_data()
@@ -466,7 +464,8 @@ def _run(self, upload=True):
 
         df = training_status.get_latest_training_information(self.session_path, one)
         if df is not None:
-            training_status.make_plots(self.session_path, self.one, df=df, save=True, upload=upload)
+            training_status.make_plots(
+                self.session_path, self.one, df=df, save=True, upload=upload, task_collection=self.collection)
             # Update status map in JSON field of subjects endpoint
             if self.one and not self.one.offline:
                 _logger.debug('Updating JSON field of subjects endpoint')