stream_app.py

import streamlit as st
import time

# record the audio imports
import sounddevice as sd
from scipy.io.wavfile import write
import queue
import sys
import sounddevice as sd
import soundfile as sf
import numpy  # Make sure NumPy is loaded before it is used in the callback

# transcribe the audio imports
import whisper

import matplotlib.pyplot as plt
import numpy as np
import wave

#StyleTTS2 imports
import torch
torch.manual_seed(0)
torch.backends.cudnn.benchmark = False
torch.backends.cudnn.deterministic = True

import random
random.seed(0)
np.random.seed(0)

# load packages
import yaml
from munch import Munch
import torch
from torch import nn
import torch.nn.functional as F
import torchaudio
import librosa
from nltk.tokenize import word_tokenize
from models import *
from utils import *
from text_utils import TextCleaner
textclenaer = TextCleaner()





st.title(" 🎈North-American Accent Training App")
st.caption("How would you sound like if you were English native speaker?")

sample_rate_value=24000
# Audio Path
audio_path = "recorded_audio.wav"
# Global queue to insert audio into
audio_queue = queue.Queue()

def callback(indata, frames, time, status):
    """This is called (from a separate thread) for each audio block."""
    if status:
        print(status, file=sys.stderr)
    audio_queue.put(indata.copy( ))

# Keep the status of start/stop toggling
if 'record_button_status' not in st.session_state:
    st.session_state.record_button_status = False
    
# Toggle start/stop recording
def click_record_button():
    st.session_state.record_button_status = not st.session_state.record_button_status


st.divider()

col1, col2, c3, c4 = st.columns([1,1,1,1]) # dirty hack to put buttons closer
with col1:
    record_button = st.button(":black_circle_for_record: Start Recording", 
        on_click=click_record_button, key="rec_button", disabled=st.session_state.get("record_button_status", False))
with col2:
    stop_button = st.button(":black_square_for_stop: Stop Recording", 
        on_click=click_record_button, key="stop_button", disabled=not st.session_state.get("record_button_status", True))


if record_button and st.session_state.record_button_status:
    with st.spinner('Recording...'):
        # Make sure the file is opened before recording anything:
        with sf.SoundFile(audio_path, mode='w', samplerate=sample_rate_value, 
        channels=1) as file:
            with sd.InputStream(samplerate=sample_rate_value, channels=1, callback=callback):
                while True:
                    if st.session_state.record_button_status:
                        # get audio from the queue and save it to the file
                        file.write(audio_queue.get())
                        
                    else:
                        break
    st.success('Finished recording!')


#########  Transcribe Audio #########
whisper_model = whisper.load_model("base")
result = whisper_model.transcribe(audio_path)
transcribed_text = result["text"]
st.write("You said: " + transcribed_text + "\n")


audio_file = open(audio_path, 'rb')
audio_bytes = audio_file.read()
st.audio(audio_bytes, format='audio/wav')









to_mel = torchaudio.transforms.MelSpectrogram(
    n_mels=80, n_fft=2048, win_length=1200, hop_length=300)
mean, std = -4, 4

def length_to_mask(lengths):
    mask = torch.arange(lengths.max()).unsqueeze(0).expand(lengths.shape[0], -1).type_as(lengths)
    mask = torch.gt(mask+1, lengths.unsqueeze(1))
    return mask

def preprocess(wave):
    wave_tensor = torch.from_numpy(wave).float()
    mel_tensor = to_mel(wave_tensor)
    mel_tensor = (torch.log(1e-5 + mel_tensor.unsqueeze(0)) - mean) / std
    return mel_tensor

def compute_style(path):
    wave, sr = librosa.load(path, sr=24000)
    audio, index = librosa.effects.trim(wave, top_db=30)
    if sr != 24000:
        audio = librosa.resample(audio, sr, 24000)
    mel_tensor = preprocess(audio).to(device)

    with torch.no_grad():
        ref_s = model.style_encoder(mel_tensor.unsqueeze(1))
        ref_p = model.predictor_encoder(mel_tensor.unsqueeze(1))

    return torch.cat([ref_s, ref_p], dim=1)

device = 'cuda' if torch.cuda.is_available() else 'cpu'

# load phonemizer
import phonemizer
global_phonemizer = phonemizer.backend.EspeakBackend(language='en-us', preserve_punctuation=True,  with_stress=True)

config = yaml.safe_load(open("Models/LibriTTS/config.yml"))

# load pretrained ASR model
ASR_config = config.get('ASR_config', False)
ASR_path = config.get('ASR_path', False)
text_aligner = load_ASR_models(ASR_path, ASR_config)

# load pretrained F0 model
F0_path = config.get('F0_path', False)
pitch_extractor = load_F0_models(F0_path)

# load BERT model
from Utils.PLBERT.util import load_plbert
BERT_path = config.get('PLBERT_dir', False)
plbert = load_plbert(BERT_path)

model_params = recursive_munch(config['model_params'])
model = build_model(model_params, text_aligner, pitch_extractor, plbert)
_ = [model[key].eval() for key in model]
_ = [model[key].to(device) for key in model]

params_whole = torch.load("Models/LibriTTS/epochs_2nd_00020.pth", map_location='cpu')
params = params_whole['net']


for key in model:
    if key in params:
        print('%s loaded' % key)
        try:
            model[key].load_state_dict(params[key])
        except:
            from collections import OrderedDict
            state_dict = params[key]
            new_state_dict = OrderedDict()
            for k, v in state_dict.items():
                name = k[7:] # remove `module.`
                new_state_dict[name] = v
            # load params
            model[key].load_state_dict(new_state_dict, strict=False)
#             except:
#                 _load(params[key], model[key])
_ = [model[key].eval() for key in model]

from Modules.diffusion.sampler import DiffusionSampler, ADPM2Sampler, KarrasSchedule

sampler = DiffusionSampler(
    model.diffusion.diffusion,
    sampler=ADPM2Sampler(),
    sigma_schedule=KarrasSchedule(sigma_min=0.0001, sigma_max=3.0, rho=9.0), # empirical parameters
    clamp=False
)

def inference(text, ref_s, alpha = 0.3, beta = 0.7, diffusion_steps=5, embedding_scale=1):
    text = text.strip()
    ps = global_phonemizer.phonemize([text])
    ps = word_tokenize(ps[0])
    ps = ' '.join(ps)
    tokens = textclenaer(ps)
    tokens.insert(0, 0)
    tokens = torch.LongTensor(tokens).to(device).unsqueeze(0)
    
    with torch.no_grad():
        input_lengths = torch.LongTensor([tokens.shape[-1]]).to(device)
        text_mask = length_to_mask(input_lengths).to(device)

        t_en = model.text_encoder(tokens, input_lengths, text_mask)
        bert_dur = model.bert(tokens, attention_mask=(~text_mask).int())
        d_en = model.bert_encoder(bert_dur).transpose(-1, -2) 

        s_pred = sampler(noise = torch.randn((1, 256)).unsqueeze(1).to(device), 
                                          embedding=bert_dur,
                                          embedding_scale=embedding_scale,
                                            features=ref_s, # reference from the same speaker as the embedding
                                             num_steps=diffusion_steps).squeeze(1)


        s = s_pred[:, 128:]
        ref = s_pred[:, :128]

        ref = alpha * ref + (1 - alpha)  * ref_s[:, :128]
        s = beta * s + (1 - beta)  * ref_s[:, 128:]

        d = model.predictor.text_encoder(d_en, 
                                         s, input_lengths, text_mask)

        x, _ = model.predictor.lstm(d)
        duration = model.predictor.duration_proj(x)

        duration = torch.sigmoid(duration).sum(axis=-1)
        pred_dur = torch.round(duration.squeeze()).clamp(min=1)


        pred_aln_trg = torch.zeros(input_lengths, int(pred_dur.sum().data))
        c_frame = 0
        for i in range(pred_aln_trg.size(0)):
            pred_aln_trg[i, c_frame:c_frame + int(pred_dur[i].data)] = 1
            c_frame += int(pred_dur[i].data)

        # encode prosody
        en = (d.transpose(-1, -2) @ pred_aln_trg.unsqueeze(0).to(device))
        if model_params.decoder.type == "hifigan":
            asr_new = torch.zeros_like(en)
            asr_new[:, :, 0] = en[:, :, 0]
            asr_new[:, :, 1:] = en[:, :, 0:-1]
            en = asr_new

        F0_pred, N_pred = model.predictor.F0Ntrain(en, s)

        asr = (t_en @ pred_aln_trg.unsqueeze(0).to(device))
        if model_params.decoder.type == "hifigan":
            asr_new = torch.zeros_like(asr)
            asr_new[:, :, 0] = asr[:, :, 0]
            asr_new[:, :, 1:] = asr[:, :, 0:-1]
            asr = asr_new

        out = model.decoder(asr, 
                                F0_pred, N_pred, ref.squeeze().unsqueeze(0))
    
        
    return out.squeeze().cpu().numpy()[..., :-50] # weird pulse at the end of the model, need to be fixed later


voice_path = "Demo/reference_audio/James2.wav"
synth_voice_path = "synth.wav"
start = time.time()
noise = torch.randn(1,1,256).to(device)
ref_s = compute_style(voice_path)
wav = inference(transcribed_text, ref_s, alpha=0.1, beta=0.5, diffusion_steps=10, embedding_scale=1)
rtf = (time.time() - start) / (len(wav) / sample_rate_value)
print('Synthesized:')
st.audio(wav, format='audio/wav', sample_rate=sample_rate_value)


import scipy.io.wavfile
scaled = np.int16(wav / np.max(np.abs(wav)) * 32767)
scipy.io.wavfile.write(synth_voice_path, sample_rate_value, scaled)

def __plot_helper(audio_file: str):
        raw = wave.open(audio_file)
        signal = raw.readframes(-1)
        signal = np.frombuffer(signal, dtype ="int16")
        f_rate = raw.getframerate()
        time = np.linspace(
            0, # start
            len(signal) / f_rate,
            num = len(signal)
        )
        return (time, signal)


# shows the sound waves
def visualize(first_wav: str, second_wav: str):
    fig, axs = plt.subplots(2)
    fig.suptitle("Original vs Native")
    plt.xlabel("Time")
    t1, s1 = __plot_helper(first_wav)
    axs[0].plot(t1,s1)
    axs[0].set_yticklabels([])
    axs[0].set_yticks([])
    t2, s2 = __plot_helper(second_wav)
    axs[1].plot(t2,s2)
    axs[1].set_yticklabels([])
    axs[1].set_yticks([])
    st.pyplot(plt)


visualize(audio_path, synth_voice_path)
#print('Reference Voice:')
#display(ipd.Audio(voice_path, rate=24000, normalize=False))