|
from botocore.exceptions import BotoCoreError, ClientError |
|
from contextlib import closing |
|
import os |
|
import urllib |
|
import boto3 |
|
import gradio as gr |
|
from openai import OpenAI |
|
import os |
|
import time |
|
import json |
|
from dotenv import load_dotenv |
|
|
|
|
|
|
|
load_dotenv(".env") |
|
api_key = os.getenv("OPENAI_API_KEY") |
|
client = OpenAI(api_key=api_key) |
|
access_key = os.getenv('access_key') |
|
secret_key = os.getenv('secret_key') |
|
|
|
|
|
def recognize_from_microphone(file_info): |
|
if not file_info: |
|
return "No audio file received.", "" |
|
file_path = file_info |
|
print(f"File path received: {file_path}") |
|
|
|
|
|
if not os.path.exists(file_path): |
|
return f"File not found: {file_path}", "" |
|
|
|
|
|
transcribe_client = boto3.client('transcribe', region_name='us-east-1', aws_access_key_id=access_key, aws_secret_access_key=secret_key) |
|
s3_client = boto3.client('s3', region_name='us-west-2', aws_access_key_id=access_key, aws_secret_access_key=secret_key) |
|
bucket_name = 'nutrition-bot' |
|
object_name = os.path.basename(file_path) |
|
|
|
|
|
s3_client.upload_file(file_path, bucket_name, object_name) |
|
job_name = f"TranscriptionJob-{int(time.time())}" |
|
job_uri = f"s3://{bucket_name}/{object_name}" |
|
|
|
|
|
transcribe_client.start_transcription_job( |
|
TranscriptionJobName=job_name, |
|
Media={'MediaFileUri': job_uri}, |
|
MediaFormat='mp3', |
|
LanguageCode='en-US' |
|
) |
|
|
|
|
|
while True: |
|
status = transcribe_client.get_transcription_job(TranscriptionJobName=job_name) |
|
if status['TranscriptionJob']['TranscriptionJobStatus'] in ['COMPLETED', 'FAILED']: |
|
break |
|
time.sleep(5) |
|
|
|
|
|
if status['TranscriptionJob']['TranscriptionJobStatus'] == 'COMPLETED': |
|
transcript_uri = status['TranscriptionJob']['Transcript']['TranscriptFileUri'] |
|
transcript_response = urllib.request.urlopen(transcript_uri) |
|
transcript_data = json.load(transcript_response) |
|
transcript_text = transcript_data['results']['transcripts'][0]['transcript'] |
|
return transcript_text, "" |
|
return "Failed to transcribe audio.", "" |
|
|
|
|
|
def synthesize_speech(text, filename="output.mp3"): |
|
"""Converts text to speech using Amazon Polly and saves it to an MP3 file.""" |
|
|
|
polly_client = boto3.client('polly', region_name='us-east-1', aws_access_key_id=access_key, aws_secret_access_key=secret_key) |
|
|
|
|
|
response = polly_client.synthesize_speech( |
|
Text=text, |
|
OutputFormat='mp3', |
|
VoiceId='Salli' |
|
) |
|
|
|
|
|
if "AudioStream" in response: |
|
with open(filename, 'wb') as file: |
|
file.write(response['AudioStream'].read()) |
|
print(f"Speech synthesized for text [{text}] and saved to {filename}") |
|
else: |
|
print(f"Failed to synthesize speech for text [{text}]") |
|
|
|
return filename |
|
|
|
|
|
|
|
def chatbot_response(user_input="", gender=None, plan_type=None, weight=None, height=None, audio_input=None): |
|
transcription, response = "", "" |
|
error_message = "" |
|
|
|
if audio_input: |
|
transcription, error = recognize_from_microphone(audio_input) |
|
if error: |
|
error_message = error |
|
else: |
|
user_input = transcription |
|
|
|
|
|
if not user_input.strip() and not transcription.strip() and not error_message: |
|
error_message = "Please provide audio input or type your question." |
|
|
|
if error_message: |
|
return error_message, "" |
|
|
|
|
|
detailed_input = f"User details - Gender: {gender}, Plan Type: {plan_type}, Weight: {weight} kg, Height: {height} cm. Question: {user_input}" |
|
try: |
|
completion = client.chat.completions.create( |
|
model="gpt-3.5-turbo", |
|
messages=[ |
|
{"role": "system", "content": "You are a nutrition consultant AI, capable of providing natural diet plans and emergency assistance based on user inputs."}, |
|
{"role": "user", "content": detailed_input}, |
|
] |
|
) |
|
response = completion.choices[0].message.content |
|
if response: |
|
audio_path = synthesize_speech(response) |
|
return transcription, response, audio_path |
|
except Exception as e: |
|
return transcription, f"An error occurred during response generation: {str(e)}" |
|
|
|
|
|
def emergency_assistance(query): |
|
if not query.strip(): |
|
return "Please provide a query for emergency assistance." |
|
try: |
|
completion = client.chat.completions.create( |
|
model="gpt-3.5-turbo", |
|
messages=[ |
|
{"role": "system", "content": "As an AI serving as an emergency nutrition advisor, your objective is to provide prompt and accurate nutritional guidance in urgent situations. When users present their concerns, you should deliver tailored advice that addresses the critical aspects of their nutritional needs quickly and effectively. Focus on offering clear, practical, and context-specific solutions to ensure their immediate dietary requirements are met."}, |
|
{"role": "user", "content": query}, |
|
] |
|
) |
|
response = completion.choices[0].message.content |
|
except Exception as e: |
|
return f"An error occurred: {str(e)}" |
|
|
|
if response: |
|
audio_path = synthesize_speech(response) |
|
return response, audio_path |
|
|
|
|
|
|
|
interface1 = gr.Interface( |
|
fn=chatbot_response, |
|
inputs=[ |
|
gr.Textbox(lines=5, label="Input Here", placeholder="Type or say your question here..."), |
|
gr.Radio(choices=["Male", "Female", "Other"], label="Gender"), |
|
gr.Radio(choices=["Weight Gain", "Weight Loss"], label="Plan Type"), |
|
gr.Number(label="Weight (kg)", info="Enter your weight in kg"), |
|
gr.Number(label="Height (cm)", info="Enter your height in cm"), |
|
gr.Audio(type="filepath", label="Record your question") |
|
], |
|
outputs=[ |
|
gr.Text(label="Transcription"), |
|
gr.Text(lines=10, label="Response"), |
|
gr.Audio(label="Listen to Response") |
|
], |
|
title="Personalized Nutrition AI Advisor", |
|
description="Ask me anything about nutrition. Provide your Gender, Plan Type, Weight and Height for personalized advice." |
|
) |
|
|
|
interface2 = gr.Interface( |
|
fn=emergency_assistance, |
|
inputs=[gr.Textbox(lines=10, label="Query", placeholder="Enter your emergency nutrition query here...")], |
|
outputs=[ |
|
gr.Text(lines=10, label="Response"), |
|
gr.Audio(label="Listen to Response") |
|
], |
|
title="Emergency Assistance", |
|
description="To better assist you, could you explain what led to this emergency?" |
|
) |
|
|
|
|
|
|
|
app = gr.TabbedInterface([interface1, interface2], ["Nutrition Consultant", "Emergency Assistance"], title="HealthyBytes: Your AI Nutrition Consultant") |
|
|
|
if __name__ == "__main__": |
|
app.launch(share=False) |