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 # Loads and set environment variables 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}") # Check file existence if not os.path.exists(file_path): return f"File not found: {file_path}", "" # Configuring Amazon Transcribe 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' # Specify your S3 bucket name object_name = os.path.basename(file_path) # Upload file to S3 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}" # Start transcription job transcribe_client.start_transcription_job( TranscriptionJobName=job_name, Media={'MediaFileUri': job_uri}, MediaFormat='mp3', # or your file format, e.g., wav LanguageCode='en-US' ) # Checking job status while True: status = transcribe_client.get_transcription_job(TranscriptionJobName=job_name) if status['TranscriptionJob']['TranscriptionJobStatus'] in ['COMPLETED', 'FAILED']: break time.sleep(5) # Process the transcription result 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.""" # Create a Polly client polly_client = boto3.client('polly', region_name='us-east-1', aws_access_key_id=access_key, aws_secret_access_key=secret_key) # Synthesize speech response = polly_client.synthesize_speech( Text=text, OutputFormat='mp3', # MP3 output format VoiceId='Salli' # Using Joanna voice, you can choose another ) # Accessing the audio stream from the response 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 = "", "" # Initialize variables for transcription and response error_message = "" # Initialize error_message at the start of the function if audio_input: transcription, error = recognize_from_microphone(audio_input) if error: error_message = error # Capture the error to return it properly else: user_input = transcription # Use the transcription if there's no error # Check if there's user input or transcription, and there's no error message 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, "" # Return the error with an empty second value # Process user_input as before (assuming previous code handled it) 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 # Return audio path along with text and transcription except Exception as e: return transcription, f"An error occurred during response generation: {str(e)}" # Return both values even in case of an exception 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)}" # After generating response: if response: audio_path = synthesize_speech(response) return response, audio_path # Return both response text and audio path # Adjust Gradio interfaces to include audio output 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") # New audio output for the synthesized speech ], 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") # New audio output for the synthesized speech ], title="Emergency Assistance", description="To better assist you, could you explain what led to this emergency?" ) # Combined interface with tabs app = gr.TabbedInterface([interface1, interface2], ["Nutrition Consultant", "Emergency Assistance"], title="HealthyBytes: Your AI Nutrition Consultant") if __name__ == "__main__": app.launch(share=False)