Browse Source

Merge pull request #45 from innovoe/main

TTS and STT
pull/42/merge
Ed Donner 5 months ago committed by GitHub
parent
commit
a8ddcfe1a9
No known key found for this signature in database
GPG Key ID: B5690EEEBB952194
  1. 196
      week2/community-contributions/TTS_STT.ipynb

196
week2/community-contributions/TTS_STT.ipynb

@ -0,0 +1,196 @@
{
"cells": [
{
"cell_type": "code",
"execution_count": null,
"id": "a60e0f78-4637-4318-9ab6-309c3f7f2799",
"metadata": {},
"outputs": [],
"source": [
"import os\n",
"import json\n",
"from dotenv import load_dotenv\n",
"from openai import OpenAI\n",
"\n",
"load_dotenv()\n",
"\n",
"openai_api_key = os.getenv('OPENAI_API_KEY')\n",
"if openai_api_key:\n",
" print(\"API Key set\")\n",
"else:\n",
" print(\"OpenAI API Key not set\")\n",
" \n",
"MODEL = \"gpt-4o-mini\"\n",
"openai = OpenAI()"
]
},
{
"cell_type": "code",
"execution_count": null,
"id": "67026ef0-23be-4101-9371-b11f96f505bf",
"metadata": {},
"outputs": [],
"source": [
"# TTS\n",
"\n",
"from pydub import AudioSegment\n",
"import os\n",
"import subprocess\n",
"from io import BytesIO\n",
"import tempfile\n",
"\n",
"# Set custom temp directory\n",
"custom_temp_dir = r\"D:\\projects\\llm_engineering-main\\temp\"\n",
"os.makedirs(custom_temp_dir, exist_ok=True)\n",
"\n",
"# Explicitly set FFmpeg paths\n",
"AudioSegment.converter = r\"D:\\Anaconda3\\envs\\llms\\Library\\bin\\ffmpeg.exe\"\n",
"AudioSegment.ffprobe = r\"D:\\Anaconda3\\envs\\llms\\Library\\bin\\ffprobe.exe\"\n",
"\n",
"def play_audio_with_ffplay(audio_segment, temp_dir):\n",
" # Explicitly create and manage a temporary file\n",
" temp_file_path = os.path.join(temp_dir, \"temp_output.wav\")\n",
" \n",
" # Export the audio to the temporary file\n",
" audio_segment.export(temp_file_path, format=\"wav\")\n",
" \n",
" try:\n",
" # Play the audio using ffplay\n",
" subprocess.call([\"ffplay\", \"-nodisp\", \"-autoexit\", temp_file_path])\n",
" finally:\n",
" # Clean up the temporary file after playback\n",
" if os.path.exists(temp_file_path):\n",
" os.remove(temp_file_path)\n",
"\n",
"def talker(message):\n",
" # Mocked OpenAI response for testing\n",
" response = openai.audio.speech.create(\n",
" model=\"tts-1\",\n",
" voice=\"nova\",\n",
" input=message\n",
" )\n",
" \n",
" # Handle audio stream\n",
" audio_stream = BytesIO(response.content)\n",
" audio = AudioSegment.from_file(audio_stream, format=\"mp3\")\n",
" \n",
" # Play the audio\n",
" play_audio_with_ffplay(audio, custom_temp_dir)"
]
},
{
"cell_type": "code",
"execution_count": null,
"id": "12c66b44-293a-4bf9-b81e-0f6905fbf607",
"metadata": {},
"outputs": [],
"source": [
"# STT Whisper\n",
"\n",
"import whisper\n",
"import sounddevice as sd\n",
"import numpy as np\n",
"from scipy.io.wavfile import write\n",
"\n",
"def record_audio(temp_dir, duration=5, samplerate=16000, device_id=2):\n",
" # print(f\"Recording for {duration} seconds...\")\n",
" sd.default.device = (device_id, None)\n",
" audio = sd.rec(int(duration * samplerate), samplerate=samplerate, channels=1, dtype=\"int16\")\n",
" sd.wait() # Wait until the recording is finished\n",
" \n",
" audio_path = os.path.join(temp_dir, \"mic_input.wav\")\n",
" write(audio_path, samplerate, audio)\n",
" # print(f\"Audio recorded and saved to {audio_path}\")\n",
"\n",
" return audio_path\n",
"\n",
"\n",
"whisper_model = whisper.load_model(\"base\")\n",
"def transcribe_audio(audio_path): \n",
" # print(\"Transcribing audio...\")\n",
" result = whisper_model.transcribe(audio_path, language=\"en\")\n",
" return result[\"text\"]\n",
"\n",
"def mic_to_text():\n",
" audio_path = record_audio(custom_temp_dir, duration=10)\n",
" transcription = transcribe_audio(audio_path)\n",
" # print(f\"Transcription: {transcription}\")\n",
" return transcription"
]
},
{
"cell_type": "code",
"execution_count": null,
"id": "0156c106-1844-444a-9a22-88c3475805d9",
"metadata": {},
"outputs": [],
"source": [
"# Chat Functions\n",
"\n",
"import requests\n",
"history = [{\"role\": \"system\", \"content\": \"You are Nova the friendly robot. Reply within couple of sentences.\"}]\n",
"\n",
"def run_chat():\n",
" running = True\n",
" while running:\n",
" input_text = input(\"press Enter to talk\") \n",
" user_input = input_text if input_text.strip() else mic_to_text()\n",
" running = False if input_text == \"bye\" or user_input.strip() == \"bye\" else True\n",
" print(f\"\\nYou: {user_input}\\n\\n\")\n",
" history.append({\"role\": \"user\", \"content\": user_input}) \n",
" api_run = requests.post(\n",
" \"http://localhost:11434/api/chat\", \n",
" json={\n",
" \"model\": \"llama3.2\",\n",
" \"messages\": history,\n",
" \"stream\": False\n",
" }, \n",
" headers={\"Content-Type\": \"application/json\"}\n",
" )\n",
" output_message = api_run.json()['message']['content']\n",
" print(f\"Nova: {output_message}\\n\\n\") \n",
" talker(output_message)\n",
" history.append({\"role\": \"assistant\", \"content\": output_message})"
]
},
{
"cell_type": "code",
"execution_count": null,
"id": "de61b54e-387e-4480-a592-c78e3245ddde",
"metadata": {},
"outputs": [],
"source": [
"run_chat()"
]
},
{
"cell_type": "code",
"execution_count": null,
"id": "ce16bee7-6ea6-46d5-a407-385e6ae31db8",
"metadata": {},
"outputs": [],
"source": []
}
],
"metadata": {
"kernelspec": {
"display_name": "Python 3 (ipykernel)",
"language": "python",
"name": "python3"
},
"language_info": {
"codemirror_mode": {
"name": "ipython",
"version": 3
},
"file_extension": ".py",
"mimetype": "text/x-python",
"name": "python",
"nbconvert_exporter": "python",
"pygments_lexer": "ipython3",
"version": "3.11.11"
}
},
"nbformat": 4,
"nbformat_minor": 5
}
Loading…
Cancel
Save