accommodate different ephys modes (chronic, no-curation, etc.)

tests/__init__.py

@@ -13,7 +13,7 @@
 
 # ------------------- SOME CONSTANTS -------------------
 
-_tear_down = True
+_tear_down = False
 
 test_user_data_dir = pathlib.Path('./tests/user_data')
 test_user_data_dir.mkdir(exist_ok=True)
@@ -84,7 +84,8 @@ def pipeline():
            'ephys': pipeline.ephys,
            'probe': pipeline.probe,
            'session': pipeline.session,
-           'get_ephys_root_data_dir': pipeline.get_ephys_root_data_dir}
+           'get_ephys_root_data_dir': pipeline.get_ephys_root_data_dir,
+           'ephys_mode': pipeline.ephys_mode}
 
     if _tear_down:
         pipeline.subject.Subject.delete()
@@ -196,7 +197,10 @@ def kilosort_paramset(pipeline):
 
     # doing the insert here as well, since most of the test will require this paramset inserted
     ephys.ClusteringParamSet.insert_new_params(
-        'kilosort2', 0, 'Spike sorting using Kilosort2', params_ks)
+        processing_method='kilosort2',
+        paramset_desc='Spike sorting using Kilosort2',
+        params=params_ks,
+        paramset_idx=0)
 
     yield params_ks
 
@@ -230,6 +234,7 @@ def clustering_tasks(pipeline, kilosort_paramset, ephys_recordings):
         kilosort_dir = next(recording_dir.rglob('spike_times.npy')).parent
         ephys.ClusteringTask.insert1({**ephys_rec_key,
                                       'paramset_idx': 0,
+                                      'task_mode': 'load',
                                       'clustering_output_dir': kilosort_dir.as_posix()},
                                      skip_duplicates=True)
 
@@ -253,12 +258,17 @@ def clustering(clustering_tasks, pipeline):
 
 @pytest.fixture
 def curations(clustering, pipeline):
-    ephys = pipeline['ephys']
+    ephys_mode = pipeline['ephys_mode']
 
-    for key in (ephys.ClusteringTask - ephys.Curation).fetch('KEY'):
-        ephys.Curation().create1_from_clustering_task(key)
+    if ephys_mode == 'no-curation':
+        return
+    else:
+        ephys = pipeline['ephys']
 
-    yield
+        for key in (ephys.ClusteringTask - ephys.Curation).fetch('KEY'):
+            ephys.Curation().create1_from_clustering_task(key)
 
-    if _tear_down:
-        ephys.Curation.delete()
+        yield
+
+        if _tear_down:
+            ephys.Curation.delete()

tests/test_populate.py

@@ -80,19 +80,19 @@ def test_curated_clustering_populate(curations, pipeline, testdata_paths):
     ephys = pipeline['ephys']
 
     rel_path = testdata_paths['npx3A-p1-ks']
-    curation_key = (ephys.Curation & f'curation_output_dir LIKE "%{rel_path}"').fetch1('KEY')
+    curation_key = _get_curation_key(rel_path, pipeline)
     ephys.CuratedClustering.populate(curation_key)
     assert len(ephys.CuratedClustering.Unit & curation_key
                & 'cluster_quality_label = "good"') == 76
 
     rel_path = testdata_paths['oe_npx3B-ks']
-    curation_key = (ephys.Curation & f'curation_output_dir LIKE "%{rel_path}"').fetch1('KEY')
+    curation_key = _get_curation_key(rel_path, pipeline)
     ephys.CuratedClustering.populate(curation_key)
     assert len(ephys.CuratedClustering.Unit & curation_key
                & 'cluster_quality_label = "good"') == 68
 
     rel_path = testdata_paths['npx3B-p1-ks']
-    curation_key = (ephys.Curation & f'curation_output_dir LIKE "%{rel_path}"').fetch1('KEY')
+    curation_key = _get_curation_key(rel_path, pipeline)
     ephys.CuratedClustering.populate(curation_key)
     assert len(ephys.CuratedClustering.Unit & curation_key
                & 'cluster_quality_label = "good"') == 55
@@ -102,7 +102,7 @@ def test_waveform_populate_npx3B_OpenEphys(curations, pipeline, testdata_paths):
     ephys = pipeline['ephys']
 
     rel_path = testdata_paths['oe_npx3B-ks']
-    curation_key = (ephys.Curation & f'curation_output_dir LIKE "%{rel_path}"').fetch1('KEY')
+    curation_key = _get_curation_key(rel_path, pipeline)
     ephys.CuratedClustering.populate(curation_key)
     ephys.WaveformSet.populate(curation_key)
 
@@ -116,11 +116,27 @@ def test_waveform_populate_npx3B_SpikeGLX(curations, pipeline, testdata_paths):
     ephys = pipeline['ephys']
 
     rel_path = testdata_paths['npx3B-p1-ks']
-    curation_key = (ephys.Curation & f'curation_output_dir LIKE "%{rel_path}"').fetch1('KEY')
+    curation_key = _get_curation_key(rel_path, pipeline)
     ephys.CuratedClustering.populate(curation_key)
     ephys.WaveformSet.populate(curation_key)
 
     waveforms = np.vstack((ephys.WaveformSet.PeakWaveform & curation_key).fetch(
         'peak_electrode_waveform'))
 
     assert waveforms.shape == (150, 64)
+
+
+# ---- HELPER FUNCTIONS ----
+
+def _get_curation_key(output_relative_path, pipeline):
+    ephys = pipeline['ephys']
+    ephys_mode = pipeline['ephys_mode']
+
+    if ephys_mode == 'no-curation':
+        EphysCuration = ephys.Clustering
+        output_dir_attr_name = 'clustering_output_dir'
+    else:
+        EphysCuration = ephys.Curation
+        output_dir_attr_name = 'curation_output_dir'
+
+    return (EphysCuration & f'{output_dir_attr_name} LIKE "%{output_relative_path}"').fetch1('KEY')

workflow_array_ephys/ingest.py

@@ -4,6 +4,7 @@
 
 from workflow_array_ephys.pipeline import subject, ephys, probe, session
 from workflow_array_ephys.paths import get_ephys_root_data_dir
+from workflow_array_ephys.pipeline import ephys_mode
 
 from element_array_ephys.readers import spikeglx, openephys
 
@@ -72,15 +73,22 @@ def ingest_sessions(session_csv_path='./user_data/sessions.csv'):
             session_dir_list.append({**session_key, 'session_dir': sess_dir.relative_to(root_data_dir).as_posix()})
             probe_insertion_list.extend([{**session_key, **insertion} for insertion in insertions])
 
-    print(f'\n---- Insert {len(session_list)} entry(s) into session.Session ----')
-    session.Session.insert(session_list)
-    session.SessionDirectory.insert(session_dir_list)
-
     print(f'\n---- Insert {len(probe_list)} entry(s) into probe.Probe ----')
     probe.Probe.insert(probe_list)
 
-    print(f'\n---- Insert {len(probe_insertion_list)} entry(s) into ephys.ProbeInsertion ----')
-    ephys.ProbeInsertion.insert(probe_insertion_list)
+    if ephys_mode == 'chronic':
+        print(f'\n---- Insert {len(probe_insertion_list)} entry(s) into ephys.ProbeInsertion ----')
+        ephys.ProbeInsertion.insert(probe_insertion_list,
+                                    ignore_extra_fields=True, skip_duplicates=True)
+        print(f'\n---- Insert {len(session_list)} entry(s) into session.Session ----')
+        session.Session.insert(session_list)
+        session.SessionDirectory.insert(session_dir_list)
+    else:
+        print(f'\n---- Insert {len(session_list)} entry(s) into session.Session ----')
+        session.Session.insert(session_list)
+        session.SessionDirectory.insert(session_dir_list)
+        print(f'\n---- Insert {len(probe_insertion_list)} entry(s) into ephys.ProbeInsertion ----')
+        ephys.ProbeInsertion.insert(probe_insertion_list)
 
     print('\n---- Successfully completed workflow_array_ephys/ingest.py ----')
 

workflow_array_ephys/pipeline.py

@@ -1,8 +1,9 @@
 import datajoint as dj
+import os
 from element_animal import subject
 from element_lab import lab
 from element_session import session
-from element_array_ephys import probe, ephys
+from element_array_ephys import probe
 
 from element_animal.subject import Subject
 from element_lab.lab import Source, Lab, Protocol, User, Project
@@ -15,6 +16,17 @@
 
 db_prefix = dj.config['custom'].get('database.prefix', '')
 
+# ------------- Import the configured "ephys mode" -------------
+ephys_mode = os.getenv('EPHYS_MODE',
+                       dj.config['custom'].get('ephys_mode', 'acute'))
+if ephys_mode == 'acute':
+    from element_array_ephys import ephys
+elif ephys_mode == 'chronic':
+    from element_array_ephys import ephys_chronic as ephys
+elif ephys_mode == 'no-curation':
+    from element_array_ephys import ephys_no_curation as ephys
+else:
+    raise ValueError(f'Unknown ephys mode: {ephys_mode}')
 
 # ------------- Activate "lab", "subject", "session" schema -------------