utils/whisper.py

# Copyright (c) 2023 Amphion.
#
# This source code is licensed under the MIT license found in the
# LICENSE file in the root directory of this source tree.

import torch
import os
import pickle
from tqdm import tqdm
import numpy as np

from modules import whisper_extractor as whisper


def whisper_encoder_batch(model, audio_paths):
    batch = len(audio_paths)
    batch_mel = torch.zeros((batch, 80, 3000), dtype=torch.float32, device=model.device)

    for i, audio_path in enumerate(audio_paths):
        # (48000,)
        audio = whisper.load_audio(str(audio_path))
        audio = whisper.pad_or_trim(audio)

        # (80, 3000)
        mel = whisper.log_mel_spectrogram(audio).to(model.device)
        batch_mel[i] = mel

    with torch.no_grad():
        # (batch, 1500, 1024)
        features = model.embed_audio(batch_mel)

    return features.cpu().detach().numpy()


def whisper_encoder(model, audio_path):
    audio = whisper.load_audio(str(audio_path))
    audio = whisper.pad_or_trim(audio)

    # (80, 3000)
    mel = whisper.log_mel_spectrogram(audio).to(model.device).unsqueeze(0)

    with torch.no_grad():
        # (1, 1500, 1024) -> # (1500, 1024)
        features = model.embed_audio(mel).squeeze(0)

    return features.cpu().detach().numpy()


def get_mapped_whisper_features(
    raw_whisper_features, mapping_features, fast_mapping=True
):
    """
    Whisper: frameshift = 20ms (30s audio -> 1500 frames), hop_size = 480 in 24k
    # Ref: https://github.com/openai/whisper/blob/7858aa9c08d98f75575035ecd6481f462d66ca27/whisper/model.py#L136

    Now it's only used for mapping to bigvgan's mels (sr = 24k, hop_size = 256, frameshift ~= 10.7 ms)
    """
    source_hop = 480
    target_hop = 256

    factor = np.gcd(source_hop, target_hop)
    source_hop //= factor
    target_hop //= factor
    print(
        "Mapping source's {} frames => target's {} frames".format(
            target_hop, source_hop
        )
    )

    max_source_len = 1500
    whisper_features = []
    for index, mapping_feat in enumerate(tqdm(mapping_features)):
        # mapping_feat: (mels_frame_len, n_mels)
        target_len = mapping_feat.shape[0]
        # The max target_len is 2812
        target_len = min(target_len, max_source_len * source_hop // target_hop)

        # (1500, dim)
        raw_feats = raw_whisper_features[index]
        width = raw_feats.shape[-1]

        if fast_mapping:
            source_len = target_len * target_hop // source_hop + 1
            raw_feats = raw_feats[:source_len]
        else:
            source_len = max_source_len

        # const ~= target_len * target_hop
        const = source_len * source_hop // target_hop * target_hop

        # (source_len * source_hop, dim)
        up_sampling_feats = np.repeat(raw_feats, source_hop, axis=0)
        # (const, dim) -> (const/target_hop, target_hop, dim) -> (const/target_hop, dim)
        down_sampling_feats = np.average(
            up_sampling_feats[:const].reshape(-1, target_hop, width), axis=1
        )
        assert len(down_sampling_feats) >= target_len

        # (target_len, dim)
        feats = down_sampling_feats[:target_len]
        whisper_features.append(feats)

    return whisper_features


def load_whisper_model(hps):
    print("Loading Whisper Model: ", hps.whisper_model)
    model = whisper.load_model(hps.whisper_model)
    if torch.cuda.is_available():
        model = model.cuda()

    model = model.eval()
    return model


def load_target_acoustic_features(
    output_path, dataset, acoustic_features_name, acoustic_features_fs, dataset_type
):
    mapping_dir = os.path.join(
        output_path,
        dataset,
        "{}/{}".format(acoustic_features_name, acoustic_features_fs),
    )
    with open(os.path.join(mapping_dir, "{}.pkl".format(dataset_type)), "rb") as f:
        mapping_features = pickle.load(f)

    # Mels: (n_mels, frame_len) -> (frame_len, n_mels)
    if acoustic_features_name == "mels":
        print("Transposing mel features...")
        mapping_features = [feat.T for feat in mapping_features]

    print(
        "Mapping to the acoustic features {}, #sz = {}, feats[0] is {}".format(
            acoustic_features_name, len(mapping_features), mapping_features[0].shape
        )
    )
    return mapping_features


def extract_whisper_features_of_dataset(
    datasets,
    model,
    batch_size,
    out_dir,
):
    audio_paths = [utt["Path"] for utt in datasets]
    if len(audio_paths) < batch_size:
        batch_size = len(audio_paths)

    start, end = 0, 0
    while end < len(audio_paths):
        # Raw features: (batch_size, 1500, dim)
        start = end
        end = start + batch_size
        tmp_raw_whisper_features = whisper_encoder_batch(model, audio_paths[start:end])

        # Mapping to acoustic features' lengths
        for index, utt in enumerate(tqdm(datasets[start:end])):
            uid = utt["Uid"]
            raw_whisper_feature = tmp_raw_whisper_features[index]

            save_path = os.path.join(out_dir, uid + ".npy")
            np.save(save_path, raw_whisper_feature)

        print("{}/{} Done...".format(end, len(audio_paths)))