Switch to Microsoft SpeechT5 model for immediate TTS functionality

app.py

@@ -3,108 +3,121 @@ import gradio as gr
 import torch
 import os
 import json
-from transformers import AutoProcessor, AutoConfig, AutoTokenizer, AutoModelForSeq2SeqLM, SpeechT5ForTextToSpeech, set_seed
 import numpy as np
 from scipy import signal
 import warnings
+from transformers import AutoConfig, AutoTokenizer, AutoProcessor, SpeechT5ForTextToSpeech, 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
-MODEL_REPO = "Lorenzob/aurora-1.6b-complete"  # Repository principale
-FALLBACK_REPO = "nari-labs/Dia-1.6B"  # Repository di fallback
-SAMPLE_RATE = 24000  # Frequenza di campionamento per il modello TTS
+MODEL_REPO = "Lorenzob/aurora-1.6b-complete"  # Repository del modello completo
+CACHE_DIR = "./model_cache"  # Directory per la cache del modello
+SAMPLE_RATE = 24000  # Frequenza di campionamento
 
-# Cache per il modello e il processor (per evitare di ricaricarli ad ogni richiesta)
-model = None
+# Cache per il modello e il processor
 processor = None
+model = None
+
+def download_file(url, save_path):
+    """Scarica un file da un URL"""
+    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
 
-def fix_model_config(model_path):
-    """Aggiunge il model_type alla configurazione se necessario"""
+def ensure_model_config():
+    """Assicura che il modello abbia una configurazione corretta"""
     try:
-        config_path = os.path.join(model_path, "config.json")
-        if os.path.exists(config_path):
-            with open(config_path, "r") as f:
-                config = json.load(f)
-            
-            # Aggiungi model_type se mancante
-            if "model_type" not in config:
-                config["model_type"] = "speecht5"
-                print(f"Aggiunto model_type 'speecht5' alla configurazione")
-            
-            # Aggiungi architectures se mancante
-            if "architectures" not in config:
-                config["architectures"] = ["SpeechT5ForTextToSpeech"]
-                print(f"Aggiunto architectures al config")
-            
-            # Salva la configurazione aggiornata
-            with open(config_path, "w") as f:
-                json.dump(config, f, indent=2)
+        # Crea la directory cache se non esiste
+        os.makedirs(CACHE_DIR, exist_ok=True)
+        
+        # Percorso per il file di configurazione locale
+        config_path = os.path.join(CACHE_DIR, "config.json")
+        
+        # Controlla se il file di configurazione esiste già
+        if not os.path.exists(config_path):
+            # URL del file di configurazione su Hugging Face
+            config_url = f"https://huggingface.co/{MODEL_REPO}/raw/main/config.json"
             
-            print(f"Configurazione aggiornata e salvata in {config_path}")
-            return True
-        else:
-            print(f"File di configurazione non trovato in {model_path}")
-            return False
+            # Scarica il file di configurazione
+            print(f"Scaricamento della configurazione da {config_url}...")
+            download_file(config_url, config_path)
+        
+        # Leggi il file di configurazione
+        with open(config_path, "r") as f:
+            config = json.load(f)
+        
+        # Modifica la configurazione per SpeechT5
+        config["model_type"] = "speecht5"
+        if "architectures" not in config or not config["architectures"]:
+            config["architectures"] = ["SpeechT5ForTextToSpeech"]
+        
+        # Salva la configurazione modificata
+        with open(config_path, "w") as f:
+            json.dump(config, f, indent=2)
+        
+        print(f"Configurazione aggiornata salvata in {config_path}")
+        return config_path
     except Exception as e:
-        print(f"Errore nella modifica del config: {e}")
-        return False
+        print(f"Errore nella configurazione del modello: {e}")
+        return None
 
 def load_model_and_processor():
-    """Carica il modello e il processor con gestione degli errori avanzata"""
+    """Carica il modello e il processor con caricamento manuale della configurazione"""
     global model, processor
     
     if model is None or processor is None:
         try:
-            print(f"📂 Tentativo di caricamento del modello da {MODEL_REPO}...")
+            print(f"📂 Caricamento del modello Aurora-1.6b-complete...")
+            
+            # Configura il modello manualmente
+            config_path = ensure_model_config()
+            if not config_path:
+                raise ValueError("Impossibile creare la configurazione del modello")
+            
+            # Carica la configurazione
+            config = AutoConfig.from_pretrained(config_path)
             
-            # Prova a caricare il processor
+            # Carica il processor
             processor = AutoProcessor.from_pretrained(MODEL_REPO)
             
-            # Carica il modello specificando esplicitamente la classe
+            # Carica il modello con la configurazione corretta
             model = SpeechT5ForTextToSpeech.from_pretrained(
-                MODEL_REPO, 
+                MODEL_REPO,
+                config=config,
                 torch_dtype=torch.float16 if torch.cuda.is_available() else torch.float32,
                 device_map="auto"
             )
-            print("✅ Modello principale caricato con successo!")
+            
+            print("✅ Modello e processor caricati con successo!")
             
         except Exception as e:
-            print(f"❌ Errore nel caricamento del modello principale: {e}")
+            print(f"❌ Errore nel caricamento del modello Aurora: {str(e)}")
+            traceback_str = traceback.format_exc()
+            print(f"Traceback completo:\n{traceback_str}")
             
-            # Prova con il modello di fallback
             try:
-                print(f"⚠️ Tentativo di caricamento del modello di fallback da {FALLBACK_REPO}...")
-                
-                # Prova a caricare il processor di fallback
-                processor = AutoProcessor.from_pretrained(FALLBACK_REPO)
-                
-                # Carica il modello di fallback specificando esplicitamente la classe
-                model = SpeechT5ForTextToSpeech.from_pretrained(
-                    FALLBACK_REPO,
-                    torch_dtype=torch.float16 if torch.cuda.is_available() else torch.float32,
-                    device_map="auto"
-                )
+                # Fallback al modello di Microsoft (più affidabile)
+                print("⚠️ Tentativo di fallback al modello Microsoft...")
+                from transformers import SpeechT5Processor, SpeechT5ForTextToSpeech
+                processor = SpeechT5Processor.from_pretrained("microsoft/speecht5_tts")
+                model = SpeechT5ForTextToSpeech.from_pretrained("microsoft/speecht5_tts")
                 print("✅ Modello di fallback caricato con successo!")
-                
             except Exception as e2:
-                print(f"❌ Errore anche nel caricamento del modello di fallback: {e2}")
-                
-                # Se entrambi i tentativi falliscono, prova con un modello TTS generico ben supportato
-                try:
-                    print("🔄 Tentativo con un modello TTS generico (microsoft/speecht5_tts)...")
-                    processor = AutoProcessor.from_pretrained("microsoft/speecht5_tts")
-                    model = SpeechT5ForTextToSpeech.from_pretrained(
-                        "microsoft/speecht5_tts",
-                        torch_dtype=torch.float16 if torch.cuda.is_available() else torch.float32,
-                        device_map="auto"
-                    )
-                    print("✅ Modello generico caricato con successo!")
-                except Exception as e3:
-                    print(f"❌ Tutti i tentativi di caricamento sono falliti: {e3}")
-                    raise RuntimeError("Impossibile caricare alcun modello TTS")
+                print(f"❌ Errore anche nel caricamento del modello di fallback: {str(e2)}")
+                raise RuntimeError("Impossibile caricare alcun modello TTS")
     
     return model, processor
 
@@ -118,39 +131,62 @@ def text_to_speech(text, language="it", speaker_id=0, speed=1.0, show_log=False)
         print(f"Richiesta TTS ricevuta: '{text}' (Lingua: {language}, Speaker: {speaker_id}, Velocità: {speed})")
     
     try:
-        # Carica il modello e il processor (se non già caricati)
+        # Carica il modello e il processor
         model, processor = load_model_and_processor()
         
-        # Prepara gli input per il modello
-        inputs = processor(
-            text=text, 
-            return_tensors="pt"
-        )
-        
-        # Aggiungi il parametro di lingua se supportato dal processor
-        if "language" in processor.model_input_names:
-            inputs["language"] = language
-        
-        # Sposta gli input sul dispositivo di calcolo
-        for k, v in inputs.items():
-            if hasattr(v, "to"):
-                inputs[k] = v.to(model.device)
-        
-        # Parametri per la generazione
-        gen_params = {
-            "do_sample": True,
-            "temperature": 0.7,
-            "top_k": 50,
-            "top_p": 0.95,
-        }
+        # Controlla se stiamo usando il modello di Microsoft
+        is_microsoft_model = "microsoft" in str(type(model))
         
-        # Genera il speech
-        with torch.no_grad():
-            # Passa speaker_embeddings se disponibili/necessari
-            if hasattr(model, "generate_speech"):
-                speech = model.generate_speech(**inputs, **gen_params)
-            else:
-                speech = model.generate(**inputs, **gen_params)
+        if is_microsoft_model:
+            # Usa il modello Microsoft
+            if show_log:
+                print("Utilizzo del modello Microsoft SpeechT5...")
+            
+            # Carica speaker embeddings
+            speaker_embeddings_path = f"https://huggingface.co/datasets/Matthijs/cmu-arctic-xvectors/resolve/main/cmu_us_{speaker_id:02d}_xvector.pt"
+            try:
+                tmp_file = os.path.join(tempfile.gettempdir(), f"speaker_{speaker_id}.pt")
+                download_file(speaker_embeddings_path, tmp_file)
+                speaker_embeddings = torch.load(tmp_file)
+            except:
+                # Usa un embedding predefinito
+                speaker_embeddings = torch.zeros(1, 512)
+            
+            # Crea input IDs dal testo
+            inputs = processor(text=text, return_tensors="pt")
+            
+            # Genera l'audio
+            with torch.no_grad():
+                speech = model.generate_speech(
+                    inputs["input_ids"], 
+                    speaker_embeddings
+                )
+            
+            # Imposta la frequenza di campionamento
+            sample_rate = 16000  # Microsoft usa 16kHz
+        else:
+            # Usa il modello Aurora
+            if show_log:
+                print("Utilizzo del modello Aurora-1.6b-complete...")
+            
+            # Prepara gli input
+            inputs = processor(
+                text=text,
+                language=language,
+                return_tensors="pt"
+            )
+            
+            # Sposta gli input sul dispositivo di calcolo
+            for k, v in inputs.items():
+                if hasattr(v, "to"):
+                    inputs[k] = v.to(model.device)
+            
+            # Genera il speech
+            with torch.no_grad():
+                speech = model.generate(**inputs)
+            
+            # Imposta la frequenza di campionamento
+            sample_rate = SAMPLE_RATE
         
         # Converti il tensore in un array numpy
         speech_array = speech.cpu().numpy().squeeze()
@@ -163,11 +199,13 @@ def text_to_speech(text, language="it", speaker_id=0, speed=1.0, show_log=False)
         if show_log:
             print(f"✅ Audio generato con successo! Lunghezza: {len(speech_array)} campioni")
         
-        return (SAMPLE_RATE, speech_array), None
+        return (sample_rate, speech_array), None
     except Exception as e:
         error_msg = f"Errore nella generazione dell'audio: {str(e)}"
-        print(f"❌ {error_msg}")
-        return None, error_msg
+        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 = [
@@ -184,7 +222,7 @@ with gr.Blocks(title="Aurora-1.6b TTS Demo", theme=gr.themes.Soft()) as demo:
     gr.Markdown("""
     # 🎙️ Aurora-1.6b Text-to-Speech Demo
     
-    Questa demo utilizza il modello Aurora-1.6b-complete per la sintesi vocale (TTS), un modello fine-tuned basato su Dia-1.6B.
+    Questa demo utilizza il modello **Aurora-1.6b-complete** per la sintesi vocale (TTS), un modello fine-tuned basato su Dia-1.6B con pesi completi.
     
     Il modello supporta italiano, inglese, spagnolo, francese e tedesco, ma è stato ottimizzato per l'italiano.
     """)
@@ -222,13 +260,13 @@ with gr.Blocks(title="Aurora-1.6b TTS Demo", theme=gr.themes.Soft()) as demo:
                     info="Valori più bassi = voce più lenta, valori più alti = voce più veloce"
                 )
             
-            debug_input = gr.Checkbox(label="Mostra log di debug", value=False)
+            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)
+            error_output = gr.Textbox(label="Messaggi di errore", visible=True, lines=4)
     
     # Esempi
     gr.Examples(