Merge pull request #783 from int-brain-lab/legacyRetirement

Legacy retirement

ibllib/pipes/ephys_preprocessing.py

@@ -39,8 +39,8 @@
 from ibllib.plots.snapshot import ReportSnapshot
 from brainbox.behavior.dlc import likelihood_threshold, get_licks, get_pupil_diameter, get_smooth_pupil_diameter
 
-_logger = logging.getLogger("ibllib")
-warnings.warn('`pipes.training_preprocessing` to be removed in favour of dynamic pipeline')
+_logger = logging.getLogger('ibllib')
+warnings.warn('`pipes.ephys_preprocessing` to be removed in favour of dynamic pipeline', DeprecationWarning)
 
 
 #  level 0
@@ -53,7 +53,7 @@ class EphysPulses(tasks.Task):
     io_charge = 30  # this jobs reads raw ap files
     priority = 90  # a lot of jobs depend on this one
     level = 0  # this job doesn't depend on anything
-    force = False  # whether or not to force download of missing data on local server if outputs already exist
+    force = False  # whether to force download of missing data on local server if outputs already exist
     signature = {
         'input_files': [('*ap.meta', 'raw_ephys_data/probe*', True),
                         ('*ap.ch', 'raw_ephys_data/probe*', False),  # not necessary when we have .bin file
@@ -219,7 +219,7 @@ def _run(self, overwrite=False):
 
 class SpikeSorting(tasks.Task):
     """
-    Pykilosort 2.5 pipeline
+    (DEPRECATED) Pykilosort 2.5 pipeline
     """
     gpu = 1
     io_charge = 100  # this jobs reads raw ap files
@@ -238,6 +238,12 @@ class SpikeSorting(tasks.Task):
         'output_files': []  # see setUp method for declaration of inputs
     }
 
+    def __init__(self, *args, **kwargs):
+        warnings.warn('`pipes.ephys_preprocessing.SpikeSorting` to be removed '
+                      'in favour of `pipes.ephys_tasks.SpikeSorting`',
+                      DeprecationWarning)
+        super().__init__(*args, **kwargs)
+
     @staticmethod
     def spike_sorting_signature(pname=None):
         pname = pname if pname is not None else "probe*"

ibllib/pipes/local_server.py

@@ -19,10 +19,10 @@
 from one.remote.globus import get_lab_from_endpoint_id, get_local_endpoint_id
 
 from ibllib import __version__ as ibllib_version
-from ibllib.io.extractors.base import get_pipeline, get_task_protocol, get_session_extractor_type
+from ibllib.io.extractors.base import get_pipeline, get_session_extractor_type
 from ibllib.pipes import tasks, training_preprocessing, ephys_preprocessing
 from ibllib.time import date2isostr
-from ibllib.oneibl.registration import IBLRegistrationClient, register_session_raw_data, get_lab
+from ibllib.oneibl.registration import IBLRegistrationClient
 from ibllib.oneibl.data_handlers import get_local_data_repository
 from ibllib.io.session_params import read_params
 from ibllib.pipes.dynamic_pipeline import make_pipeline, acquisition_description_legacy_session
@@ -88,7 +88,7 @@ def report_health(one):
         one.alyx.json_field_update(endpoint='data-repository', uuid=dr['name'], field_name='json', data=status)
 
 
-def job_creator(root_path, one=None, dry=False, rerun=False, max_md5_size=None):
+def job_creator(root_path, one=None, dry=False, rerun=False):
     """
     Create new sessions and pipelines.
 
@@ -108,8 +108,6 @@ def job_creator(root_path, one=None, dry=False, rerun=False, max_md5_size=None):
         If true, simply log the session_path(s) found, without registering anything.
     rerun : bool
         If true and session pipeline tasks already exist, set them all to waiting.
-    max_md5_size : int
-        (legacy sessions) The maximum file size to calculate the MD5 hash sum for.
 
     Returns
     -------
@@ -134,21 +132,11 @@ def job_creator(root_path, one=None, dry=False, rerun=False, max_md5_size=None):
             # if the subject doesn't exist in the database, skip
             rc.register_session(session_path, file_list=False)
 
-            # See if we need to create a dynamic pipeline
-            experiment_description_file = read_params(session_path)
-            if experiment_description_file is not None:
-                pipe = make_pipeline(session_path, one=one)
-            else:
+            # NB: all sessions now extracted using dynamic pipeline
+            if read_params(session_path) is None:
                 # Create legacy experiment description file
                 acquisition_description_legacy_session(session_path, save=True)
-                lab = get_lab(session_path, one.alyx)  # Can be set to None to do this Alyx-side if using ONE v1.20.1
-                _, dsets = register_session_raw_data(session_path, one=one, max_md5_size=max_md5_size, labs=lab)
-                if dsets:
-                    all_datasets.extend(dsets)
-                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}')
+            pipe = make_pipeline(session_path, one=one)
             if rerun:
                 rerun__status__in = '__all__'
             else:

ibllib/tests/test_pipes.py

@@ -11,27 +11,71 @@
 import random
 import string
 from uuid import uuid4
+from datetime import datetime
 
-from one.api import ONE
+from one.api import ONE, OneAlyx
 import iblutil.io.params as iopar
 from packaging.version import Version, InvalidVersion
 
 import ibllib.io.extractors.base
 import ibllib.tests.fixtures.utils as fu
 from ibllib.pipes import misc
 from ibllib.pipes.misc import sleepless
+from ibllib.pipes import local_server
 from ibllib.tests import TEST_DB
 import ibllib.pipes.scan_fix_passive_files as fix
 from ibllib.pipes.base_tasks import RegisterRawDataTask
 from ibllib.pipes.ephys_preprocessing import SpikeSorting
 
 
+class TestLocalServer(unittest.TestCase):
+    """Tests for the ibllib.pipes.local_server module."""
+    def setUp(self):
+        tmp = tempfile.TemporaryDirectory()
+        self.tmpdir = Path(tmp.name)
+        self.addCleanup(tmp.cleanup)
+        raw_behaviour_data = fu.create_fake_raw_behavior_data_folder(self.tmpdir / 'subject/2020-01-01/001', task='ephys')
+        raw_behaviour_data.parent.joinpath('raw_session.flag').touch()
+        fu.populate_task_settings(raw_behaviour_data, patch={'PYBPOD_PROTOCOL': '_iblrig_ephysChoiceWorld5.2.1'})
+        raw_behaviour_data = fu.create_fake_raw_behavior_data_folder(self.tmpdir / 'subject/2020-01-01/002')
+        raw_behaviour_data.parent.joinpath('raw_session.flag').touch()
+        fu.populate_task_settings(raw_behaviour_data, patch={'PYBPOD_PROTOCOL': 'ephys_optoChoiceWorld6.0.1'})
+
+    @mock.patch('ibllib.pipes.local_server.IBLRegistrationClient')
+    @mock.patch('ibllib.pipes.local_server.make_pipeline')
+    def test_job_creator(self, pipeline_mock, _):
+        """Test the job_creator function.
+
+        This test was created after retiring the legacy pipeline. Here we test that an experiment
+        description file is created for each legacy session, followed by dynamic pipeline creation.
+        The second session tests the behaviour of a legacy pipeline with no corresponding experiment
+        description template.  For these sessions we will simply update acquisition_description_legacy_session
+        to add support.
+        """
+        one = mock.Mock(spec=OneAlyx)
+        with self.assertLogs(local_server.__name__, 'ERROR') as log:
+            pipes, _ = local_server.job_creator(self.tmpdir, one=one)
+        self.assertIn("KeyError: 'biased_opto'", log.records[0].getMessage())
+        self.assertEqual(len(pipes), 1)
+        pipeline_mock.assert_called_once()
+
+        # In September 2024, the legacy pipeline will be removed. This entails removing the
+        # code in pipes.training_preprocessing and pipes.ephys_preprocessing, as well as the
+        # code in qc.task_extractors and the extract_all functions in the io.extractors modules.
+        # NB: some tasks such as ephys opto do not have experiment description templates and some
+        # legacy tasks are imported for use in the dynamic pipeline.
+        self.assertFalse(
+            datetime.today() > datetime(2024, 9, 1),
+            'Legacy pipeline code scheduled to be removed after 2024-09-01'
+        )
+
+
 class TestExtractors2Tasks(unittest.TestCase):
 
     def test_task_to_pipeline(self):
         dd = ibllib.io.extractors.base._get_task_types_json_config()
         types = list(set([dd[k] for k in dd]))
-        # makes sure that for every defined task type there is an acutal pipeline
+        # makes sure that for every defined task type there is an actual pipeline
         for type in types:
             assert ibllib.io.extractors.base._get_pipeline_from_task_type(type)
             print(type, ibllib.io.extractors.base._get_pipeline_from_task_type(type))
@@ -751,5 +795,64 @@ def dummy_function(arg1, arg2):
         self.assertEqual(result, ("test1", "test2"))
 
 
+class TestLegacyDeprecations(unittest.TestCase):
+    """Assert removal of old code."""
+
+    def test_remove_legacy_pipeline(self):
+        """Remove old legacy pipeline code.
+
+        The following code is an incomplete list of modules and functions that should be removed:
+
+        - pipes.ephys_preprocessing
+        - pipes.training_preprocessing
+        - io.extractors.biased_trials.extract_all
+        - io.extractors.bpod_trials.extract_all
+        - io.extractors.base.get_session_extractor_type
+        - io.extractors.base.get_pipeline
+        - io.extractors.base._get_pipeline_from_task_type
+        - io.extractors.base._get_task_types_json_config
+        - io.extractors.extractor_types.json
+        - qc.task_extractors.TaskQCExtractor.extract_data
+
+        NB: some tasks in ephys_preprocessing and maybe training_preprocessing may be directly used
+        or subclassed by the dynamic pipeline. The TaskQCExtractor class could be removed entirely.
+        Instead, a function could exist to simply fetch the relevant data from the task's extractor
+        class.  Alos, there may be plenty of iblscripts CI tests to be removed.
+        """
+        self.assertTrue(datetime.today() < datetime(2024, 9, 1), 'remove legacy pipeline')
+
+    def test_remove_legacy_rig_code(self):
+        """Remove old legacy (v7) rig code.
+
+        The following code is an incomplete list of modules and functions that should be removed:
+
+        - pipes.transfer_rig_data
+        - pipes.misc.check_transfer
+        - pipes.misc.transfer_session_folders
+        - pipes.misc.copy_with_check
+        - pipes.misc.backup_session
+        - pipes.misc.transfer_folder
+        - pipes.misc.load_videopc_params
+        - pipes.misc.load_ephyspc_params
+        - pipes.misc.create_basic_transfer_params
+        - pipes.misc.create_videopc_params
+        - pipes.misc.create_ephyspc_params
+        - pipes.misc.rdiff_install
+        - pipes.misc.rsync_paths
+        - pipes.misc.confirm_ephys_remote_folder
+        - pipes.misc.create_ephys_flags
+        - pipes.misc.create_ephys_transfer_done_flag
+        - pipes.misc.create_video_transfer_done_flag
+        - pipes.misc.create_transfer_done_flag
+
+        pipes.misc.backup_session may be worth keeping and utilized by the iblrig code (arguably
+        useful on both rig and local server). The corresponding tests should also be removed.
+
+        In addition some iblscripts.deploy files should be removed, e.g. prepare_ephys_session,
+        prepare_video_session.
+        """
+        self.assertTrue(datetime.today() < datetime(2024, 10, 1), 'remove legacy rig code')
+
+
 if __name__ == '__main__':
     unittest.main(exit=False, verbosity=2)