Merge branch 'datajoint-spikeinterface' into dev_new_main

Author: ttngu207

element_array_ephys/ephys.py

@@ -1033,6 +1033,16 @@ def make(self, key):
 
         if si_sorting_analyzer_dir.exists():  # Read from spikeinterface outputs
             import spikeinterface as si
+            from spikeinterface import sorters
+
+            sorting_file = output_dir / sorter_name / "spike_sorting" / "si_sorting.pkl"
+            si_sorting_: si.sorters.BaseSorter = si.load_extractor(
+                sorting_file, base_folder=output_dir
+            )
+            if si_sorting_.unit_ids.size == 0:
+                logger.info(f"No units found in {sorting_file}. Skipping Unit ingestion...")
+                self.insert1(key)
+                return
 
             sorting_analyzer = si.load_sorting_analyzer(folder=si_sorting_analyzer_dir)
             si_sorting = sorting_analyzer.sorting
@@ -1241,6 +1251,11 @@ def make(self, key):
         output_dir = find_full_path(get_ephys_root_data_dir(), output_dir)
         sorter_name = clustering_method.replace(".", "_")
 
+        self.insert1(key)
+        if not len(CuratedClustering.Unit & key):
+            logger.info(f"No CuratedClustering.Unit found for {key}, skipping Waveform ingestion.")
+            return
+
         # Get channel and electrode-site mapping
         electrode_query = (EphysRecording.Channel & key).proj(..., "-channel_name")
         channel2electrode_map: dict[int, dict] = {
@@ -1294,7 +1309,6 @@ def yield_unit_waveforms():
                     ]
 
                     yield unit_peak_waveform, unit_electrode_waveforms
-
         else:  # read from kilosort outputs (ecephys pipeline)
             kilosort_dataset = kilosort.Kilosort(output_dir)
 
@@ -1394,7 +1408,6 @@ def yield_unit_waveforms():
                         yield unit_peak_waveform, unit_electrode_waveforms
 
         # insert waveform on a per-unit basis to mitigate potential memory issue
-        self.insert1(key)
         for unit_peak_waveform, unit_electrode_waveforms in yield_unit_waveforms():
             if unit_peak_waveform:
                 self.PeakWaveform.insert1(unit_peak_waveform, ignore_extra_fields=True)
@@ -1501,6 +1514,11 @@ def make(self, key):
         output_dir = find_full_path(get_ephys_root_data_dir(), output_dir)
         sorter_name = clustering_method.replace(".", "_")
 
+        self.insert1(key)
+        if not len(CuratedClustering.Unit & key):
+            logger.info(f"No CuratedClustering.Unit found for {key}, skipping QualityMetrics ingestion.")
+            return
+
         si_sorting_analyzer_dir = output_dir / sorter_name / "sorting_analyzer"
         if si_sorting_analyzer_dir.exists():  # read from spikeinterface outputs
             import spikeinterface as si
@@ -1556,7 +1574,6 @@ def make(self, key):
             for unit_key in (CuratedClustering.Unit & key).fetch("KEY")
         ]
 
-        self.insert1(key)
         self.Cluster.insert(metrics_list, ignore_extra_fields=True)
         self.Waveform.insert(metrics_list, ignore_extra_fields=True)
 

element_array_ephys/spike_sorting/si_spike_sorting.py

@@ -281,11 +281,18 @@ def make(self, key):
 
         analyzer_output_dir = output_dir / sorter_name / "sorting_analyzer"
 
+        has_units = si_sorting.unit_ids.size > 0
+
         @memoized_result(
             uniqueness_dict=postprocessing_params,
             output_directory=analyzer_output_dir,
         )
         def _sorting_analyzer_compute():
+            if not has_units:
+                log.info("No units found in sorting object. Skipping sorting analyzer.")
+                analyzer_output_dir.mkdir(parents=True, exist_ok=True)  # create empty directory anyway, for consistency
+                return
+
             # Sorting Analyzer
             sorting_analyzer = si.create_sorting_analyzer(
                 sorting=si_sorting,
@@ -309,6 +316,8 @@ def _sorting_analyzer_compute():
 
         _sorting_analyzer_compute()
 
+        do_si_export = postprocessing_params.get("export_to_phy", False) or postprocessing_params.get("export_report", False)
+
         self.insert1(
             {
                 **key,
@@ -317,8 +326,7 @@ def _sorting_analyzer_compute():
                     datetime.utcnow() - execution_time
                 ).total_seconds()
                 / 3600,
-                "do_si_export": postprocessing_params.get("export_to_phy", False)
-                or postprocessing_params.get("export_report", False),
+                "do_si_export": do_si_export and has_units,
             }
         )