import os import json import logging from typing import List, Optional, Union from pydantic import BaseModel import sympy as sp from datetime import datetime # Set up basic logging configuration logging.basicConfig( level=logging.INFO, format="%(asctime)s - %(levelname)s - %(message)s" ) class Tools: MAX_MEMORY_SIZE = 100 # Limit for memory entries def __init__(self, memory_file: str = "memory.json", encryption: bool = False): self.memory_file = memory_file # Path for the memory file self.memory: List[str] = [] # List to store chat history self.encryption = encryption # Option for memory encryption self.load_memory() # Load existing memory from the specified JSON file def message_handler(self, message: str) -> str: """ Handle incoming messages and determine the appropriate action. :param message: The message from the user. :return: The response based on the message. """ # Normalize the message for easier processing message = message.strip().lower() # Check for existing memory and respond appropriately response = "" command_mapping = { "calculate": self.calculator, "current date": self.get_current_date, "read memory": self.read_memory, "clear memory": self.clear_memory, } for command, action in command_mapping.items(): if message.startswith(command): if command == "calculate": equation = message[len(command) :].strip() # Extract the equation response = action(equation) else: response = action() # Call action without parameters break else: response = self.handle_unrecognized_message(message) # Reset to initial state if needed self.reset_state() return response def handle_unrecognized_message(self, message: str) -> str: """ Handle unrecognized messages with context. :param message: The unrecognized message. :return: A contextual response. """ if message in ["ok", "uh", "no way"]: return self.generate_response_based_on_input(message) return "I didn't understand that. Please try a different command." def generate_response_based_on_input(self, message: str) -> str: """ Generate a contextual response based on specific inputs. :param message: The input message. :return: A contextual response. """ responses = { "ok": "It seems we're back to square one! What would you like to talk about?", "uh": "You're still unsure about something, aren't you? Don't worry, I'm here to help. What's on your mind?", "no way": "I remember! Your response was just 'NO WAY.', which seemed a bit unusual, but we went with it!", } return responses.get(message, "I didn't understand that.") def calculator(self, equation: str) -> str: """ Calculate the numeric result of an equation safely. :param equation: The equation to calculate. :return: The result of the equation or an error message. """ try: expr = sp.sympify(equation) # Parse the equation into a sympy expression result = expr.evalf() # Evaluate the expression numerically result_str = f"{equation} = {result}" self.add_to_memory(result_str) # Store the calculation result in memory return result_str except (sp.SympifyError, ValueError, TypeError) as e: logging.error(f"Error in calculator: {e}") return f"Invalid equation: {e}. Please provide a valid mathematical expression." def get_current_date(self) -> str: """ Get the current date in a formatted string. :return: The current date as a formatted string. """ current_date = datetime.now().strftime("%A, %B %d, %Y") logging.info("Retrieved current date.") return f"Today's date is {current_date}" def add_to_memory(self, message: str) -> None: """ Add a message to the memory and save it to a JSON file. :param message: The message to remember. """ if len(self.memory) >= self.MAX_MEMORY_SIZE: self.memory.pop(0) # Remove the oldest entry if memory exceeds the limit self.memory.append(message) self.save_memory() logging.info(f"Added message to memory: {message}") def read_memory(self) -> str: """ Read all stored messages in memory. :return: A string representation of all messages. """ return "\n".join(self.memory) if self.memory else "Memory is empty." def clear_memory(self) -> None: """ Clear all stored messages in memory. """ self.memory.clear() self.save_memory() logging.info("Memory has been cleared.") def save_memory(self) -> None: """ Save the current memory to a JSON file, with optional encryption. """ try: data = json.dumps(self.memory) if self.encryption: data = self._encrypt(data) # Encrypt if encryption is enabled with open(self.memory_file, "w") as f: f.write(data) logging.info("Memory saved successfully.") except IOError as e: logging.error(f"Error saving memory: {e}") def load_memory(self) -> None: """ Load memory from a JSON file if it exists. """ if os.path.exists(self.memory_file): try: with open(self.memory_file, "r") as f: data = f.read() if self.encryption: data = self._decrypt(data) # Decrypt if encryption is enabled self.memory = json.loads(data) logging.info("Memory loaded successfully.") except (IOError, json.JSONDecodeError) as e: logging.error(f"Error loading memory: {e}") self.memory = [] def reset_state(self) -> None: """ Reset any necessary states or flags after processing a message. This can be expanded based on requirements. """ # Reset any internal states if necessary pass def _encrypt(self, data: str) -> str: """ Placeholder encryption method. :param data: The data to encrypt. :return: Encrypted data string. """ # Implement encryption logic here return data def _decrypt(self, data: str) -> str: """ Placeholder decryption method. :param data: The data to decrypt. :return: Decrypted data string. """ # Implement decryption logic here return data # Example usage if __name__ == "__main__": tools = Tools(encryption=True) # Simulated user inputs user_inputs = [ "Hello.", "What is the date today?", "Nice.", "Uh.", "What is 94350349578 + 3450734075?", "WOAH.", "How did you do that well?", "What is that provided context?", "Oh. A calculator context.", "Ok.", "So uh. What is the date today?", "And you also remember our convo.", "What's the first message I made?", "What was your reply?", "After that?", "NO WAY.", ] for user_input in user_inputs: response = tools.message_handler(user_input) print(response) print("\nChat History:") print(tools.read_memory())