Coverage for ibllib/pipes/audio_tasks.py: 97%

1import logging (empty)

2 

3from ibllib.pipes import base_tasks (empty)

4from ibllib.io import ffmpeg (empty)

5from ibllib.io.extractors import training_audio (empty)

6 

7_logger = logging.getLogger('ibllib') (empty)

8 

9 

10class AudioCompress(base_tasks.AudioTask): (empty)

11 

12 cpu = 2 (empty)

13 priority = 10 (empty)

14 job_size = 'small' (empty)

15 

16 @property (empty)

17 def signature(self): (empty)

18 signature = { 1 ctx1b

19 'input_files': [('_iblrig_micData.raw.wav', self.device_collection, True)], 

20 'output_files': [('_iblrig_micData.raw.flac', self.device_collection, True)] 

21 } 

22 return signature 1 ctx1b

23 

24 def _run(self, overwrite=False): (empty)

25 

26 command = "ffmpeg -i {file_in} -y -nostdin -c:a flac -nostats {file_out}" 1 ctx1b

27 file_in = next(self.session_path.joinpath(self.device_collection).rglob("_iblrig_micData.raw.wav"), None) 1 ctx1b

28 if file_in is None: 1 ctx1b

29 return 

30 file_out = file_in.with_suffix(".flac") 1 ctx1b

31 status, output_file = ffmpeg.compress(file_in=file_in, file_out=file_out, command=command) 1 ctx1b

32 return [output_file] 1 ctx1b

33 

34 

35class AudioSync(base_tasks.AudioTask): (empty)

36 """ 

37 Extracts audio events and sync. N.B currently only supports bpod with xonar sound system 

38 """ 

39 

40 cpu = 2 (empty)

41 priority = 10 (empty)

42 job_size = 'small' (empty)

43 

44 def __init__(self, session_path, **kwargs): (empty)

45 super().__init__(session_path, **kwargs) 8 ctx1fdceghij

46 # Task collection (this needs to be specified in the task kwargs) 

47 self.collection = self.get_task_collection(kwargs.get('collection', None)) 8 ctx1fdceghij

48 

49 @property (empty)

50 def signature(self): (empty)

51 signature = { 3 ctx1dce

52 'input_files': [('_iblrig_micData.raw.wav', self.device_collection, True)], 

53 'output_files': [('_iblmic_audioOnsetGoCue.times_mic.npy', self.device_collection, True), 

54 ('_iblmic_audioSpectrogram.frequencies.npy', self.device_collection, True), 

55 ('_iblmic_audioSpectrogram.power.npy', self.device_collection, True), 

56 ('_iblmic_audioSpectrogram.times_mic.npy', self.device_collection, True)] 

57 } 

58 return signature 3 ctx1dce

59 

60 def _run(self): (empty)

61 if self.sync == 'bpod': 3 ctx1dce

62 return training_audio.extract_sound(self.session_path, task_collection=self.collection, 2 ctx1de

63 device_collection=self.device_collection, save=True, delete=True) 

64 else: 

65 _logger.warning('Audio Syncing not yet implemented for FPGA') 1 ctx1c

66 return 1 ctx1c