![\"Open](\"https://colab.research.google.com/assets/colab-badge.svg\"){kind=icon}
"
- ]
- },
- {
- "cell_type": "markdown",
- "source": [
- "# **GOATBookLM - The Open Source Dia-1.6B Podcast Generator**\n",
- "\n",
- "This Colab notebook enables you to generate **dual voice** podcast style audio files using Nari labs open source audio model Dia-1.B.\n"
- ],
- "metadata": {
- "id": "Z3FNfkyOEqvV"
- }
- },
- {
- "cell_type": "markdown",
- "source": [
- ""
- ],
- "metadata": {
- "id": "gEpOZ2bjkm9b"
- }
- },
- {
- "cell_type": "markdown",
- "source": [
- "# **How To Use This Notebook:**\n",
- "\n",
- "It is structured as follows...\n",
- "\n",
- "1. First an example of generating a basic short clip with Dia-1.6B using the default randomized voices functionality.\n",
- "\n",
- "2. After that, the notebook allows you choose and load base voices which will allow you to create longer podcast style audio files ENSURING CONSISTENT voices throughout.\n",
- "\n",
- "Beyond this, the notebook also includes:\n",
- "\n",
- "* A Dia formatted podcast script generator from any text source you copy paste in. This makes it super simple to quickly create a script which is optimized for how Dia works giving the best possible output. You also of course have the option to modify the script as you wish.\n",
- "\n",
- "* The script generator allows you to use a variety of OpenAI, Google Gemini or Anthropic models.\n",
- "\n",
- "* The notebook also allows you preview sections of the podcast audio you generate, and redraft/ regenerate parts if needed.\n",
- "\n",
- "* Finally the notebook exports a complete single file which you can listen to as a full podcast recording."
- ],
- "metadata": {
- "id": "WaOuFvJ2kgJ3"
- }
- },
- {
- "cell_type": "markdown",
- "source": [
- "# **Package Installs**\n",
- "\n",
- "π The installs usually take a couple of minutes."
- ],
- "metadata": {
- "id": "pisS2hVthEnC"
- }
- },
- {
- "cell_type": "code",
- "source": [
- "# @title\n",
- "!pip install -q git+https://github.com/nari-labs/dia.git\n",
- "!pip install -q soundfile --quiet\n",
- "!pip install ffmpeg-python --quiet\n",
- "!pip install openai --quiet\n",
- "!pip install openai[voice_helpers] --quiet\n",
- "!pip install tiktoken --quiet\n",
- "!pip install -U -q google-genai\n",
- "!pip install anthropic --quiet"
- ],
- "metadata": {
- "cellView": "form",
- "id": "bUJxu8NThCxM"
- },
- "execution_count": null,
- "outputs": []
- },
- {
- "cell_type": "markdown",
- "source": [
- "# **Set Your API Keys**\n",
- "\n",
- "To use this notebook and build your solution you will need...\n",
- "\n",
- "\n",
- "* A HuggingFace user account and API Key\n",
- "* One of the following: an OpenAI, Anthropic or Google AI Studio user account and API Key (with billing setup)\n",
- "\n",
- "* As pictured below add your respective API keys to your Google Secrets (see the key icon on the left panel) inserting your key in the 'value' field and assigning the following name:\n",
- "\n",
- "* HuggingFace API key name: HF_TOKEN\n",
- "* OpenAI API key name: openai_key\n",
- "* Google AI Studio API key name: GOOGLE_API_KEY\n",
- "* Anthropic API key name: claude_api\n",
- "\n",
- "Once done, run the below cell link text granting key access to the notebook access if prompted...\n",
- "\n"
- ],
- "metadata": {
- "id": "gTiliMT2JGJe"
- }
- },
- {
- "cell_type": "markdown",
- "source": [
- ""
- ],
- "metadata": {
- "id": "pacXtSTJJSf9"
- }
- },
- {
- "cell_type": "code",
- "source": [
- "# @title\n",
- "#Hugging face api connection setup with Colab\n",
- "\n",
- "import requests\n",
- "import os\n",
- "from anthropic import Anthropic\n",
- "from google.colab import userdata\n",
- "from openai import OpenAI\n",
- "from google import genai\n",
- "from google.genai import types\n",
- "\n",
- "#Retrieve the HF API key from Google Colab Secrets\n",
- "hf_token = userdata.get('HF_TOKEN')\n",
- "\n",
- "headers = {\"Authorization\": f\"Bearer {hf_token}\"}\n",
- "\n",
- "#openai_key = userdata.get(\"openai_key\")\n",
- "#if not openai_key:\n",
- "# raise ValueError(\"No OpenAI API key found. Please set openai_key in Colab > 'Table of contents' > 'User snippets'.\")\n",
- "\n",
- "#client = OpenAI(api_key=openai_key)\n",
- "#print(\"β
OpenAI AI API key set\")\n",
- "\n",
- "# Also set it as an environment variable for later use\n",
- "import os\n",
- "#os.environ[\"OPENAI_API_KEY\"] = openai_key\n",
- "\n",
- "# Get API key from Colab secrets\n",
- "#GOOGLE_API_KEY = userdata.get('GOOGLE_API_KEY')\n",
- "\n",
- "# Initialize the Google AI client\n",
- "#client = genai.Client(api_key=GOOGLE_API_KEY)\n",
- "\n",
- "#print(f\"β
Google AI API key set\")\n",
- "\n",
- "\n",
- "\n",
- "# Retrieve API keys from Google Colab Secrets\n",
- "#claude_key = userdata.get('claude_api')\n",
- "\n",
- "#os.environ[\"ANTHROPIC_API_KEY\"] = claude_key\n",
- "\n",
- "# Initialize clients\n",
- "#client = Anthropic(api_key=claude_key)\n",
- "\n",
- "#print(\"β
Anthropic AI API key set\")"
- ],
- "metadata": {
- "id": "_e1qcIwzJK8c"
- },
- "execution_count": 4,
- "outputs": []
- },
- {
- "cell_type": "markdown",
- "source": [
- "# **Generate a Test Recording (This Might Crash The First Time - Read Below!)**\n",
- "\n",
- "π ***This will probably crash on the first attempt - simply run the cell again IF it does*** π\n",
- "\n",
- "π’ ***The first loading of the model is really slow, often takes around 5 minutes, but after that it works much faster*** π’\n",
- "\n",
- "\n",
- "* This code creates a simple soundclip for the following text:\n",
- "\n",
- "```\n",
- "[S1] Welcome back to another episode of AI Unfiltered! Iβm Jamie.\n",
- "\n",
- "[S2] And Iβm Taylor. Today, we have some really exciting news from the text-to-speech frontier.\n",
- "\n",
- "[S1] Thatβs right. Thereβs a new open source TTS model on the block called Dia.\n",
- "```\n",
- "\n",
- "* Note that S1 and S2 refer to the voices of speaker 1 and speaker 2. These voices are randomly generated each time.\n",
- "\n",
- "* Below I will show you how to create consistent voices."
- ],
- "metadata": {
- "id": "m1lOmdYL3MSi"
- }
- },
- {
- "cell_type": "code",
- "source": [
- "# @title\n",
- "# Import libraries\n",
- "import soundfile as sf\n",
- "import numpy as np\n",
- "from dia.model import Dia\n",
- "import os\n",
- "import IPython.display as ipd\n",
- "from google.colab import files\n",
- "import time\n",
- "import torch\n",
- "import gc\n",
- "\n",
- "\n",
- "def clear_gpu_memory():\n",
- " \"\"\"Clear GPU memory cache\"\"\"\n",
- " if torch.cuda.is_available():\n",
- " torch.cuda.empty_cache()\n",
- " gc.collect()\n",
- "\n",
- "def generate_and_play(text, output_filename=\"output.wav\", max_retries=3):\n",
- " # Check if text is provided\n",
- " if not text or not text.strip():\n",
- " print(\"Error: No text provided for generation.\")\n",
- " return None\n",
- "\n",
- " # Check if model exists, if not, load it with retries\n",
- " global model\n",
- " retry_count = 0\n",
- "\n",
- " while retry_count < max_retries:\n",
- " try:\n",
- " if 'model' not in globals() or model is None:\n",
- " print(f\"Model not found. Loading model... (Attempt {retry_count + 1}/{max_retries})\")\n",
- " model = Dia.from_pretrained(\"nari-labs/Dia-1.6B\")\n",
- " print(\"Model loaded successfully!\")\n",
- " break\n",
- " except Exception as e:\n",
- " retry_count += 1\n",
- " print(f\"Error loading model (Attempt {retry_count}/{max_retries}): {e}\")\n",
- " if retry_count < max_retries:\n",
- " print(\"Retrying in 5 seconds...\")\n",
- " time.sleep(5)\n",
- " else:\n",
- " print(\"Failed to load model after all retries.\")\n",
- " return None\n",
- "\n",
- " print(f\"Generating audio for: \\n{text}\\n\")\n",
- "\n",
- " # Clear GPU memory before generation\n",
- " clear_gpu_memory()\n",
- "\n",
- " # Try generation with different settings if CUDA memory error occurs\n",
- " generation_attempts = [\n",
- " {\"use_torch_compile\": True, \"verbose\": True},\n",
- " {\"use_torch_compile\": False, \"verbose\": True},\n",
- " {\"use_torch_compile\": False, \"verbose\": False}\n",
- " ]\n",
- "\n",
- " for attempt_num, gen_params in enumerate(generation_attempts, 1):\n",
- " try:\n",
- " print(f\"Generation attempt {attempt_num}/3 with params: {gen_params}\")\n",
- " start_time = time.time()\n",
- "\n",
- " # Generate audio\n",
- " output = model.generate(text, **gen_params)\n",
- "\n",
- " # Check if output is valid\n",
- " if output is None or len(output) == 0:\n",
- " print(\"Error: Generated audio output is empty.\")\n",
- " continue\n",
- "\n",
- " # Calculate generation time\n",
- " generation_time = time.time() - start_time\n",
- " audio_duration = len(output) / 44100 # Assuming 44.1kHz sample rate\n",
- " speed_ratio = audio_duration / generation_time\n",
- "\n",
- " print(f\"Generation took {generation_time:.2f} seconds for {audio_duration:.2f} seconds of audio\")\n",
- " print(f\"Speed ratio: {speed_ratio:.2f}x real-time\")\n",
- "\n",
- " # Save the audio file\n",
- " sf.write(output_filename, output, 44100)\n",
- "\n",
- " # Display audio player in Colab\n",
- " print(\"Playing audio...\")\n",
- " return ipd.Audio(output_filename, autoplay=True)\n",
- "\n",
- " except RuntimeError as e:\n",
- " if \"out of memory\" in str(e).lower() or \"cuda\" in str(e).lower():\n",
- " print(f\"CUDA memory error on attempt {attempt_num}: {e}\")\n",
- " clear_gpu_memory()\n",
- " if attempt_num < len(generation_attempts):\n",
- " print(\"Clearing GPU memory and trying with different settings...\")\n",
- " time.sleep(2)\n",
- " else:\n",
- " print(\"All generation attempts failed due to memory issues.\")\n",
- " print(\"Try restarting the runtime or using a shorter text.\")\n",
- " return None\n",
- " else:\n",
- " print(f\"Error during audio generation (attempt {attempt_num}): {e}\")\n",
- " if attempt_num < len(generation_attempts):\n",
- " continue\n",
- " else:\n",
- " return None\n",
- " except Exception as e:\n",
- " print(f\"Unexpected error during audio generation (attempt {attempt_num}): {e}\")\n",
- " if attempt_num < len(generation_attempts):\n",
- " continue\n",
- " else:\n",
- " return None\n",
- "\n",
- " return None\n",
- "\n",
- "def load_model_with_retry(max_retries=3):\n",
- " \"\"\"Load the model with retry logic\"\"\"\n",
- " global model\n",
- " retry_count = 0\n",
- "\n",
- " while retry_count < max_retries:\n",
- " try:\n",
- " print(f\"Loading Dia model... This may take a few minutes. (Attempt {retry_count + 1}/{max_retries})\")\n",
- " model = Dia.from_pretrained(\"nari-labs/Dia-1.6B\")\n",
- " print(\"Model loaded successfully!\")\n",
- " return True\n",
- " except Exception as e:\n",
- " retry_count += 1\n",
- " print(f\"Error loading model (Attempt {retry_count}/{max_retries}): {e}\")\n",
- " if retry_count < max_retries:\n",
- " print(\"Retrying in 10 seconds...\")\n",
- " time.sleep(10)\n",
- " else:\n",
- " print(\"Failed to load model after all retries.\")\n",
- " return False\n",
- "\n",
- "# Only run this example if this is the first time loading the code\n",
- "if 'model' not in globals():\n",
- " # Example text - change this to whatever you want\n",
- " example_text = \"\"\"[S1] Welcome back to another episode of AI Unfiltered! Iβm Jamie.\n",
- "[S2] And Iβm Taylor. Today, we have some really exciting news from the text-to-speech frontier.\n",
- "[S1] Thatβs right. Thereβs a new open source TTS model on the block called Dia.\"\"\"\n",
- "\n",
- "\n",
- " # Load the model with retry logic\n",
- " if load_model_with_retry():\n",
- " # Test the function with the example text\n",
- " generate_and_play(example_text)\n",
- " else:\n",
- " print(\"Could not load model. Please check your internet connection and try again.\")\n",
- "else:\n",
- " print(\"Code already loaded. Model is ready to use.\")\n",
- " print(\"Call generate_and_play(your_text) to generate audio.\")\n",
- "\n",
- "# Additional helper function to split long text if memory issues persist\n",
- "def split_text_and_generate(text, max_chars=500, output_prefix=\"output_part\"):\n",
- " \"\"\"Split long text into smaller chunks and generate audio for each\"\"\"\n",
- " if len(text) <= max_chars:\n",
- " return generate_and_play(text)\n",
- "\n",
- " # Split text by sentences or dialogue markers\n",
- " parts = []\n",
- " lines = text.split('\\n')\n",
- " current_part = \"\"\n",
- "\n",
- " for line in lines:\n",
- " if len(current_part + line + '\\n') > max_chars and current_part:\n",
- " parts.append(current_part.strip())\n",
- " current_part = line + '\\n'\n",
- " else:\n",
- " current_part += line + '\\n'\n",
- "\n",
- " if current_part.strip():\n",
- " parts.append(current_part.strip())\n",
- "\n",
- " print(f\"Splitting text into {len(parts)} parts due to length.\")\n",
- " audio_files = []\n",
- "\n",
- " for i, part in enumerate(parts):\n",
- " filename = f\"{output_prefix}_{i+1}.wav\"\n",
- " print(f\"\\n--- Generating part {i+1}/{len(parts)} ---\")\n",
- " result = generate_and_play(part, filename)\n",
- " if result:\n",
- " audio_files.append(filename)\n",
- "\n",
- " return audio_files\n",
- "\n",
- "def play_audio(filename=\"output.wav\"):\n",
- " \"\"\"\n",
- " Play an audio file (supports WAV and MP3)\n",
- "\n",
- " Parameters:\n",
- " filename (str): Path to the audio file to play\n",
- " \"\"\"\n",
- " if not os.path.exists(filename):\n",
- " print(f\"Error: File '{filename}' not found\")\n",
- " return\n",
- "\n",
- " file_extension = filename.split('.')[-1].lower()\n",
- " if file_extension not in ['wav', 'mp3']:\n",
- " print(f\"Warning: File extension '{file_extension}' might not be supported\")\n",
- "\n",
- " print(f\"Playing {filename}...\")\n",
- " return ipd.Audio(filename, autoplay=True)\n",
- "\n",
- "# Example usage:\n",
- "play_audio(\"output.wav\") # Change to your filename if needed"
- ],
- "metadata": {
- "id": "5pAfRg3XyUyY",
- "outputId": "b64cf7d5-c75e-47a0-93f0-c3580310b22b",
- "colab": {
- "base_uri": "https://localhost:8080/"
- }
- },
- "execution_count": 5,
- "outputs": [
- {
- "output_type": "stream",
- "name": "stdout",
- "text": [
- "Loading Dia model... This may take a few minutes. (Attempt 1/3)\n",
- "Error loading model (Attempt 1/3): Error loading model from Hugging Face Hub (nari-labs/Dia-1.6B)\n",
- "Retrying in 10 seconds...\n",
- "Loading Dia model... This may take a few minutes. (Attempt 2/3)\n",
- "Error loading model (Attempt 2/3): Error loading model from Hugging Face Hub (nari-labs/Dia-1.6B)\n",
- "Retrying in 10 seconds...\n",
- "Loading Dia model... This may take a few minutes. (Attempt 3/3)\n",
- "Error loading model (Attempt 3/3): Error loading model from Hugging Face Hub (nari-labs/Dia-1.6B)\n",
- "Failed to load model after all retries.\n",
- "Could not load model. Please check your internet connection and try again.\n",
- "Error: File 'output.wav' not found\n"
- ]
- }
- ]
- },
- {
- "cell_type": "markdown",
- "source": [
- "# **Now Generate A Recording For a Different Text**\n",
- "\n",
- "* Now let's try the model again, but on a different text.\n",
- "* Feel free to modify the text within the quotation marks, but always keep to the S1 / S2 format."
- ],
- "metadata": {
- "id": "ebJgEh3R1Ar5"
- }
- },
- {
- "cell_type": "code",
- "source": [
- "example_text = \"\"\"[S1] Hey everyone, welcome back to the show, I'm Jamie.\n",
- "[S2] And I'm Cody, great to be with you today.\n",
- "[S1] We've got a fascinating new material discovery.\n",
- "[S2] Oh really?\n",
- "[S1] It's big.\n",
- "[S2] What is it?\n",
- "[S1] Penn Engineers found something to pull water from thin air.\n",
- "[S2] From thin air?, \"\"\"\n",
- "\n",
- "generate_and_play(example_text)"
- ],
- "metadata": {
- "id": "bEp9pYcA1ABN"
- },
- "execution_count": null,
- "outputs": []
- },
- {
- "cell_type": "markdown",
- "source": [
- "# **How To Use Dia To Create Longer Outputs with Consistent Voices**\n",
- "\n",
- "* As you may have noticed, Dia uses new voices for each audio generation, making unpredictable\n",
- "\n",
- "* However consistency can be achieved by using its voice cloning capability.\n",
- "\n",
- "* To help you experiment with this I have created some basic voices that you can use for cloned / consistent output - I saved these in a github repo, so you can simply download them and use them in this notebook.\n",
- "\n",
- "* These voices were in fact originally generated by Dia, so there should be no rights / IP issues.\n",
- "\n",
- "* The files from the repo include the voice clips AND txt file transcriptions of what the voices say, which the model uses to replicate the voices."
- ],
- "metadata": {
- "id": "b1VmKJ8u3lfb"
- }
- },
- {
- "cell_type": "markdown",
- "source": [
- "# **Download Example Voices for Cloning**\n",
- "\n",
- "There are 3 types of voices from the repo, these include:\n",
- "\n",
- "1. Two female sounding voices 'f_f_voices.mp3'\n",
- "2. A male and female sounding voice 'm_f_voices.mp3'\n",
- "1. Two male sounding voices 'm_m_voices.mp3'\n",
- "\n",
- "\n",
- "\n"
- ],
- "metadata": {
- "id": "llIqQKlK3wHg"
- }
- },
- {
- "cell_type": "code",
- "source": [
- "# @title\n",
- "import os\n",
- "import subprocess\n",
- "import json\n",
- "import re\n",
- "\n",
- "# Create destination folder\n",
- "os.makedirs('/content/clone', exist_ok=True)\n",
- "\n",
- "# GitHub repository details\n",
- "repo_owner = \"smartaces\"\n",
- "repo_name = \"dia_podcast_generator\"\n",
- "branch = \"main\"\n",
- "\n",
- "# Use GitHub API to get file listing\n",
- "print(f\"Fetching file list from {repo_owner}/{repo_name}...\")\n",
- "api_url = f\"https://api.github.com/repos/{repo_owner}/{repo_name}/git/trees/{branch}?recursive=1\"\n",
- "\n",
- "# Use curl to get the file list (already available in Colab)\n",
- "result = subprocess.run(\n",
- " ['curl', '-s', api_url],\n",
- " capture_output=True,\n",
- " text=True\n",
- ")\n",
- "\n",
- "try:\n",
- " # Parse the JSON response\n",
- " file_data = json.loads(result.stdout)\n",
- "\n",
- " # Filter for MP3 files and sample_text files\n",
- " filtered_files = []\n",
- " for item in file_data.get('tree', []):\n",
- " if item['type'] == 'blob': # It's a file, not a directory\n",
- " path = item['path']\n",
- " if path.endswith('.mp3') or 'voices.txt' in path:\n",
- " filtered_files.append(path)\n",
- "\n",
- " if not filtered_files:\n",
- " print(\"No matching files found in the repository.\")\n",
- " else:\n",
- " print(f\"Found {len(filtered_files)} files to download.\")\n",
- "\n",
- " # Download each file\n",
- " for file_path in filtered_files:\n",
- " filename = os.path.basename(file_path)\n",
- " download_url = f\"https://raw.githubusercontent.com/{repo_owner}/{repo_name}/{branch}/{file_path}\"\n",
- "\n",
- " print(f\"Downloading {filename}...\")\n",
- " subprocess.run(\n",
- " ['wget', '-q', download_url, '-O', f'/content/clone/{filename}'],\n",
- " check=True\n",
- " )\n",
- "\n",
- " print(\"\\nDownloaded files:\")\n",
- " for file_path in filtered_files:\n",
- " print(f\"- {os.path.basename(file_path)}\")\n",
- "\n",
- " print(f\"\\nAll files saved to /content/clone/\")\n",
- " subprocess.run(['ls', '-la', '/content/clone/'])\n",
- "\n",
- "except json.JSONDecodeError:\n",
- " print(\"Error parsing repository data. Using alternate method...\")\n",
- "\n",
- " # Alternative method: Clone repository temporarily and copy files\n",
- " temp_dir = \"/tmp/dia_repo\"\n",
- " subprocess.run(['rm', '-rf', temp_dir], check=True)\n",
- " subprocess.run(['mkdir', '-p', temp_dir], check=True)\n",
- "\n",
- " # Clone repository\n",
- " clone_cmd = f\"git clone https://github.com/{repo_owner}/{repo_name}.git {temp_dir}\"\n",
- " subprocess.run(clone_cmd, shell=True, check=True)\n",
- "\n",
- " # Find and copy MP3 and sample_text.txt files\n",
- " find_cmd = f\"find {temp_dir} -name '*.mp3' -o -name '*voices.txt*'\"\n",
- " files = subprocess.run(find_cmd, shell=True, capture_output=True, text=True).stdout.strip().split('\\n')\n",
- "\n",
- " for file_path in files:\n",
- " if file_path: # Skip empty lines\n",
- " filename = os.path.basename(file_path)\n",
- " copy_cmd = f\"cp '{file_path}' '/content/clone/{filename}'\"\n",
- " subprocess.run(copy_cmd, shell=True, check=True)\n",
- " print(f\"Copied {filename}\")\n",
- "\n",
- " # Clean up\n",
- " subprocess.run(['rm', '-rf', temp_dir], check=True)\n",
- "\n",
- " print(f\"\\nAll files saved to /content/clone/\")\n",
- " subprocess.run(['ls', '-la', '/content/clone/'])"
- ],
- "metadata": {
- "id": "uq8uHg073vKe"
- },
- "execution_count": null,
- "outputs": []
- },
- {
- "cell_type": "markdown",
- "source": [
- "# **Add Your Own Cloned Voices (With Transcripts)**\n",
- "\n",
- "You can also add your own voices by adding an mp3 clip, and the respective text transcription of the recording to this project via the below upload tool:\n",
- "\n",
- "* Use a 10 to 15 second audio clip for best results.\n",
- "* You can see examples of these files in the clone folder of this notebook.\n",
- "* Be sure to use this [S1]/ [S2] format structure for the text transcription file:\n",
- "\n",
- "\n",
- "```\n",
- "[S1] Welcome back to another episode of AI Unfiltered! I'm Jamie.\n",
- "\n",
- "[S2] And I'm Taylor. Today, we have some really exciting news from the text-to-speech frontier.\n",
- "\n",
- "[S1] That's right! There's a new open source TTS model on the block called Dia.\n",
- "\n",
- "[S2] So, Dia is developed by a two-person startup called Nari Labs.\n",
- "```\n",
- "\n",
- "* Note that when using the below tool the .txt transcription file will be renamed to match the name of the mp3 you uploaded, this is necessary for the podcast generator to work."
- ],
- "metadata": {
- "id": "FWfWApneRo8V"
- }
- },
- {
- "cell_type": "code",
- "source": [
- "# @title\n",
- "import ipywidgets as widgets\n",
- "from IPython.display import display, clear_output\n",
- "import IPython.display as ipd\n",
- "import os\n",
- "import json\n",
- "import re\n",
- "from datetime import datetime\n",
- "\n",
- "def natural_sort_key(s):\n",
- " \"\"\"Sort strings with numbers in natural order\"\"\"\n",
- " return [int(text) if text.isdigit() else text.lower() for text in re.split(r'(\\d+)', s)]\n",
- "\n",
- "def get_clone_audio_files():\n",
- " \"\"\"Get a sorted list of audio files from the /content/clone directory\"\"\"\n",
- " import glob\n",
- " mp3_files = glob.glob(\"/content/clone/*.mp3\")\n",
- " wav_files = glob.glob(\"/content/clone/*.wav\")\n",
- " files = mp3_files + wav_files\n",
- " return sorted(files, key=natural_sort_key)\n",
- "\n",
- "def load_text_content(file_path):\n",
- " \"\"\"Load text content from file\"\"\"\n",
- " if not file_path or not os.path.exists(file_path):\n",
- " return \"\"\n",
- "\n",
- " try:\n",
- " with open(file_path, 'r', encoding='utf-8') as f:\n",
- " content = f.read().strip()\n",
- " # Remove triple quotes if they exist\n",
- " if content.startswith('\"\"\"') and content.endswith('\"\"\"'):\n",
- " content = content[3:-3].strip()\n",
- " return content\n",
- " except Exception as e:\n",
- " print(f\"Error reading file {file_path}: {e}\")\n",
- " return \"\"\n",
- "\n",
- "def save_text_content(file_path, content):\n",
- " \"\"\"Save text content to file\"\"\"\n",
- " try:\n",
- " # Create directory if it doesn't exist\n",
- " os.makedirs(os.path.dirname(file_path), exist_ok=True)\n",
- "\n",
- " with open(file_path, 'w', encoding='utf-8') as f:\n",
- " # Save with triple quotes for consistency with your existing format\n",
- " f.write(f'\"\"\"{content}\"\"\"')\n",
- " return True\n",
- " except Exception as e:\n",
- " print(f\"Error saving file {file_path}: {e}\")\n",
- " return False\n",
- "\n",
- "def create_audio_upload_editor():\n",
- " \"\"\"Create the audio upload and editor interface\"\"\"\n",
- "\n",
- " # Create destination folder\n",
- " os.makedirs('/content/clone', exist_ok=True)\n",
- "\n",
- " # Audio file upload widget\n",
- " audio_upload = widgets.FileUpload(\n",
- " accept='.mp3,.wav,.m4a,.ogg', # Accept common audio formats\n",
- " multiple=False,\n",
- " description='Upload Audio:',\n",
- " style={'description_width': 'initial'},\n",
- " layout={'width': '400px'}\n",
- " )\n",
- "\n",
- " # Text file upload widget\n",
- " text_upload = widgets.FileUpload(\n",
- " accept='.txt', # Accept text files\n",
- " multiple=False,\n",
- " description='Upload Transcript:',\n",
- " style={'description_width': 'initial'},\n",
- " layout={'width': '400px'}\n",
- " )\n",
- "\n",
- " # Existing files dropdown\n",
- " existing_files = get_clone_audio_files()\n",
- " existing_options = ['Select existing file...'] + [os.path.basename(f) for f in existing_files]\n",
- "\n",
- " existing_dropdown = widgets.Dropdown(\n",
- " options=existing_options,\n",
- " value='Select existing file...',\n",
- " description='Or Select Existing:',\n",
- " style={'description_width': 'initial'},\n",
- " layout={'width': '400px'}\n",
- " )\n",
- "\n",
- " # Audio player area\n",
- " audio_player_area = widgets.Output()\n",
- "\n",
- " # File info display\n",
- " file_info = widgets.HTML(value=\"Upload an audio file or select an existing one\")\n",
- "\n",
- " # Text editor\n",
- " text_editor = widgets.Textarea(\n",
- " value='',\n",
- " placeholder='Upload or select an audio file to edit its transcript...',\n",
- " description='Edit Transcript:',\n",
- " style={'description_width': 'initial'},\n",
- " layout={'width': '800px', 'height': '300px'}\n",
- " )\n",
- "\n",
- " # Buttons\n",
- " refresh_btn = widgets.Button(\n",
- " description='Refresh Files',\n",
- " button_style='info',\n",
- " icon='refresh',\n",
- " tooltip='Refresh the list of existing audio files'\n",
- " )\n",
- "\n",
- " save_text_btn = widgets.Button(\n",
- " description='Save Transcript',\n",
- " button_style='warning',\n",
- " icon='save',\n",
- " tooltip='Save the transcript to a text file'\n",
- " )\n",
- "\n",
- " delete_btn = widgets.Button(\n",
- " description='Delete Selected File',\n",
- " button_style='danger',\n",
- " icon='trash',\n",
- " tooltip='Delete the currently selected audio and text files'\n",
- " )\n",
- "\n",
- " # Output area for messages\n",
- " output_area = widgets.Output()\n",
- "\n",
- " # Store current file paths\n",
- " current_audio_path = None\n",
- " current_text_path = None\n",
- "\n",
- " def display_audio_and_text(audio_path):\n",
- " \"\"\"Display audio player and load corresponding text\"\"\"\n",
- " nonlocal current_audio_path, current_text_path\n",
- "\n",
- " current_audio_path = audio_path\n",
- " audio_basename = os.path.splitext(audio_path)[0]\n",
- " current_text_path = f\"{audio_basename}.txt\"\n",
- "\n",
- " # Display audio player\n",
- " with audio_player_area:\n",
- " clear_output()\n",
- " print(f\"π΅ Playing: {os.path.basename(audio_path)}\")\n",
- " display(ipd.Audio(audio_path, autoplay=False))\n",
- "\n",
- " # Load corresponding text file\n",
- " text_content = load_text_content(current_text_path)\n",
- " text_editor.value = text_content\n",
- "\n",
- " # Update file info\n",
- " text_exists = os.path.exists(current_text_path)\n",
- " file_info.value = f\"\"\"\n",
- " \n",
- " Audio: {os.path.basename(audio_path)}Upload audio files and their corresponding transcript files from your computer, or select existing ones from /content/clone/. \"\n", - " \"Edit transcripts and save them as text files with the same name as the audio file.
\"\n", - " )\n", - "\n", - " upload_section = widgets.VBox([\n", - " widgets.HTML(\"Enter text to convert into a podcast script formatted for Dia. Choose between Anthropic Sonnet 4, OpenAI GPT-4o or Google Gemini 2.5 Flash, and select your desired script length.
\"))\n", - "display(text_input)\n", - "display(model_dropdown)\n", - "display(length_dropdown)\n", - "display(generate_btn)\n", - "display(output)\n", - "display(widgets.HTML(\"You can edit the script below before splitting it into files so that it can be processed by the AI model.
\"))\n", - "display(script_editor)\n", - "display(split_btn)\n", - "display(split_output)" - ], - "metadata": { - "cellView": "form", - "id": "tRdUUGZzWFBd" - }, - "execution_count": null, - "outputs": [] - }, - { - "cell_type": "markdown", - "source": [ - "# **Batch Text to Voice Generator**" - ], - "metadata": { - "id": "C8c_w9bzYeL-" - } - }, - { - "cell_type": "markdown", - "source": [ - "Now that you have generated your script and confirmed you want to use it for the podcast (via the green button π above), you can now generate your podcast audio.\n", - "\n", - "1. After running the below cell, simply click on the dropdown menu which says 'click here to select a voice to clone'\n", - "\n", - "2. Then select a voice, if you are testing this for the first time I recommend you use the voice m_f_voices.mp3\n", - "\n", - "3. You can of course preview the voices\n", - "\n", - "4. Then select your output format - I recommend wav\n", - "\n", - "5. Then press process all files to start the audio file generation, this may take a few minutes depending on how long your script is.\n", - "\n", - "6. In the advanced parameters tab you can change the CFG (prompt adherence) or speed factor - but doing so can make the model prone to crash. If you want to speed up the voices a little bump them up to 0.93 or 0.94, but usually the default works well. " - ], - "metadata": { - "id": "_h52PLBjXMCz" - } - }, - { - "cell_type": "markdown", - "source": [ - "dia rotate" - ], - "metadata": { - "id": "NUGj-YQ7gL39" - } - }, - { - "cell_type": "code", - "source": [ - "# @title\n", - "import time\n", - "import torch\n", - "import soundfile as sf\n", - "import ipywidgets as widgets\n", - "from IPython.display import display, clear_output\n", - "import IPython.display as ipd\n", - "import numpy as np\n", - "from dia.model import Dia\n", - "import os\n", - "import glob\n", - "import re\n", - "import json\n", - "import threading\n", - "\n", - "# Animation control variables\n", - "animation_running = False\n", - "animation_thread = None\n", - "\n", - "# Default example text - will be replaced by loaded text files\n", - "example_text = \"\"\"[S1] Welcome back to another episode of AI Unfiltered! I'm Jamie.\n", - "[S2] And I'm Taylor. Today, we have some really exciting news from the text-to-speech frontier.\"\"\"\n", - "\n", - "# Default example transcript for voice cloning\n", - "example_transcript = \"\"\"[S1] This is an example of a voice cloning transcript.\n", - "[S2] Make sure this transcript matches exactly what's said in your audio file.\"\"\"\n", - "\n", - "def animate_loading_emojis(output_widget, base_message):\n", - " \"\"\"Animate rotating emojis while generation is running\"\"\"\n", - " global animation_running\n", - " emojis = [\"π€\", \"π\", \"π΅\"] # microphone, speaker, musical note\n", - " emoji_index = 0\n", - "\n", - " while animation_running:\n", - " with output_widget:\n", - " clear_output(wait=True)\n", - " current_emoji = emojis[emoji_index % len(emojis)]\n", - " print(f\"{base_message} {current_emoji}\")\n", - "\n", - " emoji_index += 1\n", - " time.sleep(0.6) # Rotate every 0.6 seconds\n", - "\n", - "def start_loading_animation(output_widget, message):\n", - " \"\"\"Start the loading animation\"\"\"\n", - " global animation_running, animation_thread\n", - " animation_running = True\n", - " animation_thread = threading.Thread(target=animate_loading_emojis, args=(output_widget, message))\n", - " animation_thread.daemon = True\n", - " animation_thread.start()\n", - "\n", - "def stop_loading_animation():\n", - " \"\"\"Stop the loading animation\"\"\"\n", - " global animation_running\n", - " animation_running = False\n", - " if animation_thread:\n", - " animation_thread.join(timeout=1)\n", - "\n", - "def save_clone_config(clone_file, transcript_source, transcript_text=\"\", cfg_scale=3.0, speed_factor=0.92):\n", - " \"\"\"Save the current voice cloning configuration to clone_config.txt\"\"\"\n", - " config = {\n", - " \"clone_file\": clone_file,\n", - " \"transcript_source\": transcript_source,\n", - " \"transcript_text\": transcript_text,\n", - " \"cfg_scale\": cfg_scale,\n", - " \"speed_factor\": speed_factor,\n", - " \"timestamp\": time.strftime(\"%Y-%m-%d %H:%M:%S\")\n", - " }\n", - "\n", - " try:\n", - " with open(\"/content/clone_config.txt\", \"w\") as f:\n", - " json.dump(config, f, indent=2)\n", - " print(f\"β Clone configuration saved: {clone_file} with {transcript_source}\")\n", - " print(f\"β Audio parameters saved: CFG Scale={cfg_scale}, Speed Factor={speed_factor}\")\n", - " return True\n", - " except Exception as e:\n", - " print(f\"β Error saving clone config: {e}\")\n", - " return False\n", - "\n", - "def load_clone_config():\n", - " \"\"\"Load the voice cloning configuration from clone_config.txt\"\"\"\n", - " try:\n", - " if os.path.exists(\"/content/clone_config.txt\"):\n", - " with open(\"/content/clone_config.txt\", \"r\") as f:\n", - " config = json.load(f)\n", - " return config\n", - " else:\n", - " return None\n", - " except Exception as e:\n", - " print(f\"β Error loading clone config: {e}\")\n", - " return None\n", - "\n", - "def natural_sort_key(s):\n", - " \"\"\"\n", - " Sort strings with numbers in natural order (e.g., part_1.txt, part_2.txt, ..., part_10.txt)\n", - " instead of lexicographical order (part_1.txt, part_10.txt, part_2.txt, ...)\n", - " \"\"\"\n", - " return [int(text) if text.isdigit() else text.lower() for text in re.split(r'(\\d+)', s)]\n", - "\n", - "def ensure_directories():\n", - " \"\"\"Create the necessary directories if they don't exist\"\"\"\n", - " os.makedirs(\"/content/texts\", exist_ok=True)\n", - " os.makedirs(\"/content/recordings\", exist_ok=True)\n", - " os.makedirs(\"/content/clone\", exist_ok=True)\n", - "\n", - " # If no text files exist in /content/texts, create a sample file\n", - " if not glob.glob(\"/content/texts/*.txt\"):\n", - " with open(\"/content/texts/part_1.txt\", \"w\") as f:\n", - " f.write(example_text)\n", - "\n", - "def get_text_files():\n", - " \"\"\"Get a sorted list of .txt files from the /content/texts directory\"\"\"\n", - " files = glob.glob(\"/content/texts/*.txt\")\n", - " return sorted(files, key=natural_sort_key)\n", - "\n", - "def get_clone_audio_files():\n", - " \"\"\"Get a sorted list of audio files from the /content/clone directory\"\"\"\n", - " mp3_files = glob.glob(\"/content/clone/*.mp3\")\n", - " wav_files = glob.glob(\"/content/clone/*.wav\")\n", - " files = mp3_files + wav_files\n", - " return sorted(files, key=natural_sort_key)\n", - "\n", - "def get_transcript_for_audio(audio_path):\n", - " \"\"\"\n", - " Attempt to find and load a transcript file for the given audio file.\n", - " The transcript file should have the same name as the audio file but with .txt extension.\n", - "\n", - " Args:\n", - " audio_path: Path to the audio file\n", - "\n", - " Returns:\n", - " The transcript content if found, None otherwise\n", - " \"\"\"\n", - " base_name = os.path.splitext(audio_path)[0]\n", - " transcript_path = f\"{base_name}.txt\"\n", - "\n", - " if os.path.exists(transcript_path):\n", - " try:\n", - " with open(transcript_path, 'r', encoding='utf-8') as f:\n", - " return f.read().strip()\n", - " except Exception as e:\n", - " print(f\"Error reading transcript file {transcript_path}: {e}\")\n", - " return None\n", - " return None\n", - "\n", - "def load_text_from_file(file_path):\n", - " \"\"\"Load text content from a file\"\"\"\n", - " try:\n", - " with open(file_path, 'r', encoding='utf-8') as file:\n", - " return file.read()\n", - " except Exception as e:\n", - " print(f\"Error reading file {file_path}: {e}\")\n", - " return \"\"\n", - "\n", - "def generate_audio(\n", - " text,\n", - " output_path,\n", - " output_format=\"wav\",\n", - " max_tokens=3072,\n", - " cfg_scale=3.0,\n", - " temperature=1.2,\n", - " top_p=0.95,\n", - " cfg_filter_top_k=45,\n", - " speed_factor=0.92,\n", - " seed=None,\n", - " compute_dtype=\"float16\",\n", - " use_torch_compile=True,\n", - " use_voice_clone=False,\n", - " clone_audio_path=None,\n", - " clone_transcript=None\n", - "):\n", - " \"\"\"\n", - " Generate speech using the Dia model with customizable parameters.\n", - "\n", - " Args:\n", - " text: Input text to convert to speech\n", - " output_path: Path to save the output audio file\n", - " output_format: Format to save the audio file (wav or mp3)\n", - " max_tokens: Maximum number of tokens to generate (default 3072 ~ 36 seconds)\n", - " cfg_scale: Guidance scale to adhere to the text prompt (default 3.0)\n", - " temperature: Randomness of generation (default 1.2)\n", - " top_p: Nucleus sampling threshold (default 0.95)\n", - " cfg_filter_top_k: Number of top logits for CFG filtering (default 45)\n", - " speed_factor: Speed adjustment for audio output (default 0.94)\n", - " seed: Random seed for reproducibility (default None)\n", - " compute_dtype: Computation precision (default \"float16\")\n", - " use_torch_compile: Whether to use torch.compile (default True)\n", - " use_voice_clone: Whether to use voice cloning (default False)\n", - " clone_audio_path: Path to the audio file to clone (default None)\n", - " clone_transcript: Transcript of the audio to clone (default None)\n", - "\n", - " Returns:\n", - " Path to the generated audio file\n", - " \"\"\"\n", - " # Ensure numpy is imported in this scope\n", - " import numpy as np\n", - "\n", - " # Check if model exists, if not, load it\n", - " global model\n", - " if 'model' not in globals():\n", - " model = Dia.from_pretrained(\"nari-labs/Dia-1.6B\", compute_dtype=compute_dtype)\n", - "\n", - " start_time = time.time()\n", - "\n", - " # Set seed if provided - implemented according to Dia's official method\n", - " if seed is not None:\n", - " # Set all necessary random seeds for full determinism\n", - " import random\n", - " import numpy as np\n", - "\n", - " random.seed(seed)\n", - " np.random.seed(seed)\n", - " torch.manual_seed(seed)\n", - " if torch.cuda.is_available():\n", - " torch.cuda.manual_seed(seed)\n", - " torch.cuda.manual_seed_all(seed)\n", - " # Ensure deterministic behavior for cuDNN\n", - " torch.backends.cudnn.deterministic = True\n", - " torch.backends.cudnn.benchmark = False\n", - "\n", - " # Generate audio with parameters\n", - " if use_voice_clone and clone_audio_path and clone_transcript:\n", - " # Voice cloning mode\n", - " combined_text = f\"{clone_transcript}\\n{text}\"\n", - " output = model.generate(\n", - " text=combined_text,\n", - " audio_prompt=clone_audio_path,\n", - " max_tokens=max_tokens,\n", - " cfg_scale=cfg_scale,\n", - " temperature=temperature,\n", - " top_p=top_p,\n", - " cfg_filter_top_k=cfg_filter_top_k,\n", - " use_torch_compile=use_torch_compile,\n", - " verbose=False # Set to False to reduce output during batch processing\n", - " )\n", - " else:\n", - " # Standard generation mode\n", - " output = model.generate(\n", - " text=text,\n", - " max_tokens=max_tokens,\n", - " cfg_scale=cfg_scale,\n", - " temperature=temperature,\n", - " top_p=top_p,\n", - " cfg_filter_top_k=cfg_filter_top_k,\n", - " use_torch_compile=use_torch_compile,\n", - " verbose=False # Set to False to reduce output during batch processing\n", - " )\n", - "\n", - " # Apply speed factor if different from 1.0\n", - " if speed_factor != 1.0:\n", - " new_length = int(len(output) / speed_factor)\n", - " indices = np.linspace(0, len(output) - 1, new_length)\n", - " output = np.interp(indices, np.arange(len(output)), output)\n", - "\n", - " # Ensure the output path has the correct extension\n", - " base_output_path = os.path.splitext(output_path)[0]\n", - " temp_wav_file = f\"{base_output_path}.wav\"\n", - "\n", - " # Save the audio file as WAV (required for MP3 conversion)\n", - " sf.write(temp_wav_file, output, 44100)\n", - "\n", - " # If MP3 is selected, convert the WAV file to MP3\n", - " final_output_file = temp_wav_file\n", - " if output_format == \"mp3\":\n", - " try:\n", - " from pydub import AudioSegment\n", - " final_output_file = f\"{base_output_path}.mp3\"\n", - " AudioSegment.from_wav(temp_wav_file).export(final_output_file, format=\"mp3\")\n", - " # Remove the temporary WAV file if MP3 conversion was successful\n", - " if os.path.exists(final_output_file):\n", - " os.remove(temp_wav_file)\n", - " except ImportError:\n", - " final_output_file = temp_wav_file\n", - " except Exception as e:\n", - " final_output_file = temp_wav_file\n", - "\n", - " return final_output_file\n", - "\n", - "def batch_process_files(\n", - " output_format,\n", - " max_tokens,\n", - " cfg_scale,\n", - " temperature,\n", - " top_p,\n", - " cfg_filter_top_k,\n", - " speed_factor,\n", - " seed,\n", - " compute_dtype,\n", - " use_torch_compile,\n", - " use_voice_clone,\n", - " clone_audio_path,\n", - " clone_transcript,\n", - " output_area\n", - "):\n", - " \"\"\"Process all text files in the /content/texts directory\"\"\"\n", - " with output_area:\n", - " clear_output()\n", - "\n", - " # Get all text files\n", - " text_files = get_text_files()\n", - "\n", - " if not text_files:\n", - " print(\"No text files found in /content/texts directory!\")\n", - " return\n", - "\n", - " print(f\"Found {len(text_files)} text files to process.\")\n", - "\n", - " # Determine the base message for animation\n", - " if use_voice_clone and clone_audio_path:\n", - " base_message = f\"Generating audio with voice cloning for {len(text_files)} files...\"\n", - " clone_info = f\"Using voice cloning with: {os.path.basename(clone_audio_path)}\"\n", - " transcript_preview = clone_transcript[:200] + \"...\" if len(clone_transcript) > 200 else clone_transcript\n", - " print(clone_info)\n", - " print(f\"Transcript: {transcript_preview}\")\n", - " elif seed is not None:\n", - " base_message = f\"Generating audio with seed {seed} for {len(text_files)} files...\"\n", - " print(f\"Using seed: {seed} for consistent voices\")\n", - " else:\n", - " base_message = f\"Generating audio for {len(text_files)} files...\"\n", - " print(\"Using random voice generation\")\n", - "\n", - " # Start the loading animation\n", - " start_loading_animation(output_area, base_message)\n", - "\n", - " # Process each file\n", - " try:\n", - " for i, file_path in enumerate(text_files):\n", - " # Extract file name without extension\n", - " file_name = os.path.basename(file_path)\n", - " base_name = os.path.splitext(file_name)[0]\n", - "\n", - " # Update animation message with current file progress\n", - " current_message = f\"Processing file {i+1}/{len(text_files)}: {file_name}...\"\n", - " stop_loading_animation() # Stop current animation\n", - " start_loading_animation(output_area, current_message) # Start with new message\n", - "\n", - " # Load text from file\n", - " text = load_text_from_file(file_path)\n", - " if not text:\n", - " continue\n", - "\n", - " # Generate output path\n", - " output_path = os.path.join(\"/content/recordings\", base_name)\n", - "\n", - " # Generate audio (this is where the actual generation happens)\n", - " audio_path = generate_audio(\n", - " text=text,\n", - " output_path=output_path,\n", - " output_format=output_format,\n", - " max_tokens=max_tokens,\n", - " cfg_scale=cfg_scale,\n", - " temperature=temperature,\n", - " top_p=top_p,\n", - " cfg_filter_top_k=cfg_filter_top_k,\n", - " speed_factor=speed_factor,\n", - " seed=seed,\n", - " compute_dtype=compute_dtype,\n", - " use_torch_compile=use_torch_compile,\n", - " use_voice_clone=use_voice_clone,\n", - " clone_audio_path=clone_audio_path,\n", - " clone_transcript=clone_transcript\n", - " )\n", - "\n", - " finally:\n", - " # Always stop animation when done (success or error)\n", - " stop_loading_animation()\n", - "\n", - " with output_area:\n", - " clear_output()\n", - " print(f\"β Batch processing complete!\")\n", - " print(f\"All {len(text_files)} audio files have been saved to /content/recordings\")\n", - "\n", - " # Show final results\n", - " for i, file_path in enumerate(text_files):\n", - " file_name = os.path.basename(file_path)\n", - " base_name = os.path.splitext(file_name)[0]\n", - "\n", - " # Determine the final file extension\n", - " final_extension = output_format\n", - " final_audio_path = os.path.join(\"/content/recordings\", f\"{base_name}.{final_extension}\")\n", - "\n", - " if os.path.exists(final_audio_path):\n", - " print(f\"[{i+1}] {file_name} β {base_name}.{final_extension}\")\n", - " # Display the audio player for each generated file\n", - " display(ipd.Audio(final_audio_path, autoplay=False))\n", - " else:\n", - " print(f\"[{i+1}] {file_name} β ERROR: File not generated\")\n", - "\n", - "# Create a validator function for numeric input fields\n", - "def create_validator(min_val, max_val, default_val, step=1):\n", - " def validate(change):\n", - " widget = change.owner\n", - " value = change.new\n", - " try:\n", - " # Convert to appropriate numeric type\n", - " if isinstance(step, int):\n", - " num_value = int(value)\n", - " else:\n", - " num_value = float(value)\n", - "\n", - " # Validate range\n", - " if num_value < min_val:\n", - " widget.value = str(min_val)\n", - " elif num_value > max_val:\n", - " widget.value = str(max_val)\n", - " except (ValueError, TypeError):\n", - " # Reset to default if input is invalid\n", - " widget.value = str(default_val)\n", - " return validate\n", - "\n", - "# Silently check and install dependencies\n", - "def check_and_install_dependencies():\n", - " try:\n", - " import pydub\n", - " except ImportError:\n", - " os.system('pip install pydub > /dev/null 2>&1')\n", - "\n", - " # Check if ffmpeg is installed\n", - " if os.system('which ffmpeg > /dev/null 2>&1') != 0:\n", - " os.system('apt-get update > /dev/null 2>&1 && apt-get install -y ffmpeg > /dev/null 2>&1')\n", - "\n", - "# Create the UI interface\n", - "def create_batch_dia_ui():\n", - " # First, ensure the required directories exist\n", - " ensure_directories()\n", - "\n", - " # Get list of text files and clone audio files\n", - " text_files = get_text_files()\n", - " clone_files = get_clone_audio_files()\n", - " file_count = len(text_files)\n", - " clone_count = len(clone_files)\n", - "\n", - " # Load existing clone configuration if it exists\n", - " existing_config = load_clone_config()\n", - "\n", - " # Input directory info\n", - " directory_info = widgets.HTML(\n", - " value=f\"π§ Saved Configuration: {existing_config['clone_file']} with {existing_config['transcript_source']} (saved {existing_config['timestamp']})
\"\n", - " else:\n", - " config_status.value = \"π§ No saved configuration found. Voice settings will be saved when you process files.
\"\n", - "\n", - " # Refresh button for text files\n", - " refresh_button = widgets.Button(\n", - " description='Refresh File Lists',\n", - " button_style='info',\n", - " tooltip='Refresh the list of text and clone files',\n", - " icon='refresh'\n", - " )\n", - "\n", - " def on_refresh_button_clicked(b):\n", - " nonlocal text_files, clone_files\n", - " text_files = get_text_files()\n", - " clone_files = get_clone_audio_files()\n", - " file_count = len(text_files)\n", - " clone_count = len(clone_files)\n", - "\n", - " directory_info.value = f\"Select an audio file to view and edit its corresponding text. \"\n", - " \"You can modify the text and regenerate the audio using your saved voice cloning settings.
\"\n", - " )\n", - "\n", - " controls_row1 = widgets.HBox([audio_dropdown, refresh_btn])\n", - " controls_row2 = widgets.HBox([cfg_scale_input, speed_factor_input])\n", - " controls_row3 = widgets.HBox([save_text_btn, regenerate_btn])\n", - "\n", - " interface = widgets.VBox([\n", - " header,\n", - " description,\n", - " config_display,\n", - " controls_row1,\n", - " audio_player_area,\n", - " file_info,\n", - " text_editor,\n", - " param_help,\n", - " controls_row2,\n", - " controls_row3,\n", - " output_area\n", - " ], layout=widgets.Layout(spacing='10px'))\n", - "\n", - " display(interface)\n", - "\n", - "# Run the editor\n", - "print(\"π Initializing Audio & Text Editor with Voice Cloning...\")\n", - "create_audio_text_editor()" - ], - "metadata": { - "cellView": "form", - "id": "o74IATNjMS1l" - }, - "execution_count": null, - "outputs": [] - }, - { - "cell_type": "markdown", - "source": [ - "# **Final Audio File**\n", - "\n", - "* Now you have checked all of your segment recordings, you can join them together into a single file using the below tool.\n", - "\n", - "* Simple click 'Concatenate Files'\n", - "\n", - "* Then when the process is run you can listen to your recording or download it via the 3 dots on the player.\n", - "\n", - "* Or you can find your recording in the folders section of this project named as... combined_*most_recent_timestamp*.wav" - ], - "metadata": { - "id": "59DNspLaZ2nv" - } - }, - { - "cell_type": "code", - "source": [ - "# @title\n", - "\n", - "import os\n", - "import glob\n", - "import re\n", - "import ipywidgets as widgets\n", - "from IPython.display import display, clear_output, Audio\n", - "import IPython.display as ipd\n", - "from datetime import datetime\n", - "import subprocess\n", - "\n", - "# Install ffmpeg and ffmpeg-python if not already installed\n", - "try:\n", - " import ffmpeg\n", - "except ImportError:\n", - " print(\"Installing ffmpeg-python...\")\n", - " !pip install ffmpeg-python\n", - " import ffmpeg\n", - "\n", - "# Check if ffmpeg is installed\n", - "def check_ffmpeg():\n", - " try:\n", - " !ffmpeg -version > /dev/null 2>&1\n", - " return True\n", - " except:\n", - " print(\"Installing ffmpeg...\")\n", - " !apt-get update && apt-get install -y ffmpeg\n", - " return True\n", - "\n", - "def natural_sort_key(s):\n", - " \"\"\"\n", - " Sort strings with numbers in natural order (e.g., part_1.mp3, part_2.mp3, ..., part_10.mp3)\n", - " instead of lexicographical order (part_1.mp3, part_10.mp3, part_2.mp3, ...)\n", - " \"\"\"\n", - " return [int(text) if text.isdigit() else text.lower() for text in re.split(r'(\\d+)', s)]\n", - "\n", - "def get_audio_files(directory=\"/content/recordings\", file_format=None):\n", - " \"\"\"\n", - " Get a sorted list of audio files from the specified directory\n", - " If file_format is specified, only files with that extension will be returned\n", - " \"\"\"\n", - " if file_format:\n", - " # Get files with the specific extension\n", - " files = glob.glob(os.path.join(directory, f\"*.{file_format}\"))\n", - " else:\n", - " # Get all audio files (mp3 and wav)\n", - " mp3_files = glob.glob(os.path.join(directory, \"*.mp3\"))\n", - " wav_files = glob.glob(os.path.join(directory, \"*.wav\"))\n", - " files = mp3_files + wav_files\n", - "\n", - " return sorted(files, key=natural_sort_key)\n", - "\n", - "def get_audio_duration(file_path):\n", - " \"\"\"Get the duration of an audio file using ffprobe\"\"\"\n", - " try:\n", - " probe = ffmpeg.probe(file_path)\n", - " duration = float(probe['format']['duration'])\n", - " return duration\n", - " except Exception as e:\n", - " print(f\"Error getting duration for {file_path}: {e}\")\n", - " return 0\n", - "\n", - "def get_audio_info(file_path):\n", - " \"\"\"Get information about an audio file using ffprobe\"\"\"\n", - " try:\n", - " probe = ffmpeg.probe(file_path)\n", - " duration = float(probe['format']['duration'])\n", - " bitrate = int(probe['format']['bit_rate']) / 1000 # kbps\n", - " sample_rate = probe['streams'][0]['sample_rate']\n", - " channels = probe['streams'][0]['channels']\n", - "\n", - " return {\n", - " 'duration': duration,\n", - " 'bitrate': bitrate,\n", - " 'sample_rate': sample_rate,\n", - " 'channels': channels\n", - " }\n", - " except Exception as e:\n", - " print(f\"Error getting info for {file_path}: {e}\")\n", - " return {\n", - " 'duration': 0,\n", - " 'bitrate': 0,\n", - " 'sample_rate': '?',\n", - " 'channels': '?'\n", - " }\n", - "\n", - "def concatenate_audio_files_ffmpeg(files, output_path, output_format):\n", - " \"\"\"\n", - " Concatenate audio files using concat demuxer\n", - "\n", - " Args:\n", - " files: List of audio file paths\n", - " output_path: Path to save the concatenated audio\n", - " output_format: Output audio format ('mp3' or 'wav')\n", - "\n", - " Returns:\n", - " Path to the concatenated audio file\n", - " \"\"\"\n", - " if not files:\n", - " print(\"No files to concatenate\")\n", - " return None\n", - "\n", - " try:\n", - " # Create a temporary file list for ffmpeg\n", - " filelist_path = '/tmp/filelist.txt'\n", - " with open(filelist_path, 'w', encoding='utf-8') as filelist:\n", - " for file_path in files:\n", - " # Ensure the path is absolute\n", - " abs_path = os.path.abspath(file_path)\n", - " filelist.write(f\"file '{abs_path}'\\n\")\n", - "\n", - " # Set output codec based on format\n", - " if output_format == 'mp3':\n", - " output_codec = 'libmp3lame'\n", - " output_opts = {\n", - " 'audio_bitrate': '192k',\n", - " 'vn': None,\n", - " 'map_metadata': '-1' # Remove metadata\n", - " }\n", - " else: # wav\n", - " output_codec = 'pcm_s16le'\n", - " output_opts = {\n", - " 'vn': None,\n", - " 'map_metadata': '-1' # Remove metadata\n", - " }\n", - "\n", - " print(f\"Starting concatenation of {len(files)} files...\")\n", - "\n", - " # Use ffmpeg to concatenate the files\n", - " (\n", - " ffmpeg\n", - " .input(filelist_path, format='concat', safe=0)\n", - " .output(output_path, acodec=output_codec, **output_opts)\n", - " .overwrite_output()\n", - " .run(quiet=True, capture_stdout=True, capture_stderr=True)\n", - " )\n", - "\n", - " # Clean up the filelist\n", - " os.remove(filelist_path)\n", - "\n", - " print(f\"Concatenation successful: {output_path}\")\n", - "\n", - " # Get file size\n", - " file_size_mb = os.path.getsize(output_path) / (1024 * 1024)\n", - " print(f\"Output file size: {file_size_mb:.2f} MB\")\n", - "\n", - " # Get duration\n", - " duration = get_audio_duration(output_path)\n", - " print(f\"Total duration: {duration:.2f} seconds ({duration/60:.2f} minutes)\")\n", - "\n", - " return output_path\n", - "\n", - " except Exception as e:\n", - " print(f\"Error during concatenation: {e}\")\n", - " if os.path.exists(filelist_path):\n", - " os.remove(filelist_path)\n", - " return None\n", - "\n", - "def create_concatenator_ui():\n", - " \"\"\"Create a user interface for the audio concatenation tool\"\"\"\n", - " # Make sure ffmpeg is installed\n", - " check_ffmpeg()\n", - "\n", - " # Make sure the recordings directory exists\n", - " os.makedirs(\"/content/recordings\", exist_ok=True)\n", - "\n", - " # Get available audio files\n", - " audio_files = get_audio_files()\n", - " file_count = len(audio_files)\n", - "\n", - " # Directory information\n", - " directory_info = widgets.HTML(\n", - " value=f\"