Spaces:
Paused
Paused
File size: 5,873 Bytes
ee703ac | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 | import asyncio
import websockets
import pyaudio
import threading
import logging
import json
import time
import struct
import openai
from fastapi import FastAPI, WebSocket
from fastapi.responses import HTMLResponse
from openai import OpenAI
from dotenv import load_dotenv
import os
from fastapi.middleware.cors import CORSMiddleware
from speech import record_audio
from fastapi import FastAPI, File, UploadFile,Form
from fastapi.responses import JSONResponse
load_dotenv()
client = OpenAI()
OpenAI_API_KEY = os.getenv("OPENAI_API_KEY")
# Audio configuration
FORMAT = pyaudio.paInt16
CHANNELS = 1
RATE = 16000
CHUNK = 1024
# Initialize FastAPI
app = FastAPI()
logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(levelname)s - %(message)s')
app.add_middleware( CORSMiddleware, allow_origins=["http://localhost:3000"], # Allow requests from this origin
allow_credentials=True,
allow_methods=["*"],
allow_headers=["*"],
)
chat_history = []
# OpenAI API key
openai.api_key = OpenAI_API_KEY
@app.get("/api-key")
def get_api_key():
return {"API_KEY": os.getenv("OPENAI_API_KEY")}
@app.post("/upload")
async def upload_file(file: UploadFile = File(...)):
try:
contents = await file.read()
with open("audio.wav", "wb") as f:
f.write(contents) # Process the audio file with Whisper model
text = process_audio_with_whisper("audio.wav") # Generate response with GPT-4.0
if "generate an image" in text.lower():
image_url = generate_image_with_dalle(text)
chat_history.append({"type": "image", "content": image_url})
return JSONResponse(content={"image_url": image_url})
else:
response = generate_response_with_gpt4(text)
chat_history.append({"type": "text", "content": response})
return JSONResponse(content={"response": response})
except Exception as e:
logging.error(f"Error processing file: {e}")
return JSONResponse(content={"error": str(e)}, status_code=500)
@app.post("/text-input")
async def text_input(prompt: str = Form(...)):
try: # Determine if the user is asking for an image
if "generate an image" in prompt.lower() or "generate a realistic image" in prompt.lower():
image_url = generate_image_with_dalle(prompt)
chat_history.append({"type": "image", "content": image_url})
return JSONResponse(content={"image_url": image_url})
else: response = generate_response_with_gpt4(prompt)
chat_history.append({"type": "text", "content": response})
return JSONResponse(content={"response": response})
except Exception as e:
logging.error(f"Error processing text input: {e}")
return JSONResponse(content={"error": str(e)}, status_code=500)
@app.post("/image-url-input")
async def image_input(url: str = Form(...), prompt: str = Form(...)):
try:
image_url = url
response = process_image_with_gpt4(image_url, prompt)
chat_history.append({"type": "text", "content": response})
return JSONResponse(content={"response": response})
except Exception as e:
logging.error(f"Error processing image input: {e}")
return JSONResponse(content={"error": str(e)}, status_code=500)
@app.get("/chat-history")
async def get_chat_history():
return JSONResponse(content={"chat_history": chat_history})
filepath = "audio.wav"
def process_audio_with_whisper(filepath): # Save the audio data to a file
# with open("audio.wav", "wb") as f:
# f.write(audio_data) # Transcribe the audio file using OpenAI's Whisper model
try:
audio_file= open(filepath, "rb")
transcription = client.audio.transcriptions.create(
model="whisper-1",
file=audio_file,
)
print(transcription.text)
return transcription.text
except Exception as e:
logging.error(f"Error transcribing audio: {e}")
raise
def generate_response_with_gpt4(text):
try:
completion = client.chat.completions.create(
model="gpt-4-turbo",
messages=[
{"role": "system", "content": "You are a helpful assistant."},
{
"role": "user",
"content": text
}
]
)
print(completion.choices[0].message.content)
return completion.choices[0].message.content
except Exception as e:
logging.error(f"Error generating response: {e}")
raise
# response.choices[0].text.strip()
def generate_image_with_dalle(prompt):
response = client.images.generate(
model="dall-e-3",
prompt=prompt,
size="1024x1024",
quality="hd",
n=1,
)
return response.data[0].url
def process_image_with_gpt4(url,text):
try:
completion = client.chat.completions.create(
model="gpt-4o",
messages=[
{
"role": "user",
"content": [
{"type": "text", "text": text},
{
"type": "image_url",
"image_url": {
"url": url,
}
},
],
}
],
)
return completion.choices[0].message.content
except Exception as e:
logging.error(f"Error processing image: {e}")
raise
if __name__ == "__main__":
import uvicorn
uvicorn.run(app, host="0.0.0.0", port=8000)
|