main_assistant.py

import os
import subprocess
import sys
import logging
import webbrowser
import time
from datetime import datetime
from voice_recognition import recognize_speech
from text_to_speech import speak
from wikipedia_info import search_wikipedia
from weather_info import get_weather
from news_info import get_news
from jokes import tell_joke
from open_app import open_application
from open_files import *
from gemini_info import get_gemini_response
from song_data import SONGS
from website_data import WEBSITES
from collections import defaultdict
from reminderr import *
import threading
import json
from email_sender import send_bulk_email


USER_MAIL, USER_PASS, RECEPTIONISTS = "YOUR_EMAIL", "YOUR_PASSWORD", ["List of receptionists"]



### MAIN

""" 

Guidelines for new ability/feature contributions

1. Please create your code into a separate file or a separate function. 
2. Define your functions's inputs properly.
3. Add your function's tool name to the following TOOLS list, add a description, inputs and action : ( your function name )
4. list your tool properly in the last prompt in the main function to get inputs from gemini.
%% Please be as specific as possible and describe as good as poosible about your inputs, their formats 
%% and the function's intention when to be called to get a proper output.

Thank You.

"""


def recognize_speech(timeout=10):
    # Simulate speech recognition
    return input("You: ")

def speak(message):
    print(f"Assistant: {message}")

def just_Say(msg):
    logging.info(msg)
    speak(msg)

def abort(message):
    global RUNNING
    RUNNING = False
    print(message)

def openfile(file):
    logging.info("openinig....", file)
    return open_file(file)

def play_music(song_name=""):
    if song_name == "":
        response = "Which song would you like to play?"
        logging.info(response)
        speak(response)
        song_name = recognize_speech()
    
    if song_name:
        youtube_url = SONGS.get(song_name.lower())
        if youtube_url:
            response = f"Playing {song_name} in your browser."
            logging.info(response)
            speak(response)
            webbrowser.open(youtube_url)
        else:
            response = "Song not found in the database."
            logging.warning(response)
            speak(response)

def open_website(website_name = ""):
    if website_name == "":
        response = "Which website would you like to open?"
        logging.info(response)
        speak(response)
        website_name = recognize_speech()
    if website_name:
        website_url = WEBSITES.get(website_name.lower())
        if website_url:
            response = f"Opening {website_name} in your browser."
            logging.info(response)
            speak(response)
            webbrowser.open(website_url)
        else:
            response = "Website not found in the database.Searching in google."
            logging.warning(response)
            speak(response)
            from googlesearch import search
            webs = search(website_name, advanced=True)
            webbrowser.open(list(webs)[0].url)
            

TOOLS = [
    {
        'name': 'just_Say',
        'description': '',
        'inputs': ['msg'],
        'action': just_Say
    },
    {
        'name': 'get_news',
        'description': 'Get news',
        'inputs': [],
        'action': get_news
    },
    {
        'name': 'set_reminder',
        'description': 'Sets a reminder based on user query.',
        'inputs': ['input_text'],
        'action': extract_reminder
    },
    {
        'name': 'search_wikipedia',
        'description': 'Searches Wikipedia for the given query.',
        'inputs': ['query'],
        'action': search_wikipedia
    },
    {
        'name': 'get_weather',
        'description': 'Fetches the weather information for the specified city.',
        'inputs': ['city_name'],
        'action': get_weather
    },
    {
        'name': 'tell_joke',
        'description': 'Tells a random joke.',
        'inputs': [],
        'action': tell_joke
    },
    {
        'name': 'play_music',
        'description': 'Plays music based on the song name provided.',
        'inputs': ['song_name'],
        'action': play_music
    },
    {
        'name': 'open_website',
        'description': 'Opens a website in the browser.',
        'inputs': ['website_name'],
        'action': open_website
    },
    {
        'name': 'open_application',
        'description': 'Opens an Application',
        'inputs': ['app_name'],
        'action': open_application
    },
    {
        'name': 'abort',
        'description': 'Aborts',
        'inputs': ['message'],
        'action': abort
    },
    {
        'name': 'openfile',
        'description': 'Opens files',
        'inputs': ['file'],
        'action': openfile
    },
    
]


## MAIN

global REMINDERS
global RUNNING

REMINDERS = defaultdict(list)


# Setup logging
logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(levelname)s - %(message)s')


def get_greeting():
    """Get the appropriate greeting based on the current time."""
    current_hour = datetime.now().hour
    if current_hour < 12:
        return "Good morning"
    elif 12 <= current_hour < 18:
        return "Good afternoon"
    else:
        return "Good evening"

def check_reminders():
    global REMINDERS
    while RUNNING:
        current_time = datetime.now()

        # Convert current time to ISO 8601 format
        current_time_iso = current_time.isoformat()
        if REMINDERS[current_time_iso]:
            print("Reminder:", REMINDERS[current_time_iso])
            logging.info("Reminder: " + REMINDERS[current_time_iso])
            try:
                send_bulk_email(USER_MAIL, USER_PASS, RECEPTIONISTS, "Your Assisstant", REMINDERS[current_time_iso])
                logging.info("Sent reminder "+str(REMINDERS[current_time_iso])+" to "+str(RECEPTIONISTS))
            except:
                logging.error("Error sending reminder!")
            del REMINDERS[current_time_iso]
        time.sleep(10)
reminder_thread = threading.Thread(target=check_reminders)

import json
import logging

# Process the response from Gemini to extract the tool and its inputs
def process_gemini_response(response):
    try:
        response_data = json.loads(response)
        tool_name = response_data['tool']
        inputs = response_data['inputs']
        return tool_name, inputs
    except Exception as e:
        logging.error(f"Error processing Gemini AI response: {e}")
        return None, None

# Predefined headstart options
HEADSTART_OPTIONS = {
    "1": "Get today's weather in New York.",
    "2": "Set a reminder for tomorrow's meeting.",
    "3": "Tell me a joke.",
    "4": "Play some music.",
    "5": "Search Wikipedia for Python programming."
}

# Function to display headstart options
def display_headstart_options():
    print("Choose a headstart option:")
    for key, option in HEADSTART_OPTIONS.items():
        print(f"{key}: {option}")

# The main loop to handle user commands
def main():
    global RUNNING
    global REMINDERS

    head_start_not_done = 0
    while RUNNING:
        logging.info("Listening for command...")
        if head_start_not_done < 1:
            display_headstart_options()  # Display options before listening for commands
            head_start_not_done = 1

        try:
            query = recognize_speech(timeout=10)  # Replace with actual speech recognition function
        except Exception as e:
            logging.error(f"Error recognizing speech: {e}")
            speak("Sorry, I didn't catch that.")
            continue

        if not query:
            logging.warning("No command recognized, continuing...")
            continue

        query = query.lower()
        logging.info(f"Recognized command: {query}")

        # Check if the query matches any headstart option
        if query in HEADSTART_OPTIONS.values() and head_start_not_done == 1:
            head_start_not_done += 1
            # Map the option to the appropriate command
            option_key = next(key for key, option in HEADSTART_OPTIONS.items() if option == query)
            query = HEADSTART_OPTIONS[option_key]  # Use the selected headstart option
        else:
            head_start_not_done += 1
            # Prompt Gemini AI to select tool and provide inputs
            gemini_prompt = f"""
            I have a list of tools that can perform specific actions. Based on the user query, select the appropriate tool from the list and extract the necessary inputs. Please respond in the following JSON format:

            {{
                "tool": "<tool_name>",
                "inputs": {{
                    "input1": "<value1>",
                    "input2": "<value2>",
                    ...
                }}
            }}

            Here is the list of tools:
            1. "just_Say": Requires "msg", a generic conversational response if none of the following tools fits best.
            2. "set_reminder": Requires "input_text" (simply return enhanced the input Query which I give).
            3. "search_wikipedia": Requires "query".
            4. "get_weather": Requires "city_name".
            5. "tell_joke": Requires no inputs.
            6. "play_music": Requires "song_name".
            7. "open_website": Requires "website_name" (return an empty string if no website is specified).
            8. "open_application": Requires "app_name".
            9. "get_news": Requires no inputs.
            10. "openfile": Requires "file" (If path is there, give file name; if there is a location description, please build file path from description).
            11. "abort": Requires a "message". Call this if the user wants to leave.
            If the tool requires no inputs, leave the "inputs" field empty.

            Query: "{query}"
            """
        
        gemini_response = get_gemini_response(gemini_prompt)
        tool_name, inputs = process_gemini_response(gemini_response)
        
        print(tool_name, inputs)
        if tool_name:
            # Find the corresponding tool and call its action with the provided inputs
            tool = next((t for t in TOOLS if t['name'] == tool_name), None)
            if tool:
                try:
                    tool['action'](**inputs)
                except Exception as e:
                    logging.error(f"Error executing {tool_name}: {e}")
                    speak(f"Sorry, there was an error with {tool_name}.")
            else:
                logging.error(f"Tool {tool_name} not found.")
                speak(f"Sorry, I couldn't find the tool {tool_name}.")
        else:
            speak("Sorry, I couldn't process the query.")

if __name__ == "__main__":
    RUNNING = True
    
    reminder_thread.start()
    response = "Hello, I'm Your PA. How can I assist you today?"
    logging.info(response)
    speak(response)
    try:
        main()
    except KeyboardInterrupt:
        RUNNING = False