app.py

import gradio as gr
import torch
import os
import numpy as np
from scipy import signal
import warnings
from transformers import SpeechT5Processor, SpeechT5ForTextToSpeech, SpeechT5HifiGan, set_seed
import requests
import tempfile
from pathlib import Path
import traceback
warnings.filterwarnings("ignore")

# Imposta un seed per la riproducibilità
set_seed(42)

# Definizioni di variabili globali
CACHE_DIR = "./model_cache"  # Directory per la cache del modello
SAMPLE_RATE = 16000  # Frequenza di campionamento per il modello Microsoft

# Cache per i componenti del modello
processor = None
model = None
vocoder = None
speaker_embeddings_cache = {}

def download_file(url, save_path):
    """Scarica un file da un URL"""
    try:
        response = requests.get(url, stream=True)
        response.raise_for_status()
        
        os.makedirs(os.path.dirname(save_path), exist_ok=True)
        with open(save_path, 'wb') as f:
            for chunk in response.iter_content(chunk_size=8192):
                f.write(chunk)
        
        return save_path
    except Exception as e:
        print(f"Errore nel download del file: {e}")
        return None

def get_correct_speaker_embeddings(speaker_id):
    """Ottieni gli speaker embeddings corretti per il modello SpeechT5"""
    global speaker_embeddings_cache
    
    # Correggi il valore di speaker_id per essere tra 1 e 10 (e non 0-9)
    # poiché i file su Hugging Face sono numerati da 01 a 24
    speaker_id = max(1, min(10, speaker_id + 1))
    
    if speaker_id in speaker_embeddings_cache:
        return speaker_embeddings_cache[speaker_id]
    
    try:
        # Url corretto per gli speaker embeddings
        url = f"https://huggingface.co/datasets/Matthijs/cmu-arctic-xvectors/resolve/main/cmu_us_{speaker_id:02d}_xvector.pt"
        
        # Crea un file temporaneo per gli embeddings
        tmp_dir = os.path.join(CACHE_DIR, "speakers")
        os.makedirs(tmp_dir, exist_ok=True)
        tmp_file = os.path.join(tmp_dir, f"speaker_{speaker_id:02d}.pt")
        
        # Scarica gli embeddings se non esistono già
        if not os.path.exists(tmp_file):
            print(f"Scaricamento degli speaker embeddings per lo speaker {speaker_id}...")
            filepath = download_file(url, tmp_file)
            if not filepath:
                raise ValueError(f"Impossibile scaricare gli embeddings per lo speaker {speaker_id}")
        
        # Carica gli embeddings
        speaker_embeddings = torch.load(tmp_file)
        
        # Memorizza gli embeddings nella cache
        speaker_embeddings_cache[speaker_id] = speaker_embeddings
        
        print(f"Speaker embeddings caricati per lo speaker {speaker_id}")
        return speaker_embeddings
    except Exception as e:
        print(f"Errore nel caricamento degli speaker embeddings: {e}")
        
        # Prova a utilizzare un embedding predefinito dal speaker 1
        try:
            fallback_id = 1
            url = f"https://huggingface.co/datasets/Matthijs/cmu-arctic-xvectors/resolve/main/cmu_us_{fallback_id:02d}_xvector.pt"
            tmp_file = os.path.join(tmp_dir, f"speaker_{fallback_id:02d}.pt")
            
            if not os.path.exists(tmp_file):
                download_file(url, tmp_file)
            
            speaker_embeddings = torch.load(tmp_file)
            speaker_embeddings_cache[speaker_id] = speaker_embeddings
            print(f"Utilizzando speaker embeddings di fallback (speaker {fallback_id})")
            return speaker_embeddings
        except:
            # Se tutto fallisce, crea una embedding vuota
            print("Creazione di embeddings vuoti")
            default_embeddings = torch.zeros((1, 512))
            speaker_embeddings_cache[speaker_id] = default_embeddings
            return default_embeddings

def load_tts_components():
    """Carica tutti i componenti TTS"""
    global processor, model, vocoder
    
    if processor is None or model is None or vocoder is None:
        try:
            print("📂 Caricamento dei componenti TTS...")
            # Carica il processor, modello e vocoder
            processor = SpeechT5Processor.from_pretrained("microsoft/speecht5_tts")
            model = SpeechT5ForTextToSpeech.from_pretrained("microsoft/speecht5_tts")
            vocoder = SpeechT5HifiGan.from_pretrained("microsoft/speecht5_hifigan")
            
            print("✅ Componenti TTS caricati con successo!")
            return True
        except Exception as e:
            print(f"❌ Errore nel caricamento dei componenti TTS: {e}")
            traceback_str = traceback.format_exc()
            print(f"Traceback completo:\n{traceback_str}")
            return False

def text_to_speech(text, language="it", speaker_id=0, speed=1.0, show_log=False):
    """Converte testo in voce utilizzando il modello TTS"""
    if not text.strip():
        return None, "Per favore, inserisci del testo da convertire in voce."
    
    # Log di debug
    if show_log:
        print(f"Richiesta TTS ricevuta: '{text}' (Lingua: {language}, Speaker: {speaker_id}, Velocità: {speed})")
    
    try:
        # Carica i componenti se non già caricati
        if not load_tts_components():
            return None, "Errore nel caricamento dei componenti TTS. Riprova più tardi."
        
        # Ottieni gli speaker embeddings
        speaker_emb = get_correct_speaker_embeddings(speaker_id)
        
        # Microsoft SpeechT5 non supporta il parametro language, quindi lo ignoriamo
        if show_log:
            print(f"Elaborazione del testo: '{text}'")
            
        # Crea input IDs dal testo
        inputs = processor(text=text, return_tensors="pt")
        
        # Genera l'audio con speaker embeddings
        if show_log:
            print("Generazione dell'audio...")
            
        with torch.no_grad():
            # Sposta gli speaker embeddings sul dispositivo di calcolo
            if hasattr(model, "device"):
                speaker_emb = speaker_emb.to(model.device)
                
            # Genera l'audio
            speech = model.generate_speech(
                inputs["input_ids"], 
                speaker_emb,
                vocoder=vocoder
            )
        
        # Imposta la frequenza di campionamento
        sample_rate = SAMPLE_RATE
        
        # Converti il tensore in un array numpy
        speech_array = speech.numpy()
        
        # Applica il controllo della velocità
        if speed != 1.0:
            # Usa scipy.signal per ricampionare l'audio e cambiare la velocità
            speech_array = signal.resample(speech_array, int(len(speech_array) / speed))
        
        if show_log:
            print(f"✅ Audio generato con successo! Lunghezza: {len(speech_array)} campioni")
        
        return (sample_rate, speech_array), None
    except Exception as e:
        error_msg = f"Errore nella generazione dell'audio: {str(e)}"
        traceback_str = traceback.format_exc()
        detailed_error = f"{error_msg}\n\nTraceback dettagliato:\n{traceback_str}"
        print(f"❌ {detailed_error}")
        return None, detailed_error

# Esempi predefiniti per l'interfaccia
examples = [
    ["Ciao, mi chiamo Aurora e sono un assistente vocale italiano.", "it", 0, 1.0, False],
    ["Hello, my name is Aurora and I'm an Italian voice assistant.", "en", 1, 1.0, False],
    ["Hola, me llamo Aurora y soy un asistente de voz italiano.", "es", 2, 1.0, False],
    ["La vita è bella e il sole splende nel cielo azzurro.", "it", 3, 1.0, False],
    ["Mi piace viaggiare e scoprire nuove città e culture.", "it", 4, 1.2, False],
    ["L'intelligenza artificiale sta trasformando il modo in cui interagiamo con i computer e con il mondo che ci circonda.", "it", 5, 0.9, False]
]

# Definizione dell'interfaccia Gradio
with gr.Blocks(title="Aurora TTS Demo", theme=gr.themes.Soft()) as demo:
    gr.Markdown("""
    # 🎙️ Text-to-Speech Demo
    
    Questa demo utilizza il modello Microsoft SpeechT5 per la sintesi vocale (TTS).
    
    Puoi selezionare diversi stili di voce cambiando lo Speaker ID.
    """)
    
    with gr.Row():
        with gr.Column(scale=2):
            text_input = gr.Textbox(
                label="Testo da convertire in voce",
                placeholder="Inserisci qui il testo da convertire...",
                lines=5,
                value="Ciao, sono Aurora, un assistente vocale italiano basato su intelligenza artificiale."
            )
            
            with gr.Row():
                language_input = gr.Dropdown(
                    choices=["it", "en", "es", "fr", "de"], 
                    label="Lingua", 
                    value="it",
                    info="Seleziona la lingua del testo (nota: il modello attuale si comporta meglio con l'inglese)"
                )
                speaker_input = gr.Slider(
                    label="Speaker ID",
                    value=0,
                    minimum=0,
                    maximum=9,
                    step=1,
                    info="ID dello speaker (0-9, ogni ID ha caratteristiche vocali diverse)"
                )
                speed_input = gr.Slider(
                    minimum=0.5, 
                    maximum=1.5, 
                    value=1.0, 
                    step=0.1, 
                    label="Velocità",
                    info="Valori più bassi = voce più lenta, valori più alti = voce più veloce"
                )
            
            debug_input = gr.Checkbox(label="Mostra log di debug", value=True)
            
            submit_btn = gr.Button("Genera Audio", variant="primary")
            
        with gr.Column(scale=1):
            audio_output = gr.Audio(label="Audio generato", show_share_button=True)
            error_output = gr.Textbox(label="Messaggi di errore", visible=True, lines=4)
    
    # Esempi
    gr.Examples(
        examples=examples,
        inputs=[text_input, language_input, speaker_input, speed_input, debug_input],
        outputs=[audio_output, error_output],
        fn=text_to_speech,
        cache_examples=True,
    )
    
    # Info aggiuntive
    gr.Markdown("""
    ## 📝 Note sull'utilizzo
    
    - Il modello funziona meglio con frasi di lunghezza media (fino a 20-30 parole)
    - Puoi cambiare lo Speaker ID per ottenere voci con caratteristiche diverse
    - La velocità di generazione dipende dalle risorse disponibili sul server
    
    ## 🔗 Crediti
    
    - [Microsoft SpeechT5](https://huggingface.co/microsoft/speecht5_tts) (modello base)
    - [CMU Arctic XVectors](https://huggingface.co/datasets/Matthijs/cmu-arctic-xvectors) (speaker embeddings)
    """)
    
    # Configurazione degli eventi
    submit_btn.click(
        fn=text_to_speech,
        inputs=[text_input, language_input, speaker_input, speed_input, debug_input],
        outputs=[audio_output, error_output],
    )

# Avvia l'interfaccia
demo.launch()