Built architecture
Still need to implement get_traces()
+from typing import List, Union
import numpy as np
-from spikeinterface.core import BaseRecording, BaseRecordingSegment, NpzSortingExtractor
+from spikeinterface.core import BaseRecording, BaseRecordingSegment, BaseSorting, NpzSortingExtractor
from spikeinterface.extractors import synthesize_random_firings
class HybridUnitsRecording(BaseRecording):
- def __init__(self, target_recording: BaseRecording, templates: np.ndarray, n_before=None, sorting=None,
- frequency: float = 10, amplitude_std: float = 0.0, refrac_period: float = 2.0):
+ def __init__(self, target_recording: BaseRecording, templates: np.ndarray, n_before: Union[int, None] = None,
+ sorting: Union[BaseSorting, None] = None, frequency: float = 10, amplitude_std: float = 0.0,
+ refrac_period: float = 2.0):
"""
TODO
"""
+ # Propagate information from the target recording.
+ BaseRecording.__init__(self, target_recording.sampling_frequency, target_recording.channel_ids, target_recording.dtype)
+ target_recording.copy_metadata(self)
+
assert target_recording.get_num_segments() == 1 # For now, make things simple with 1 segment.
recording_segment = HybridUnitsRecordingSegment(target_recording._recording_segments[0], templates,
@@ -20,12 +26,15 @@ class HybridUnitsRecording(BaseRecording):
class HybridUnitsRecordingSegment(BaseRecordingSegment):
- def __init__(self, target_recording: BaseRecordingSegment, templates: np.ndarray, n_before=None, sorting=None,
- frequency: float = 10, amplitude_std: float = 0.0, refrac_period: float = 2.0):
+ def __init__(self, target_recording: BaseRecordingSegment, templates: np.ndarray, n_before: Union[int, None] = None,
+ sorting: Union[BaseSorting, None] = None, frequency: float = 10, amplitude_std: float = 0.0,
+ refrac_period: float = 2.0):
"""
TODO
"""
+ self.parent_recording = target_recording
+ self.templates = templates
n_units = len(templates)
if sorting is None: # TODO: Move out of toy_example for better spike train generation.
@@ -37,10 +46,17 @@ class HybridUnitsRecordingSegment(BaseRecordingSegment):
npz_file = "TODO" # np.savez(file, **spike_trains)
sorting = NpzSortingExtractor(npz_file)
+ self.sorting = sorting
- self.amplitude_factor = {unit_id: np.random.normal(loc=1.0, scale=amplitude_std, size=len(sorting.get_unit_spike_train(unit_id)))
+ amplitudes = {unit_id: np.random.normal(loc=1.0, scale=amplitude_std, size=len(sorting.get_unit_spike_train(unit_id)))
for unit_id in range(n_units)}
+ self.amplitude_factor = np.where(amplitudes < 0, 0, amplitudes)
+
+
+ def get_traces(self, start_frame: Union[int, None] = None, end_frame: Union[int, None] = None,
+ channel_indices: Union[List, None] = None) -> np.ndarray:
+ traces = self.parent_recording.get_traces(start_frame, end_frame)
+ # Do stuff.
- def get_traces():
- pass
+ return traces
