Coverage for ibllib/pipes/sdsc_tasks.py: 0%

1import numpy as np 

2 

3import spikeglx 

4from ibllib.ephys.sync_probes import apply_sync 

5from ibllib.pipes.tasks import Task 

6 

7 

8class RegisterSpikeSortingSDSC(Task): 

9 

10 @property 

11 def signature(self): 

12 signature = { 

13 'input_files': [('*sync.npy', f'raw_ephys_data/{self.pname}', False), 

14 ('*ap.meta', f'raw_ephys_data/{self.pname}', False)], 

15 'output_files': [] 

16 } 

17 return signature 

18 

19 def __init__(self, session_path, pname=None, revision_label='#test#', **kwargs): 

20 super().__init__(session_path, **kwargs) 

21 

22 self.pname = pname 

23 self.revision_label = revision_label 

24 

25 def _run(self): 

26 

27 out_path = self.session_path.joinpath('alf', self.pname, 'pykilosort', self.revision_label) 

28 

29 def _fs(meta_file): 

30 # gets sampling rate from data 

31 md = spikeglx.read_meta_data(meta_file) 

32 return spikeglx._get_fs_from_meta(md) 

33 

34 sync_file = next(self.session_path.joinpath('raw_ephys_data', self.pname).glob('*sync.npy')) 

35 meta_file = next(self.session_path.joinpath('raw_ephys_data', self.pname).glob('*ap.meta')) 

36 

37 st_file = out_path.joinpath('spikes.times.npy') 

38 spike_samples = np.load(out_path.joinpath('spikes.samples.npy')) 

39 interp_times = apply_sync(sync_file, spike_samples / _fs(meta_file), forward=True) 

40 np.save(st_file, interp_times) 

41 

42 out = list(self.session_path.joinpath('alf', self.pname, 'pykilosort', self.revision_label).glob('*')) 

43 return out