import logging
from ibllib.pipes import base_tasks
from ibllib.io import ffmpeg
from ibllib.io.extractors import training_audio
_logger = logging.getLogger('ibllib')
[docs]
class AudioCompress(base_tasks.AudioTask):
cpu = 2
priority = 10
job_size = 'small'
@property
def signature(self):
signature = {
'input_files': [('_iblrig_micData.raw.wav', self.device_collection, True)],
'output_files': [('_iblrig_micData.raw.flac', self.device_collection, True)]
}
return signature
def _run(self, overwrite=False):
command = "ffmpeg -i {file_in} -y -nostdin -c:a flac -nostats {file_out}"
file_in = next(self.session_path.joinpath(self.device_collection).rglob("_iblrig_micData.raw.wav"), None)
if file_in is None:
return
file_out = file_in.with_suffix(".flac")
status, output_file = ffmpeg.compress(file_in=file_in, file_out=file_out, command=command)
return [output_file]
[docs]
class AudioSync(base_tasks.AudioTask):
"""
Extracts audio events and sync. N.B currently only supports bpod with xonar sound system
"""
cpu = 2
priority = 10
job_size = 'small'
def __init__(self, session_path, **kwargs):
super().__init__(session_path, **kwargs)
# Task collection (this needs to be specified in the task kwargs)
self.collection = self.get_task_collection(kwargs.get('collection', None))
@property
def signature(self):
signature = {
'input_files': [('_iblrig_micData.raw.wav', self.device_collection, True)],
'output_files': [('_iblmic_audioOnsetGoCue.times_mic.npy', self.device_collection, True),
('_iblmic_audioSpectrogram.frequencies.npy', self.device_collection, True),
('_iblmic_audioSpectrogram.power.npy', self.device_collection, True),
('_iblmic_audioSpectrogram.times_mic.npy', self.device_collection, True)]
}
return signature
def _run(self):
if self.sync == 'bpod':
return training_audio.extract_sound(self.session_path, task_collection=self.collection,
device_collection=self.device_collection, save=True, delete=True)
else:
_logger.warning('Audio Syncing not yet implemented for FPGA')
return