Troubleshooting Cartesia API Connection in Python TTS Application

import os
import requests
import json
from pydub import AudioSegment
import io
import tkinter as tk
from tkinter import ttk, messagebox
import tempfile
import threading
import re
from dotenv import load_dotenv

# Cartesia imports
from cartesia.tts import CartesiaTTS
from cartesia._types import (
    AudioOutputFormat,
    AudioDataReturnType
)

# Load environment variables
load_dotenv()

class ScriptToConversation:
    def __init__(self):
        print("Initializing application...")
        self.api_key = os.getenv('X-API-Key')
        if not self.api_key:
            tk.messagebox.showerror("API Key Error", "No API key found in .env file")
            raise ValueError("API Key not found in .env file")

        print(f"Using API key: {self.api_key[:8]}...")

        # Initialize Cartesia client
        try:
            # Set the API key as an environment variable
            import httpx  # Import here to help diagnose network issues

            # Validate API key first
            response = httpx.get(
                'https://api.cartesia.ai/voices/',
                headers={
                    'X-API-Key': self.api_key,
                    'Cartesia-Version': '2024-06-10'  # Latest version from the error message
                }
            )

            # If the request is successful, proceed with CartesiaTTS
            if response.status_code == 200:
                os.environ['CARTESIA_API_KEY'] = self.api_key
                os.environ['CARTESIA_BASE_URL'] = 'https://api.cartesia.ai'
                os.environ['CARTESIA_API_VERSION'] = '2024-06-10'  # Set the version for the SDK
                self.client = CartesiaTTS()
                print("Cartesia client initialized successfully")
            else:
                raise ValueError(f"API request failed with status {response.status_code}: {response.text}")

        except Exception as e:
            error_msg = f"Failed to initialize Cartesia client: {e}"
            print(error_msg)
            tk.messagebox.showerror("Initialization Error", error_msg)
            raise

        # Create main window
        self.root = tk.Tk()
        self.root.title("Script to Conversation Converter")
        self.root.geometry("600x800")

        # Script input
        script_frame = ttk.Frame(self.root, padding="10")
        script_frame.pack(fill=tk.BOTH, expand=True)
        ttk.Label(script_frame, text="Paste your script here:").pack()
        self.script_text = tk.Text(script_frame, height=20)
        self.script_text.pack(fill=tk.BOTH, expand=True)

        # Voice selection
        voice_frame = ttk.Frame(self.root, padding="10")
        voice_frame.pack(fill=tk.X)

        # Populate voice selection dropdowns with a variety of voices
        self.available_voices = []

        # Speaker A voice selection
        ttk.Label(voice_frame, text="Speaker A Voice:").grid(row=0, column=0, padx=5)
        self.speaker_a_voice = ttk.Combobox(voice_frame, width=30, values=self.available_voices)
        self.speaker_a_voice.grid(row=0, column=1, padx=5)

        # Speaker B voice selection
        ttk.Label(voice_frame, text="Speaker B Voice:").grid(row=1, column=0, padx=5)
        self.speaker_b_voice = ttk.Combobox(voice_frame, width=30, values=self.available_voices)
        self.speaker_b_voice.grid(row=1, column=1, padx=5)

        # Convert button
        ttk.Button(self.root, text="Convert to Audio", command=self.process_script).pack(pady=10)

        # Status label
        self.status_label = ttk.Label(self.root, text="")
        self.status_label.pack(pady=5)

        print("GUI initialized, loading voices...")
        # Start voice loading in a separate thread
        threading.Thread(target=self.load_voices, daemon=True).start()

    def update_status(self, message):
        """Update status label safely from any thread"""
        if not self.root.winfo_exists():
            return
        self.root.after(0, lambda: self.status_label.config(text=message))

    def load_voices(self):
        try:
            self.update_status("Loading voices...")
            print("Attempting to retrieve voices...")
            # Retrieve available voices
            available_voices = self.client.get_voices()

            # Convert voices to a list of names for the dropdown
            voice_names = list(available_voices.keys())

            # Update the dropdown menu
            self.available_voices = voice_names
            self.speaker_a_voice['values'] = self.available_voices
            self.speaker_b_voice['values'] = self.available_voices

            # Set a default voice if available
            if voice_names:
                self.speaker_a_voice.set(voice_names[0])
                if len(voice_names) > 1:
                    self.speaker_b_voice.set(voice_names[1])

            print(f"Loaded {len(voice_names)} voices successfully")
            self.update_status("Voices loaded successfully")

        except Exception as e:
            error_msg = f"Error loading voices: {e}"
            print(error_msg)
            print(f"Base URL: {os.environ.get('CARTESIA_BASE_URL')}")
            print(f"API Key: {self.api_key[:8]}...")
            tk.messagebox.showerror("Voice Loading Error", error_msg)

    def generate_audio(self, text, voice_name):
        """Generate audio using Cartesia API with enhanced error handling"""
        try:
            print(f"Generating audio for text: {text}")

            # Extract voice ID from the voice name (assuming format like "Voice Name (voice_id)")
            match = re.search(r'\(([^)]+)\)', voice_name)
            if not match:
                raise ValueError(f"Could not extract voice ID from: {voice_name}")
            voice_id = match.group(1)

            # Use Cartesia's bytes method for audio generation
            try:
                audio_response = self.client.tts.bytes(
                    model_id=voice_id,
                    transcript=text,
                    voice={"id": voice_id},  # Use a dictionary instead of a custom type
                    language="en",
                    output_format=AudioOutputFormat.PCM_44100,
                    duration=None,  # Let the API determine duration
                    return_type=AudioDataReturnType.BYTES
                )

                return {
                    'audio': audio_response,
                    'sampling_rate': 44100
                }

            except Exception as api_err:
                error_msg = f"Cartesia API Error: {api_err}"
                print(error_msg)
                tk.messagebox.showerror("Audio Generation Failed", error_msg)
                return None

        except Exception as e:
            error_msg = f"Error in audio generation: {e}"
            print(error_msg)
            tk.messagebox.showerror("Generation Error", error_msg)
            return None

    def process_script(self):
        """Process the script and generate the conversation audio"""
        def process():
            try:
                self.update_status("Processing script...")

                script = self.script_text.get("1.0", tk.END).strip()

                # Split the script into sections
                lines = script.split('\n')
                current_speaker = None
                current_text = []
                conversation_segments = []

                for line in lines:
                    line = line.strip()
                    if not line:
                        continue

                    if line.startswith("Speaker A:"):
                        if current_speaker and current_text:
                            conversation_segments.append((current_speaker, ' '.join(current_text)))
                        current_speaker = "A"
                        current_text = [line.replace("Speaker A:", "").strip()]
                    elif line.startswith("Speaker B:"):
                        if current_speaker and current_text:
                            conversation_segments.append((current_speaker, ' '.join(current_text)))
                        current_speaker = "B"
                        current_text = [line.replace("Speaker B:", "").strip()]
                    else:
                        current_text.append(line)

                # Add the last segment
                if current_speaker and current_text:
                    conversation_segments.append((current_speaker, ' '.join(current_text)))

                # Generate audio for each segment
                combined_audio = AudioSegment.empty()

                for speaker, text in conversation_segments:
                    voice_name = self.speaker_a_voice.get() if speaker == "A" else self.speaker_b_voice.get()
                    self.update_status(f"Generating audio for Speaker {speaker}...")
                    audio_response = self.generate_audio(text, voice_name)

                    if audio_response:
                        # Convert the audio bytes to AudioSegment
                        audio_segment = AudioSegment(
                            data=audio_response['audio'],
                            sample_width=2,  # 16-bit
                            frame_rate=audio_response['sampling_rate'],
                            channels=1
                        )

                        # Add a small pause between segments
                        if len(combined_audio) > 0:
                            combined_audio += AudioSegment.silent(duration=500)  # 500ms pause

                        combined_audio += audio_segment

                # Save the final audio file
                output_file = "conversation.wav"
                combined_audio.export(output_file, format="wav")
                self.update_status(f"Conversation saved as {output_file}")

            except Exception as e:
                print(f"Error: {str(e)}")
                self.update_status(f"Error: {str(e)}")

        # Run processing in background thread
        threading.Thread(target=process, daemon=True).start()

    def run(self):
        """Start the application"""
        print("Starting application...")
        self.root.mainloop()

if __name__ == "__main__":
    app = ScriptToConversation()
    app.run()