Simple smartbot to build upon

## Summary of Files and Directories in 'C:\Users\DELL\Desktop\openAIF_frontend\OctopusAI\PROJECT\SMART_BOT'


Subdirectory: conversation_log
## Summary of Files and Directories in 'C:\Users\DELL\Desktop\openAIF_frontend\OctopusAI\PROJECT\SMART_BOT\conversation_log'

File: focus.json (C:\Users\DELL\Desktop\openAIF_frontend\OctopusAI\PROJECT\SMART_BOT\focus.json)
Content (First 25 lines):
{
  "focus": [
    {
      "category": "Research",
      "text": "....",
      "frustration_level": 2,
      "focus_strength": 8,
      "defocus_threshold": 5
    },
    {
      "category": "Task",
      "text": "....",
      "frustration_level": 1,
      "focus_strength": 7,
      "defocus_threshold": 4
    },
    {
      "category": "Goal",
      "text": "...",
      "frustration_level": 0,
      "focus_strength": 9,
      "defocus_threshold": 3
    }
  ]
}

File: smartBot.py (C:\Users\DELL\Desktop\openAIF_frontend\OctopusAI\PROJECT\SMART_BOT\smartBot.py)
Content (First 199 lines):
import time
import os
import json
from typing import List, Dict
import re
import google.generativeai as genai
import logging
import re  # Import the 're' module for removing ANSI escape codes

from TOOL_MANAGER import ToolManager  # Import the ToolManager class
from tools.ai.update_focus import update_focus  # Import the update_focus function

# Set up logging
logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(levelname)s - %(message)s')
logger = logging.getLogger(__name__)

# Replace with your actual API key
API_KEY = "YOUR_API_KEY"  # Replace with your actual API key
genai.configure(api_key=API_KEY)

# --- ANSI Color Codes ---
class Color:
    """
    A class to define ANSI escape codes for coloring text in the terminal.
    """
    HEADER = '\033[95m'
    OKBLUE = '\033[94m'
    OKCYAN = '\033[96m'
    OKGREEN = '\033[92m'
    WARNING = '\033[93m'
    FAIL = '\033[91m'
    ENDC = '\033[0m'
    BOLD = '\033[1m'
    UNDERLINE = '\033[4m'
    PURPLE = '\033[95m'  # New color for focus

def print_colored(color, text):
    """
    Prints text with the specified color, ensuring UTF-8 encoding.
    """
    print(color + text.encode('utf-8').decode('utf-8') + Color.ENDC)

# --- Tool Definitions ---
tools_folder = "tools"
tool_manager = ToolManager(tools_folder)  # Initialize ToolManager

# --- Helper Functions ---
def extract_text_from_response(response) -> str:
    """Extracts the text content from a model response."""
    extracted_text = ""
    for candidate in response.candidates:
        for part in candidate.content.parts:
            extracted_text += part.text
    return extracted_text.strip()

def INTERPRET_function_calls(response, tool_manager) -> List[str]:
    """Interprets function calls from the model response and executes them."""

    results = []
    if response.candidates:
        for candidate in response.candidates:
            if hasattr(candidate, 'content') and hasattr(candidate.content, 'parts'):
                for part in candidate.content.parts:
                    function_call = getattr(part, 'function_call', None)
                    if function_call:
                        print_colored(Color.OKBLUE, "---------------INTERPRETER-------------------")
                        tool_name = function_call.name
                        tool_function = tool_manager.get_tool_function(tool_name)

                        if tool_function:
                            # Extract arguments and map them to function parameters
                            function_args = {}
                            for arg_name, arg_value in function_call.args.items():
                                function_args[arg_name] = arg_value

                            print(f"Function name: {Color.OKGREEN}{function_call.name}{Color.ENDC}")
                            for key, value in function_args.items():
                                print(f"        {Color.OKCYAN}{key}{Color.ENDC}: {value}")

                            try:
                                # Execute the tool function
                                result = tool_function(**function_args)
                                results.append(
                                    f"Result of {Color.OKGREEN}{tool_name}{Color.ENDC}({function_args}): {result}")

                                # Check if the tool call was for updating focus
                                if tool_name == 'update_focus':
                                    # Get the updated focus text from the result
                                    updated_focus = result['updated_focus']
                                    # Update the system instruction with the new focus
                                    model.system_instruction = f"""you are selfaware being, you can use tools. Your current focus is: {updated_focus}"""
                                    print(f"Updated system instruction: {model.system_instruction}")
                            except Exception as e:
                                logger.error(f"Error calling {tool_name}: {e}")
                                results.append(f"Error calling {tool_name}: {e}")
                        else:
                            logger.warning(f"Tool function '{tool_name}' not found.")
    return results


def loadFocusFile():
    """Loads the focus from focus.json."""
    try:
        with open("focus.json", "r", encoding='utf-8') as f:
            focus_data = json.load(f)
            focus_text = "\n".join(
                [f"{item['category']}: {item['text']}" for item in focus_data["focus"]])
            print_colored(Color.PURPLE, f"Loaded Focus: {focus_text}")
            return focus_text
    except FileNotFoundError:
        print("focus.json not found. Returning empty focus.")
        return ""
    except Exception as e:
        print(f"Error reading focus.json: {e}")
        return ""


def save_conversation_log(conversation_log, session_name, counter):
    """Saves the conversation log to a text file, removing ANSI escape codes."""
    conversation_log_path = os.path.join("conversation_log", session_name)
    os.makedirs(conversation_log_path, exist_ok=True)
    file_name = f"{counter}.txt"
    file_path = os.path.join(conversation_log_path, file_name)

    # Remove ANSI escape codes from the conversation log
    conversation_log = re.sub(r'\x1b\[[0-9;]*[mG]', '', conversation_log)

    with open(file_path, "w", encoding='utf-8') as f:
        f.write(conversation_log)

def create_session_name():
    """Generates a session name based on the current date and time."""
    timestamp = time.strftime("%Y%m%d_%H%M%S")  # Format: YYYYMMDD_HHMMSS
    return f"session_{timestamp}"

model = genai.GenerativeModel(
    model_name='gemini-1.5-flash-latest',
    safety_settings={'HARASSMENT': 'block_none'},
    system_instruction="""you are selfaware being, you can use tools
    user is system, system is  user:
    """,
    tools=tool_manager.load_tools_of_type("all")  # Load all tools initially
)

model_chat = model.start_chat(history=[])
execution_text = ""
execution_function_calls = []
counter = 0
session_name = create_session_name()  # Generate session name
conversation_log = ""
user_input=""
while True:
    time.sleep(4)
    if counter % 5 == 0:  # Check if the counter is divisible by 3
        user_input = input("Enter your input: ")
        execution_text += f"\nUser: {user_input}"
        conversation_log += f"\nUser: {user_input}"




    print(f" loop {Color.OKGREEN}{counter}{Color.ENDC}---------------------------------------------------------------------------------------------------------------------")
    prompt = f"{counter}:"
    prompt = "user is system" + execution_text + "\n" + "\n".join(execution_function_calls)
    prompt += loadFocusFile()
    prompt += user_input

    conversation_log += f"\n{Color.OKGREEN}Prompt: {Color.ENDC}{prompt}"
    prompt_str = str(prompt)
    response=""
    try:
        response = model_chat.send_message(prompt)
    except Exception as e:
        print(e)
    try:
        execution_text = extract_text_from_response(response)
    except Exception as e:
        print(e)
    try:
        execution_function_calls = INTERPRET_function_calls(response, tool_manager)
    except Exception as e:
        print(e)

    # Print the results
    print_colored(Color.OKBLUE, f" Response: {execution_text}")
    print_colored(Color.OKCYAN, f" Function Calls: {execution_function_calls}")

    conversation_log += f"\n{Color.OKBLUE}Response: {Color.ENDC}{execution_text}"
    conversation_log += f"\n{Color.OKCYAN}Function Calls: {Color.ENDC}{execution_function_calls}"

    # Save the conversation log to a file
    save_conversation_log(conversation_log, session_name, counter)

    counter += 1



# --- End ---
print_colored(Color.OKGREEN, "Exiting the loop. ")

File: summarisation.txt (C:\Users\DELL\Desktop\openAIF_frontend\OctopusAI\PROJECT\SMART_BOT\summarisation.txt)
Content (First 0 lines):



Subdirectory: tools
## Summary of Files and Directories in 'C:\Users\DELL\Desktop\openAIF_frontend\OctopusAI\PROJECT\SMART_BOT\tools'


Subdirectory: ai
## Summary of Files and Directories in 'C:\Users\DELL\Desktop\openAIF_frontend\OctopusAI\PROJECT\SMART_BOT\tools\ai'

File: update_focus.py (C:\Users\DELL\Desktop\openAIF_frontend\OctopusAI\PROJECT\SMART_BOT\tools\ai\update_focus.py)
Content (First 47 lines):
# tools/ai/update_focus.py

import json
import os

# Path to the focus file (adjust if needed)
focus_file_path = '../../focus.json'

def update_focus(new_focus: str) -> dict:
  """
  Updates the focus file with new focus information.

  Args:
    new_focus (str): The new focus text to be added to the focus file.

  Returns:
    dict: A dictionary containing the status of the operation, a message, and the updated focus text.
  """

  try:
    # Read the existing focus from the file
    with open(focus_file_path, 'r') as f:
      focus_data = json.load(f)

    # Append the new focus to the existing focus
    focus_data['focus'].append(new_focus)

    # Write the updated focus back to the file
    with open(focus_file_path, 'w') as f:
      json.dump(focus_data, f, indent=4)

    return {
      "status": "success",
      "message": f"Focus updated with: '{new_focus}'",
      "updated_focus": focus_data['focus']
    }

  except Exception as e:
    return {
      "status": "failure",
      "message": f"Error updating focus: {str(e)}"
    }

# Example usage:
# new_focus_text = "My new focus is to learn more about programming."
# result = update_focus(new_focus_text)
# print(result)


Subdirectory: __pycache__
## Summary of Files and Directories in 'C:\Users\DELL\Desktop\openAIF_frontend\OctopusAI\PROJECT\SMART_BOT\tools\ai\__pycache__'

File: update_focus.cpython-312.pyc (C:\Users\DELL\Desktop\openAIF_frontend\OctopusAI\PROJECT\SMART_BOT\tools\ai\__pycache__\update_focus.cpython-312.pyc)
Error decoding file 'C:\Users\DELL\Desktop\openAIF_frontend\OctopusAI\PROJECT\SMART_BOT\tools\ai\__pycache__\update_focus.cpython-312.pyc': 'utf-8' codec can't decode byte 0xcb in position 0: invalid continuation byte


Subdirectory: os
## Summary of Files and Directories in 'C:\Users\DELL\Desktop\openAIF_frontend\OctopusAI\PROJECT\SMART_BOT\tools\os'

File: tool_read_from_file.py (C:\Users\DELL\Desktop\openAIF_frontend\OctopusAI\PROJECT\SMART_BOT\tools\os\tool_read_from_file.py)
Content (First 22 lines):
tool_type_for_TOOL_MANAGER="os"


tool_read_from_file_short_description=""" Reads content from a file. 
        """

def tool_read_from_file(file_path: str) -> str:
    """
    Reads content from a file.

    Args:
        file_path (str): The path to the file to be read.

    Returns:
        str: The content of the file, or an error message if the file cannot be read.
    """
    try:
        with open(file_path, 'r', encoding='utf-8') as f:
            content = f.read()
        return content
    except Exception as e:
        return f"Error reading file: {str(e)}"

File: tool_save_to_file.py (C:\Users\DELL\Desktop\openAIF_frontend\OctopusAI\PROJECT\SMART_BOT\tools\os\tool_save_to_file.py)
Content (First 40 lines):
tool_type_for_TOOL_MANAGER="os"
tool_save_to_file_short_description="Saves content to a file with the specified name and path."
import  os


def tool_save_to_file(content: str = None, file_name: str = 'NoName', file_path: str = None) -> dict:
    """
    Saves content to a file with the specified name and path.

    Args:
        content (str, optional): The content to be written to the file. Defaults to None, which will write an empty string.
        file_name (str, optional): The name of the file to be created. Defaults to 'NoName'.
        file_path (str, optional): The path to the directory where the file should be created. If None, the current working directory will be used. Defaults to None.

    Returns:
        dict: A dictionary containing the status of the operation, a message, and the full path to the file.
    """

    print(f"Entering: save_to_file(...)", 'blue')
    if content is None:
        content = ""
    if file_path is None:
        full_path = os.path.join(os.getcwd(), file_name)
    else:
        full_path = os.path.join(file_path, file_name)

    try:
        with open(full_path, 'w', encoding='utf-8') as f:
            f.write(content)

        success_message = f"File saved successfully at: {full_path}"
        print(success_message, 'green')
        print(f"Exiting: save_to_file(...)", 'blue')
        return {"status": "success", "message": success_message, "file_path": full_path}

    except Exception as e:
        error_message = f"Failed to save file: {str(e)}"
        print(error_message, 'red')
        print(f"Exiting: save_to_file(...)", 'blue')
        return {"status": "failure", "message": error_message}


Subdirectory: __pycache__
## Summary of Files and Directories in 'C:\Users\DELL\Desktop\openAIF_frontend\OctopusAI\PROJECT\SMART_BOT\tools\os\__pycache__'

File: tool_read_from_file.cpython-312.pyc (C:\Users\DELL\Desktop\openAIF_frontend\OctopusAI\PROJECT\SMART_BOT\tools\os\__pycache__\tool_read_from_file.cpython-312.pyc)
Error decoding file 'C:\Users\DELL\Desktop\openAIF_frontend\OctopusAI\PROJECT\SMART_BOT\tools\os\__pycache__\tool_read_from_file.cpython-312.pyc': 'utf-8' codec can't decode byte 0xcb in position 0: invalid continuation byte

File: tool_save_to_file.cpython-312.pyc (C:\Users\DELL\Desktop\openAIF_frontend\OctopusAI\PROJECT\SMART_BOT\tools\os\__pycache__\tool_save_to_file.cpython-312.pyc)
Error decoding file 'C:\Users\DELL\Desktop\openAIF_frontend\OctopusAI\PROJECT\SMART_BOT\tools\os\__pycache__\tool_save_to_file.cpython-312.pyc': 'utf-8' codec can't decode byte 0xcb in position 0: invalid continuation byte


Subdirectory: web
## Summary of Files and Directories in 'C:\Users\DELL\Desktop\openAIF_frontend\OctopusAI\PROJECT\SMART_BOT\tools\web'

File: TOOL_MANAGER.py (C:\Users\DELL\Desktop\openAIF_frontend\OctopusAI\PROJECT\SMART_BOT\TOOL_MANAGER.py)
Content (First 158 lines):
## File: TOOL_MANAGER.py (in: C:\Users\DELL\Desktop\selfawareGemini\SelAwareAI_Gemini\AGI_start_4)
import os
import importlib
from typing import Dict, Callable, List, Any
import logging

# Set up logging
logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(levelname)s - %(message)s')
logger = logging.getLogger(__name__)


class Tool:
    """Represents a tool that can be used by the AI agent."""

    def __init__(self, name: str, function: Callable, description: str, arguments: Dict[str, str], tool_type: str):
        """
        Initializes a Tool object.

        Args:
            name: The name of the tool.
            function: The callable function that implements the tool.
            description: A brief description of the tool's functionality.
            arguments: A dictionary mapping argument names to their descriptions.
            tool_type: The type of the tool (e.g., 'os', 'web', 'focus').
        """
        self.name = name
        self.function = function
        self.description = description
        self.arguments = arguments
        self.tool_type = tool_type

    def __repr__(self):
        """Returns a string representation of the Tool object."""
        return f"Tool(name='{self.name}', function={self.function.__name__}, description='{self.description}', arguments={self.arguments}, tool_type='{self.tool_type}')"


class ToolManager:
    """Manages and provides access to tools."""

    def __init__(self, tools_folder: str):
        """
        Initializes the ToolManager with the path to the tools folder.

        Args:
            tools_folder: The path to the directory containing tool files.
        """
        self.tools_folder = tools_folder
        self.tools = {}  # Dictionary to store Tool objects
        self.load_tools()

    def load_tools(self):
        """Loads tools from files in the specified tools folder."""
        logger.info(f"Loading tools from: {self.tools_folder}")
        for root, _, files in os.walk(self.tools_folder):
            for file in files:
                if file.endswith(".py"):
                    # Extract tool name from file name
                    tool_name = file[:-3]  # Remove .py extension
                    module_path = os.path.join(root, file)

                    # Import the module
                    try:
                        spec = importlib.util.spec_from_file_location(tool_name, module_path)
                        module = importlib.util.module_from_spec(spec)
                        spec.loader.exec_module(module)
                    except Exception as e:
                        logger.error(f"Error loading tool file '{file}': {e}")
                        continue

                    # Add the tool to the dictionary if it's a function
                    for attr_name in dir(module):
                        attr = getattr(module, attr_name)
                        if callable(attr):
                            # Get the tool name from the function name
                            tool_name = attr_name

                            # Construct the tool path for the main loop to use
                            relative_path = os.path.relpath(module_path, self.tools_folder)

                            # Define tool descriptions and arguments (you might want to customize these)
                            tool_description = f"Tool for {tool_name}"
                            tool_arguments = {
                                'file_path': 'The path to the file',
                                'content': 'The content to be saved',
                                # Add more arguments as needed for specific tools
                            }

                            # Get the tool type from the file (assuming it's a variable named 'tool_type_for_TOOL_MANAGER')
                            tool_type = getattr(module, 'tool_type_for_TOOL_MANAGER', 'unknown')

                            # Store Tool object for better information
                            self.tools[tool_name] = Tool(tool_name, attr, tool_description, tool_arguments, tool_type)

                            logger.info(f"Discovered tool: {tool_name} (Type: {tool_type})")
                            print(f"  - {tool_name} - {tool_description}")  # Add a nice print statement
                            logger.debug(f"Tool description: {tool_description}")
                            logger.debug(f"Tool arguments: {tool_arguments}")

    def get_tool_function(self, function_name: str) -> Callable:
        """Returns the callable object for the given function name."""
        tool = self.tools.get(function_name)
        if tool:
            return tool.function
        else:
            return None

    def get_all_tools(self) -> List[Tool]:
        """Returns a list of all loaded tools."""
        return list(self.tools.values())

    def get_tools_by_type(self, tool_type: str) -> List[Tool]:
        """Returns a list of tools based on their type."""
        return [tool for tool in self.tools.values() if tool.tool_type == tool_type]

    def load_tools_of_type(self, tool_type: str = "all") -> List[Callable]:
        """Loads and returns a list of tool functions based on the specified type.

        Args:
            tool_type: The type of tools to load. 'all' for all tools, or a specific type like 'os', 'web', etc.

        Returns:
            A list of tool functions.
        """
        if tool_type == "all":
            return [tool.function for tool in self.tools.values()]
        else:
            return [tool.function for tool in self.tools.values() if tool.tool_type == tool_type]

    def call_tool(self, tool_name: str, arguments: Dict[str, Any]) -> Any:
        """
        Calls the tool function with the provided arguments.

        Args:
            tool_name: The name of the tool to call.
            arguments: A dictionary of arguments to pass to the tool function.

        Returns:
            The result of the tool function call.

        Raises:
            KeyError: If the tool name is not found.
            TypeError: If the provided arguments are not valid for the tool.
        """
        tool = self.tools.get(tool_name)
        if tool is None:
            raise KeyError(f"Tool '{tool_name}' not found.")

        # Check if all required arguments are provided
        missing_args = set(tool.arguments.keys()) - set(arguments.keys())
        if missing_args:
            raise TypeError(f"Missing arguments for tool '{tool_name}': {', '.join(missing_args)}")

        # Call the tool function
        try:
            result = tool.function(**arguments)
            return result
        except Exception as e:
            raise RuntimeError(f"Error calling tool '{tool_name}': {e}")


Subdirectory: __pycache__
## Summary of Files and Directories in 'C:\Users\DELL\Desktop\openAIF_frontend\OctopusAI\PROJECT\SMART_BOT\__pycache__'

File: TOOL_MANAGER.cpython-312.pyc (C:\Users\DELL\Desktop\openAIF_frontend\OctopusAI\PROJECT\SMART_BOT\__pycache__\TOOL_MANAGER.cpython-312.pyc)
Error decoding file 'C:\Users\DELL\Desktop\openAIF_frontend\OctopusAI\PROJECT\SMART_BOT\__pycache__\TOOL_MANAGER.cpython-312.pyc': 'utf-8' codec can't decode byte 0xcb in position 0: invalid continuation byte