Spaces:
Running
on
A10G
Running
on
A10G
import cv2 | |
import numpy as np | |
import torch | |
# import torchaudio | |
from torchvision import transforms | |
from transformers import ProcessorMixin, BatchEncoding | |
from transformers.image_processing_utils import BatchFeature | |
from torch.nn import functional as F | |
def make_list_of_images(x): | |
if not isinstance(x, list): | |
return [x] | |
return x | |
#torchaudio.set_audio_backend("soundfile") | |
def torchaudio_loader(path): | |
return torchaudio.load(path) | |
def int16_to_float32_torch(x): | |
return (x / 32767.0).type(torch.float32) | |
def float32_to_int16_torch(x): | |
x = torch.clamp(x, min=-1., max=1.) | |
return (x * 32767.).type(torch.int16) | |
DEFAULT_AUDIO_FRAME_SHIFT_MS = 10 | |
class AudioTransform: | |
def __init__(self, config): | |
self.sample_rate = config.audio_sample_rate | |
self.num_mel_bins = config.num_mel_bins | |
self.target_length = config.target_length | |
self.audio_mean = config.audio_mean | |
self.audio_std = config.audio_std | |
# mean=-4.2677393 | |
# std=4.5689974 | |
self.norm = transforms.Normalize(mean=self.audio_mean, std=self.audio_std) | |
def __call__(self, audio_data_and_origin_sr): | |
audio_data, origin_sr = audio_data_and_origin_sr | |
if self.sample_rate != origin_sr: | |
# print(audio_data.shape, origin_sr) | |
audio_data = torchaudio.functional.resample(audio_data, orig_freq=origin_sr, new_freq=self.sample_rate) | |
waveform_melspec = self.waveform2melspec(audio_data[0]) | |
return self.norm(waveform_melspec) | |
def waveform2melspec(self, audio_data): | |
max_len = self.target_length * self.sample_rate // 100 | |
if audio_data.shape[-1] > max_len: | |
mel = self.get_mel(audio_data) | |
# split to three parts | |
chunk_frames = self.target_length | |
total_frames = mel.shape[0] | |
ranges = np.array_split(list(range(0, total_frames - chunk_frames + 1)), 3) | |
# print('total_frames-chunk_frames:', total_frames-chunk_frames, | |
# 'len(audio_data):', len(audio_data), | |
# 'chunk_frames:', chunk_frames, | |
# 'total_frames:', total_frames) | |
if len(ranges[1]) == 0: # if the audio is too short, we just use the first chunk | |
ranges[1] = [0] | |
if len(ranges[2]) == 0: # if the audio is too short, we just use the first chunk | |
ranges[2] = [0] | |
# randomly choose index for each part | |
# idx_front = np.random.choice(ranges[0]) | |
# idx_middle = np.random.choice(ranges[1]) | |
# idx_back = np.random.choice(ranges[2]) | |
idx_front = ranges[0][0] # fixed | |
idx_middle = ranges[1][0] | |
idx_back = ranges[2][0] | |
# select mel | |
mel_chunk_front = mel[idx_front:idx_front + chunk_frames, :] | |
mel_chunk_middle = mel[idx_middle:idx_middle + chunk_frames, :] | |
mel_chunk_back = mel[idx_back:idx_back + chunk_frames, :] | |
# stack | |
mel_fusion = torch.stack([mel_chunk_front, mel_chunk_middle, mel_chunk_back], dim=0) | |
elif audio_data.shape[-1] < max_len: # padding if too short | |
n_repeat = int(max_len / len(audio_data)) | |
audio_data = audio_data.repeat(n_repeat) | |
audio_data = F.pad( | |
audio_data, | |
(0, max_len - len(audio_data)), | |
mode="constant", | |
value=0, | |
) | |
mel = self.get_mel(audio_data) | |
mel_fusion = torch.stack([mel, mel, mel], dim=0) | |
else: # if equal | |
mel = self.get_mel(audio_data) | |
mel_fusion = torch.stack([mel, mel, mel], dim=0) | |
# twice check | |
p = self.target_length - mel_fusion.shape[1] | |
# if abs(p) / self.target_length > 0.2: | |
# logging.warning( | |
# "Large gap between audio n_frames(%d) and " | |
# "target_length (%d). Is the audio_target_length " | |
# "setting correct?", | |
# mel_fusion.shape[1], | |
# self.target_length, | |
# ) | |
# cut and pad | |
if p > 0: | |
m = torch.nn.ZeroPad2d((0, 0, 0, p)) | |
mel_fusion = m(mel_fusion) | |
elif p < 0: | |
mel_fusion = mel_fusion[:, 0: self.target_length, :] | |
mel_fusion = mel_fusion.transpose(1, 2) # [3, target_length, mel_bins] -> [3, mel_bins, target_length] | |
return mel_fusion | |
def get_mel(self, audio_data): | |
# mel shape: (n_mels, T) | |
audio_data -= audio_data.mean() | |
mel = torchaudio.compliance.kaldi.fbank( | |
audio_data.unsqueeze(0), | |
htk_compat=True, | |
sample_frequency=self.sample_rate, | |
use_energy=False, | |
window_type="hanning", | |
num_mel_bins=self.num_mel_bins, | |
dither=0.0, | |
frame_length=25, | |
frame_shift=DEFAULT_AUDIO_FRAME_SHIFT_MS, | |
) | |
return mel # (T, n_mels) | |
def get_audio_transform(config): | |
config = config.vision_config | |
return AudioTransform(config) | |
def load_and_transform_audio( | |
audio_path, | |
transform, | |
): | |
waveform_and_sr = torchaudio_loader(audio_path) | |
audio_outputs = transform(waveform_and_sr) | |
return audio_outputs | |
class LanguageBindAudioProcessor(ProcessorMixin): | |
attributes = [] | |
tokenizer_class = ("LanguageBindAudioTokenizer") | |
def __init__(self, config, tokenizer=None, **kwargs): | |
super().__init__(**kwargs) | |
self.config = config | |
self.transform = get_audio_transform(config) | |
self.image_processor = load_and_transform_audio | |
self.tokenizer = tokenizer | |
def __call__(self, images=None, text=None, context_length=77, return_tensors=None, **kwargs): | |
if text is None and images is None: | |
raise ValueError("You have to specify either text or images. Both cannot be none.") | |
if text is not None: | |
encoding = self.tokenizer(text, max_length=context_length, padding='max_length', | |
truncation=True, return_tensors=return_tensors, **kwargs) | |
if images is not None: | |
images = make_list_of_images(images) | |
image_features = [self.image_processor(image, self.transform) for image in images] | |
image_features = torch.stack(image_features) | |
if text is not None and images is not None: | |
encoding["pixel_values"] = image_features | |
return encoding | |
elif text is not None: | |
return encoding | |
else: | |
return {"pixel_values": image_features} | |
def batch_decode(self, skip_special_tokens=True, *args, **kwargs): | |
""" | |
This method forwards all its arguments to CLIPTokenizerFast's [`~PreTrainedTokenizer.batch_decode`]. Please | |
refer to the docstring of this method for more information. | |
""" | |
return self.tokenizer.batch_decode(*args, skip_special_tokens=skip_special_tokens, **kwargs) | |
def decode(self, skip_special_tokens=True, *args, **kwargs): | |
""" | |
This method forwards all its arguments to CLIPTokenizerFast's [`~PreTrainedTokenizer.decode`]. Please refer to | |
the docstring of this method for more information. | |
""" | |
return self.tokenizer.decode(*args, skip_special_tokens=skip_special_tokens, **kwargs) | |