Deal the case with same shift.
import numpy as np
-from typing import Union
-from spikeinterface import WaveformExtractor, BaseSorting
+from spikeinterface import WaveformExtractor
from ..postprocessing import get_template_extremum_channel_peak_shift, get_template_amplitudes, align_sorting
@@ -13,8 +12,7 @@ def remove_redundant_units(sorting_or_waveform_extractor,
agreement_threshold=0.2,
duplicate_threshold=0.8,
remove_strategy='minimum_shift',
- peak_sign="neg",
- verbose=False):
+ peak_sign="neg"):
"""
Removes redundant or duplicate units by comparing the sorting output with itself.
@@ -40,6 +38,7 @@ def remove_redundant_units(sorting_or_waveform_extractor,
remove_strategy: str
Which stragtegy to remove one of the two duplicated units:
'minimum_shift': keep the unit with best peak alignment (minimum shift)
+ If shift are equal then the 'highest_amplitude' is used
'highest_amplitude': keep the unit with the best amplitude on un shifted max.
peak_sign: str ('neg', 'pos', 'both')
Used when remove_strategy='highest_amplitude'
@@ -68,22 +67,35 @@ def remove_redundant_units(sorting_or_waveform_extractor,
remove_unit_ids = []
+ if remove_strategy in ('minimum_shift', 'highest_amplitude'):
+ # this is the values at spike index !
+ peak_values = get_template_amplitudes(we, peak_sign=peak_sign, mode="at_index")
+ peak_values = {unit_id: np.max(np.abs(values)) for unit_id, values in peak_values.items()}
+
if remove_strategy == 'minimum_shift':
assert align, 'remove_strategy with minimum_shift need align=True'
for u1, u2 in redundant_unit_pairs:
if np.abs(unit_peak_shifts[u1]) > np.abs(unit_peak_shifts[u2]):
remove_unit_ids.append(u1)
+ elif np.abs(unit_peak_shifts[u1]) < np.abs(unit_peak_shifts[u2]):
+ remove_unit_ids.append(u2)
+ else:
+ # equal shift use peak values
+ if np.abs(peak_values[u1]) < np.abs(peak_values[u2]):
+ remove_unit_ids.append(u1)
else:
remove_unit_ids.append(u2)
elif remove_strategy == 'highest_amplitude':
- peak_values = get_template_amplitudes(we, peak_sign=peak_sign, mode="at_index")
- peak_values = {unit_id: np.max(np.abs(values)) for unit_id, values in peak_values.items()}
for u1, u2 in redundant_unit_pairs:
if np.abs(peak_values[u1]) < np.abs(peak_values[u2]):
remove_unit_ids.append(u1)
else:
remove_unit_ids.append(u2)
- elif remove_strategy == 'max_spikes':
+ elif remove_strategy == 'with_metrics':
+ # TODO
+ # @aurelien @alessio
+ # here we can implement the choice of the best one given an external metrics table
+ # this will be implemented in a futur PR by the first who need it!
raise NotImplementedError()
else:
raise ValueError(f'remove_strategy : {remove_strategy}')
