Add updated Dockerfile and app.py

Dockerfile

@@ -0,0 +1,36 @@
+FROM python:3.9-slim
+
+# Set environment variables
+ENV PYTHONDONTWRITEBYTECODE=1
+ENV PYTHONUNBUFFERED=1
+ENV PORT=8080
+
+# Create a non-root user
+RUN useradd -m -u 1000 user
+USER user
+ENV PATH="/home/user/.local/bin:$PATH"
+
+WORKDIR /app
+
+# Install system dependencies
+RUN apt-get update && apt-get install -y \
+    build-essential \
+    libsndfile1 \
+    ffmpeg \
+    && rm -rf /var/lib/apt/lists/*
+
+# Copy requirements and install Python dependencies
+COPY --chown=user:users requirements.txt /app/requirements.txt
+RUN pip install --no-cache-dir --upgrade pip && pip install --no-cache-dir -r requirements.txt
+
+# Copy the rest of the project files
+COPY --chown=user:users . /app
+
+# Ensure outputs and temp directories exist
+RUN mkdir -p outputs temp
+
+# Expose port 8080
+EXPOSE 8080
+
+# Run the application
+CMD ["python", "app.py"]

api.py

@@ -0,0 +1,195 @@
+import torch
+import numpy as np
+import re
+import soundfile
+import utils
+import commons
+import os
+import librosa
+from text import text_to_sequence
+from mel_processing import spectrogram_torch
+from models import SynthesizerTrn
+
+
+class OpenVoiceBaseClass(object):
+    def __init__(self, 
+                config_path, 
+                device='cuda:0'):
+        if 'cuda' in device:
+            assert torch.cuda.is_available()
+
+        hps = utils.get_hparams_from_file(config_path)
+
+        model = SynthesizerTrn(
+            len(getattr(hps, 'symbols', [])),
+            hps.data.filter_length // 2 + 1,
+            n_speakers=hps.data.n_speakers,
+            **hps.model,
+        ).to(device)
+
+        model.eval()
+        self.model = model
+        self.hps = hps
+        self.device = device
+
+    def load_ckpt(self, ckpt_path):
+        checkpoint_dict = torch.load(ckpt_path, map_location=torch.device(self.device))
+        a, b = self.model.load_state_dict(checkpoint_dict['model'], strict=False)
+        print("Loaded checkpoint '{}'".format(ckpt_path))
+        print('missing/unexpected keys:', a, b)
+
+
+class BaseSpeakerTTS(OpenVoiceBaseClass):
+    language_marks = {
+        "english": "EN",
+        "chinese": "ZH",
+    }
+
+    @staticmethod
+    def get_text(text, hps, is_symbol):
+        text_norm = text_to_sequence(text, hps.symbols, [] if is_symbol else hps.data.text_cleaners)
+        if hps.data.add_blank:
+            text_norm = commons.intersperse(text_norm, 0)
+        text_norm = torch.LongTensor(text_norm)
+        return text_norm
+
+    @staticmethod
+    def audio_numpy_concat(segment_data_list, sr, speed=1.):
+        audio_segments = []
+        for segment_data in segment_data_list:
+            audio_segments += segment_data.reshape(-1).tolist()
+            audio_segments += [0] * int((sr * 0.05)/speed)
+        audio_segments = np.array(audio_segments).astype(np.float32)
+        return audio_segments
+
+    @staticmethod
+    def split_sentences_into_pieces(text, language_str):
+        texts = utils.split_sentence(text, language_str=language_str)
+        print(" > Text splitted to sentences.")
+        print('\n'.join(texts))
+        print(" > ===========================")
+        return texts
+
+    def tts(self, text, output_path, speaker, language='English', speed=1.0):
+        mark = self.language_marks.get(language.lower(), None)
+        assert mark is not None, f"language {language} is not supported"
+
+        texts = self.split_sentences_into_pieces(text, mark)
+
+        audio_list = []
+        for t in texts:
+            t = re.sub(r'([a-z])([A-Z])', r'\1 \2', t)
+            t = f'[{mark}]{t}[{mark}]'
+            stn_tst = self.get_text(t, self.hps, False)
+            device = self.device
+            speaker_id = self.hps.speakers[speaker]
+            with torch.no_grad():
+                x_tst = stn_tst.unsqueeze(0).to(device)
+                x_tst_lengths = torch.LongTensor([stn_tst.size(0)]).to(device)
+                sid = torch.LongTensor([speaker_id]).to(device)
+                audio = self.model.infer(x_tst, x_tst_lengths, sid=sid, noise_scale=0.667, noise_scale_w=0.6,
+                                    length_scale=1.0 / speed)[0][0, 0].data.cpu().float().numpy()
+            audio_list.append(audio)
+        audio = self.audio_numpy_concat(audio_list, sr=self.hps.data.sampling_rate, speed=speed)
+
+        if output_path is None:
+            return audio
+        else:
+            soundfile.write(output_path, audio, self.hps.data.sampling_rate)
+
+
+class ToneColorConverter(OpenVoiceBaseClass):
+    def __init__(self, *args, **kwargs):
+        super().__init__(*args, **kwargs)
+
+        self.watermark_model = None
+
+    def extract_se(self, ref_wav_list, se_save_path=None):
+        if isinstance(ref_wav_list, str):
+            ref_wav_list = [ref_wav_list]
+        
+        device = self.device
+        hps = self.hps
+        gs = []
+        
+        for fname in ref_wav_list:
+            audio_ref, sr = librosa.load(fname, sr=hps.data.sampling_rate)
+            y = torch.FloatTensor(audio_ref)
+            y = y.to(device)
+            y = y.unsqueeze(0)
+            y = spectrogram_torch(y, hps.data.filter_length,
+                                        hps.data.sampling_rate, hps.data.hop_length, hps.data.win_length,
+                                        center=False).to(device)
+            with torch.no_grad():
+                g = self.model.ref_enc(y.transpose(1, 2)).unsqueeze(-1)
+                gs.append(g.detach())
+        gs = torch.stack(gs).mean(0)
+
+        if se_save_path is not None:
+            os.makedirs(os.path.dirname(se_save_path), exist_ok=True)
+            torch.save(gs.cpu(), se_save_path)
+
+        return gs
+
+    def convert(self, audio_src_path, src_se, tgt_se, output_path=None, tau=0.3, message="default"):
+        hps = self.hps
+        # load audio
+        audio, sample_rate = librosa.load(audio_src_path, sr=hps.data.sampling_rate)
+        audio = torch.tensor(audio).float()
+        
+        with torch.no_grad():
+            y = torch.FloatTensor(audio).to(self.device)
+            y = y.unsqueeze(0)
+            spec = spectrogram_torch(y, hps.data.filter_length,
+                                    hps.data.sampling_rate, hps.data.hop_length, hps.data.win_length,
+                                    center=False).to(self.device)
+            spec_lengths = torch.LongTensor([spec.size(-1)]).to(self.device)
+            audio = self.model.voice_conversion(spec, spec_lengths, sid_src=src_se, sid_tgt=tgt_se, tau=tau)[0][
+                        0, 0].data.cpu().float().numpy()
+            audio = self.add_watermark(audio, message)
+            if output_path is None:
+                return audio
+            else:
+                soundfile.write(output_path, audio, hps.data.sampling_rate)
+    
+    def add_watermark(self, audio, message):
+        if self.watermark_model is None:
+            return audio
+        device = self.device
+        bits = utils.string_to_bits(message).reshape(-1)
+        n_repeat = len(bits) // 32
+
+        K = 16000
+        coeff = 2
+        for n in range(n_repeat):
+            trunck = audio[(coeff * n) * K: (coeff * n + 1) * K]
+            if len(trunck) != K:
+                print('Audio too short, fail to add watermark')
+                break
+            message_npy = bits[n * 32: (n + 1) * 32]
+            
+            with torch.no_grad():
+                signal = torch.FloatTensor(trunck).to(device)[None]
+                message_tensor = torch.FloatTensor(message_npy).to(device)[None]
+                signal_wmd_tensor = self.watermark_model.encode(signal, message_tensor)
+                signal_wmd_npy = signal_wmd_tensor.detach().cpu().squeeze()
+            audio[(coeff * n) * K: (coeff * n + 1) * K] = signal_wmd_npy
+        return audio
+
+    def detect_watermark(self, audio, n_repeat):
+        bits = []
+        K = 16000
+        coeff = 2
+        for n in range(n_repeat):
+            trunck = audio[(coeff * n) * K: (coeff * n + 1) * K]
+            if len(trunck) != K:
+                print('Audio too short, fail to detect watermark')
+                return 'Fail'
+            with torch.no_grad():
+                signal = torch.FloatTensor(trunck).to(self.device).unsqueeze(0)
+                message_decoded_npy = (self.watermark_model.decode(signal) >= 0.5).int().detach().cpu().numpy().squeeze()
+            bits.append(message_decoded_npy)
+        bits = np.stack(bits).reshape(-1, 8)
+        message = utils.bits_to_string(bits)
+        return message
+    

app.py

@@ -0,0 +1,226 @@
+import os
+import torch
+import argparse
+import gradio as gr
+import openai
+from zipfile import ZipFile
+import requests
+import se_extractor
+from api import BaseSpeakerTTS, ToneColorConverter
+import langid
+import traceback
+from dotenv import load_dotenv
+
+# Load environment variables
+load_dotenv()
+
+# Function to download and extract checkpoints
+def download_and_extract_checkpoints():
+    zip_url = "https://huggingface.co/camenduru/OpenVoice/resolve/main/checkpoints_1226.zip"
+    zip_path = "checkpoints.zip"
+
+    if not os.path.exists("checkpoints"):
+        print("Downloading checkpoints...")
+        response = requests.get(zip_url, stream=True)
+        with open(zip_path, "wb") as zip_file:
+            for chunk in response.iter_content(chunk_size=8192):
+                if chunk:
+                    zip_file.write(chunk)
+        print("Extracting checkpoints...")
+        with ZipFile(zip_path, "r") as zip_ref:
+            zip_ref.extractall(".")
+        os.remove(zip_path)
+        print("Checkpoints are ready.")
+
+# Call the function to ensure checkpoints are available
+download_and_extract_checkpoints()
+
+# Initialize OpenAI API key
+openai.api_key = os.getenv("OPENAI_API_KEY")
+if not openai.api_key:
+    raise ValueError("Please set the OPENAI_API_KEY environment variable.")
+
+parser = argparse.ArgumentParser()
+parser.add_argument("--share", action='store_true', default=False, help="make link public")
+args = parser.parse_args()
+
+# Define paths to checkpoints
+en_ckpt_base = 'checkpoints/base_speakers/EN'
+zh_ckpt_base = 'checkpoints/base_speakers/ZH'
+ckpt_converter = 'checkpoints/converter'
+device = 'cuda' if torch.cuda.is_available() else 'cpu'
+output_dir = 'outputs'
+os.makedirs(output_dir, exist_ok=True)
+
+# Load TTS models
+en_base_speaker_tts = BaseSpeakerTTS(f'{en_ckpt_base}/config.json', device=device)
+en_base_speaker_tts.load_ckpt(f'{en_ckpt_base}/checkpoint.pth')
+zh_base_speaker_tts = BaseSpeakerTTS(f'{zh_ckpt_base}/config.json', device=device)
+zh_base_speaker_tts.load_ckpt(f'{zh_ckpt_base}/checkpoint.pth')
+
+tone_color_converter = ToneColorConverter(f'{ckpt_converter}/config.json', device=device)
+tone_color_converter.load_ckpt(f'{ckpt_converter}/checkpoint.pth')
+
+# Load speaker embeddings
+en_source_default_se = torch.load(f'{en_ckpt_base}/en_default_se.pth').to(device)
+en_source_style_se = torch.load(f'{en_ckpt_base}/en_style_se.pth').to(device)
+zh_source_se = torch.load(f'{zh_ckpt_base}/zh_default_se.pth').to(device)
+
+# Extract speaker embedding from the default Mickey Mouse audio
+default_speaker_audio = "resources/output.wav"
+try:
+    target_se, _ = se_extractor.get_se(
+        default_speaker_audio,
+        tone_color_converter,
+        target_dir='processed',
+        vad=True
+    )
+    print("Speaker embedding extracted successfully.")
+except Exception as e:
+    raise RuntimeError(f"Failed to extract speaker embedding from {default_speaker_audio}: {str(e)}")
+
+# Supported languages
+supported_languages = ['zh', 'en']
+
+def predict(audio_file_pth, agree):
+    text_hint = ''
+    synthesized_audio_path = None
+
+    # Agree with the terms
+    if not agree:
+        text_hint += '[ERROR] Please accept the Terms & Conditions!\n'
+        return (text_hint, None)
+
+    # Check if audio file is provided
+    if audio_file_pth is not None:
+        speaker_wav = audio_file_pth
+    else:
+        text_hint += "[ERROR] Please record your voice using the Microphone.\n"
+        return (text_hint, None)
+
+    # Transcribe audio to text using OpenAI Whisper
+    try:
+        with open(speaker_wav, 'rb') as audio_file:
+            transcription_response = openai.Audio.transcribe(
+                model="whisper-1",
+                file=audio_file,
+                response_format='text'
+            )
+        input_text = transcription_response.strip()
+        print(f"Transcribed Text: {input_text}")
+    except Exception as e:
+        text_hint += f"[ERROR] Transcription failed: {str(e)}\n"
+        return (text_hint, None)
+
+    if len(input_text) == 0:
+        text_hint += "[ERROR] No speech detected in the audio.\n"
+        return (text_hint, None)
+
+    # Detect language
+    language_predicted = langid.classify(input_text)[0].strip()
+    print(f"Detected language: {language_predicted}")
+
+    if language_predicted not in supported_languages:
+        text_hint += f"[ERROR] The detected language '{language_predicted}' is not supported. Supported languages are: {supported_languages}\n"
+        return (text_hint, None)
+
+    # Select TTS model based on language
+    if language_predicted == "zh":
+        tts_model = zh_base_speaker_tts
+        language = 'Chinese'
+        speaker_style = 'default'
+    else:
+        tts_model = en_base_speaker_tts
+        language = 'English'
+        speaker_style = 'default'
+
+    # Generate response using OpenAI GPT-4
+    try:
+        response = openai.ChatCompletion.create(
+            model="gpt-4o-mini",
+            messages=[
+                {"role": "system", "content": "You are Mickey Mouse, a friendly and cheerful character who responds to children's queries in a simple and engaging manner. Please keep your response up to 200 characters."},
+                {"role": "user", "content": input_text}
+            ],
+            max_tokens=200,
+            temperature=0.7,
+        )
+        reply_text = response['choices'][0]['message']['content'].strip()
+        print(f"GPT-4 Reply: {reply_text}")
+    except Exception as e:
+        text_hint += f"[ERROR] Failed to get response from OpenAI GPT-4: {str(e)}\n"
+        return (text_hint, None)
+
+    # Synthesize reply text to audio
+    try:
+        src_path = os.path.join(output_dir, 'tmp_reply.wav')
+
+        tts_model.tts(reply_text, src_path, speaker=speaker_style, language=language)
+        print(f"Audio synthesized and saved to {src_path}")
+
+        save_path = os.path.join(output_dir, 'output_reply.wav')
+
+        tone_color_converter.convert(
+            audio_src_path=src_path, 
+            src_se=en_source_default_se if language == 'English' else zh_source_se,
+            tgt_se=target_se,
+            output_path=save_path,
+            message="@MickeyMouse"
+        )
+        print(f"Tone color conversion completed and saved to {save_path}")
+
+        text_hint += "Response generated successfully.\n"
+        synthesized_audio_path = save_path
+
+    except Exception as e:
+        text_hint += f"[ERROR] Failed to synthesize audio: {str(e)}\n"
+        traceback.print_exc()
+        return (text_hint, None)
+
+    return (text_hint, synthesized_audio_path)
+
+with gr.Blocks(analytics_enabled=False) as demo:
+    gr.Markdown("# Mickey Mouse Voice Assistant")
+
+    with gr.Row():
+        with gr.Column():
+            audio_input = gr.Audio(
+                source="microphone",
+                type="filepath",
+                label="Record Your Voice",
+                info="Click the microphone button to record your voice."
+            )
+            tos_checkbox = gr.Checkbox(
+                label="Agree to Terms & Conditions",
+                value=False,
+                info="I agree to the terms of service."
+            )
+            submit_button = gr.Button("Send")
+
+        with gr.Column():
+            info_output = gr.Textbox(
+                label="Info",
+                interactive=False,
+                lines=4,
+            )
+            audio_output = gr.Audio(
+                label="Mickey's Response",
+                interactive=False,
+                autoplay=True,
+            )
+
+    submit_button.click(
+        predict, 
+        inputs=[audio_input, tos_checkbox], 
+        outputs=[info_output, audio_output]
+    )
+
+# Launch the Gradio app
+demo.queue()
+demo.launch(
+    server_name="0.0.0.0",
+    server_port=int(os.environ.get("PORT", 8080)),
+    debug=True,
+    show_api=True,
+    share=False
+)

requirements.txt

@@ -0,0 +1,25 @@
+librosa==0.9.1
+faster-whisper==0.9.0
+pydub==0.25.1
+wavmark==0.0.2
+numpy==1.22.0
+eng_to_ipa==0.0.2
+inflect==7.0.0
+unidecode==1.3.7
+whisper-timestamped==1.14.2
+openai
+python-dotenv
+pypinyin==0.50.0
+cn2an==0.5.22
+jieba==0.42.1
+gradio==3.50.2
+ffmpeg-python
+fastapi
+uvicorn
+torch
+langid
+requests
+fastapi
+uvicorn[standard]
+webrtcvad
+

se_extractor.py

@@ -0,0 +1,139 @@
+import os
+import glob
+import torch
+from glob import glob
+import numpy as np
+from pydub import AudioSegment
+from faster_whisper import WhisperModel
+from whisper_timestamped.transcribe import get_audio_tensor, get_vad_segments
+
+model_size = "medium"
+# Run on GPU with FP16
+model = None
+def split_audio_whisper(audio_path, target_dir='processed'):
+    global model
+    if model is None:
+        model = WhisperModel(model_size, device="cuda", compute_type="float16")
+    audio = AudioSegment.from_file(audio_path)
+    max_len = len(audio)
+
+    audio_name = os.path.basename(audio_path).rsplit('.', 1)[0]
+    target_folder = os.path.join(target_dir, audio_name)
+    
+    segments, info = model.transcribe(audio_path, beam_size=5, word_timestamps=True)
+    segments = list(segments)    
+
+    # create directory
+    os.makedirs(target_folder, exist_ok=True)
+    wavs_folder = os.path.join(target_folder, 'wavs')
+    os.makedirs(wavs_folder, exist_ok=True)
+
+    # segments
+    s_ind = 0
+    start_time = None
+    
+    for k, w in enumerate(segments):
+        # process with the time
+        if k == 0:
+            start_time = max(0, w.start)
+
+        end_time = w.end
+
+        # calculate confidence
+        if len(w.words) > 0:
+            confidence = sum([s.probability for s in w.words]) / len(w.words)
+        else:
+            confidence = 0.
+        # clean text
+        text = w.text.replace('...', '')
+
+        # left 0.08s for each audios
+        audio_seg = audio[int( start_time * 1000) : min(max_len, int(end_time * 1000) + 80)]
+
+        # segment file name
+        fname = f"{audio_name}_seg{s_ind}.wav"
+
+        # filter out the segment shorter than 1.5s and longer than 20s
+        save = audio_seg.duration_seconds > 1.5 and \
+                audio_seg.duration_seconds < 20. and \
+                len(text) >= 2 and len(text) < 200 
+
+        if save:
+            output_file = os.path.join(wavs_folder, fname)
+            audio_seg.export(output_file, format='wav')
+
+        if k < len(segments) - 1:
+            start_time = max(0, segments[k+1].start - 0.08)
+
+        s_ind = s_ind + 1
+    return wavs_folder
+
+
+def split_audio_vad(audio_path, target_dir, split_seconds=10.0):
+    SAMPLE_RATE = 16000
+    audio_vad = get_audio_tensor(audio_path)
+    segments = get_vad_segments(
+        audio_vad,
+        output_sample=True,
+        min_speech_duration=0.1,
+        min_silence_duration=1,
+        method="silero",
+    )
+    segments = [(seg["start"], seg["end"]) for seg in segments]
+    segments = [(float(s) / SAMPLE_RATE, float(e) / SAMPLE_RATE) for s,e in segments]
+    print(segments)
+    audio_active = AudioSegment.silent(duration=0)
+    audio = AudioSegment.from_file(audio_path)
+
+    for start_time, end_time in segments:
+        audio_active += audio[int( start_time * 1000) : int(end_time * 1000)]
+    
+    audio_dur = audio_active.duration_seconds
+    print(f'after vad: dur = {audio_dur}')
+    audio_name = os.path.basename(audio_path).rsplit('.', 1)[0]
+    target_folder = os.path.join(target_dir, audio_name)
+    wavs_folder = os.path.join(target_folder, 'wavs')
+    os.makedirs(wavs_folder, exist_ok=True)
+    start_time = 0.
+    count = 0
+    num_splits = int(np.round(audio_dur / split_seconds))
+    assert num_splits > 0, 'input audio is too short'
+    interval = audio_dur / num_splits
+
+    for i in range(num_splits):
+        end_time = min(start_time + interval, audio_dur)
+        if i == num_splits - 1:
+            end_time = audio_dur
+        output_file = f"{wavs_folder}/{audio_name}_seg{count}.wav"
+        audio_seg = audio_active[int(start_time * 1000): int(end_time * 1000)]
+        audio_seg.export(output_file, format='wav')
+        start_time = end_time
+        count += 1
+    return wavs_folder
+
+
+    
+
+
+def get_se(audio_path, vc_model, target_dir='processed', vad=True):
+    device = vc_model.device
+
+    audio_name = os.path.basename(audio_path).rsplit('.', 1)[0]
+    se_path = os.path.join(target_dir, audio_name, 'se.pth')
+
+    if os.path.isfile(se_path):
+        se = torch.load(se_path).to(device)
+        return se, audio_name
+    if os.path.isdir(audio_path):
+        wavs_folder = audio_path
+    elif vad:
+        wavs_folder = split_audio_vad(audio_path, target_dir)
+    else:
+        wavs_folder = split_audio_whisper(audio_path, target_dir)
+    
+    audio_segs = glob(f'{wavs_folder}/*.wav')
+    if len(audio_segs) == 0:
+        raise NotImplementedError('No audio segments found!')
+    
+    return vc_model.extract_se(audio_segs, se_save_path=se_path), audio_name
+