An innovative open-source and free alternative to traditional Code Interpreters.
"""
This is the main file for the Open-Code-Interpreter.
It handles command line arguments and initializes the Interpreter.

Command line arguments:
--exec, -e: Executes the code generated from the user's instructions.
--save_code, -s: Saves the generated code.
--mode, -md: Selects the mode of operation. Choices are 'code', 'script', and 'command'.
--model, -m: Sets the model for code generation. Default is 'code-llama'.
--version, -v: Displays the version of the program.
--lang, -l: Sets the interpreter language. Default is 'python'.
--display_code, -dc: Displays the generated code in the output.

Author: HeavenHM
Date: 2023/10/12
"""

from libs.interpreter_lib import Interpreter
import argparse
import sys
import traceback

def main():
        parser = argparse.ArgumentParser(description='Code - Interpreter')
        parser.add_argument('--exec', '-e', action='store_true', help='Execute the code')
        parser.add_argument('--save_code', '-s', action='store_true', help='Save the generated code')
        parser.add_argument('--mode', '-md', choices=['code', 'script', 'command'], help='Select the mode (`code` for generating code, `script` for generating shell scripts, `command` for generating single line commands)')
        parser.add_argument('--model', '-m', type=str, default='code-llama', help='Set the model for code generation. (Defaults to gpt-3.5-turbo)')
        parser.add_argument('--version', '-v', action='version', version='%(prog)s 1.5')
        parser.add_argument('--lang', '-l', type=str, default='python', help='Set the interpreter language. (Defaults to Python)')
        parser.add_argument('--display_code', '-dc', action='store_true', help='Display the code in output')
        args = parser.parse_args()

        # Check if only the application name is passed
        if len(sys.argv) <= 1:
            parser.print_help()
            return

        # Create an instance of the Interpreter class and call the main method.
        interpreter = Interpreter(args)
        interpreter.interpreter_main()

if __name__ == "__main__":
    try:
        main()
    except SystemExit:
        pass  # Ignore the SystemExit exception caused by --version argument
    except:
        traceback.print_exc()
interpreter.py
haseeb-heaven/Open-Code-Interpreter
"""
This is the Code Interpreter class. It provides all methods for Code LLM like Display, Execute, Format code from different llm's.
It includes features like:
- Code execution in multiple languages
- Code extraction from strings
- Saving code to a file
- Executing Code,Scripts
- Checking for compilers
"""

import os
import subprocess
import traceback
from libs.logger import initialize_logger

class CodeInterpreter:

    def __init__(self):
        self.logger = initialize_logger("logs/code-interpreter.log")
    
    def _execute_script(self, script: str, shell: str):
        stdout = stderr = None
        try:
            self.logger.info(f"Running {shell} script")
            if shell == "bash":
                process = subprocess.Popen(['bash', '-c', script], stdout=subprocess.PIPE, stderr=subprocess.PIPE)
            elif shell == "powershell":
                process = subprocess.Popen(['powershell', '-Command', script], stdout=subprocess.PIPE, stderr=subprocess.PIPE)
            elif shell == "applescript":
                process = subprocess.Popen(['osascript', '-'], stdin=subprocess.PIPE, stdout=subprocess.PIPE, stderr=subprocess.PIPE)
            else:
                self.logger.error(f"Invalid shell selected: {shell}")
                return None, f"Invalid shell selected: {shell}"
            stdout, stderr = process.communicate()
            self.logger.info(f"Output is {stdout.decode()} and error is {stderr.decode()}")
            if process.returncode != 0:
                self.logger.error(f"Error in running {shell} script: {stderr.decode()}")
        except Exception as exception:
            self.logger.error(f"Exception in running {shell} script: {str(exception)}")
            stderr = str(exception)
        finally:
            return stdout.decode().strip() if stdout else None, stderr.decode().strip() if stderr else None
        
    def _check_compilers(self, language):
        try:
            language = language.lower().strip()
            
            compilers = {
                "python": ["python", "--version"],
                "javascript": ["node", "--version"],
            }

            if language not in compilers:
                self.logger.error("Invalid language selected.")
                return False

            compiler = subprocess.run(compilers[language], capture_output=True, text=True)
            if compiler.returncode != 0:
                self.logger.error(f"{language.capitalize()} compiler not found.")
                return False

            return True
        except Exception as exception:
            self.logger.error(f"Error occurred while checking compilers: {exception}")
            raise Exception(f"Error occurred while checking compilers: {exception}")
  
    def save_code(self, filename='output/code_generated.py', code=None):
        """
        Saves the provided code to a file.
        The default filename is 'code_generated.py'.
        """
        try:
            # Check if the directory exists, if not create it
            directory = os.path.dirname(filename)
            if not os.path.exists(directory):
                os.makedirs(directory)

            with open(filename, 'w') as file:
                file.write(code)
                self.logger.info(f"Code saved successfully to {filename}.")
        except Exception as exception:
            self.logger.error(f"Error occurred while saving code to file: {exception}")
            raise Exception(f"Error occurred while saving code to file: {exception}")

    def extract_code(self, code:str, start_sep='```', end_sep='```',skip_first_line=False,code_mode=False):
        """
        Extracts the code from the provided string.
        If the string contains the start and end separators, it extracts the code between them.
        Otherwise, it returns the original string.
        """
        try:
            has_newline = False
            if start_sep in code and end_sep in code:
                start = code.find(start_sep) + len(start_sep)
                # Skip the newline character after the start separator
                if code[start] == '\n':
                    start += 1
                    has_newline = True
                    
                end = code.find(end_sep, start)
                # Skip the newline character before the end separator
                if code[end - 1] == '\n':
                    end -= 1
                    
                if skip_first_line and code_mode and not has_newline:
                    # Skip the first line after the start separator
                    start = code.find('\n', start) + 1
                    
                extracted_code = code[start:end]
                # Remove extra words for commands present.
                if not code_mode and 'bash' in extracted_code:
                    extracted_code = extracted_code.replace('bash', '')
                    
                self.logger.info("Code extracted successfully.")
                return extracted_code
            else:
                self.logger.info("No special characters found in the code. Returning the original code.")
                return code
        except Exception as exception:
            self.logger.error(f"Error occurred while extracting code: {exception}")
            raise Exception(f"Error occurred while extracting code: {exception}")
          
    def execute_code(self, code, language):
        """
        This method is used to execute the provided code in the specified language.

        Parameters:
        code (str): The code to be executed.
        language (str): The programming language in which the code is written.

        Returns:
        str: The output of the executed code.
        """
        try:
            language = language.lower()
            self.logger.info(f"Running code: {code[:100]} in language: {language}")

            # Check for code and language validity
            if not code or len(code.strip()) == 0:
                return "Code is empty. Cannot execute an empty code."
            
            # Check for compilers on the system
            compilers_status = self._check_compilers(language)
            if not compilers_status:
                raise Exception("Compilers not found. Please install compilers on your system.")
            
            if language == "python":
                process = subprocess.Popen(["python", "-c", code], stdout=subprocess.PIPE, stderr=subprocess.PIPE)
                stdout, stderr = process.communicate()
                stdout_output = stdout.decode("utf-8")
                stderr_output = stderr.decode("utf-8")
                self.logger.info(f"Python Output execution: {stdout_output}, Errors: {stderr_output}")
                return stdout_output, stderr_output
            
            elif language == "javascript":
                process = subprocess.Popen(["node", "-e", code], stdout=subprocess.PIPE, stderr=subprocess.PIPE)
                stdout, stderr = process.communicate()
                stdout_output = stdout.decode("utf-8")
                stderr_output = stderr.decode("utf-8")
                self.logger.info(f"JavaScript Output execution: {stdout_output}, Errors: {stderr_output}")
                return stdout_output, stderr_output
            
            else:
                self.logger.info("Unsupported language.")
                raise Exception("Unsupported language.")
                
        except Exception as exception:
            self.logger.error(f"Exception in running code: {str(exception)}")
            raise exception
        
    def execute_script(self, script:str, os_type:str='macos'):
        output = error = None
        try:
            if not script:
                raise ValueError("Script must be provided.")
            if not os_type:
                raise ValueError("OS type must be provided.")
            
            self.logger.info(f"Attempting to execute script: {script[:50]}")
            if os_type.lower() == 'macos':
                output, error = self._execute_script(script, shell='applescript')
            elif os_type.lower() == 'linux':
                output, error = self._execute_script(script, shell='bash')
            elif os_type.lower() == 'windows':
                output, error = self._execute_script(script, shell='powershell')
            else:
                raise ValueError("Invalid OS type. Please provide 'macos', 'linux', or 'windows'.")
            
            if output:
                self.logger.info(f"Script executed successfully with output: {output[:50]}...")
            if error:
                self.logger.error(f"Script executed with error: {error}...")
        except Exception as exception:
            self.logger.error(f"Error in executing script: {traceback.format_exc()}")
            error = str(exception)
        finally:
            return output, error
        
    def execute_command(self, command:str):
        try:
            if not command:
                raise ValueError("Command must be provided.")
            
            self.logger.info(f"Attempting to execute command: {command}")
            process = subprocess.run(command, shell=True, stdout=subprocess.PIPE, stderr=subprocess.PIPE)
            
            stdout_output = process.stdout.decode("utf-8")
            stderr_output = process.stderr.decode("utf-8")
            
            if stdout_output:
                self.logger.info(f"Command executed successfully with output: {stdout_output}")
            if stderr_output:
                self.logger.error(f"Command executed with error: {stderr_output}")
            
            return stdout_output, stderr_output
        except Exception as exception:
            self.logger.error(f"Error in executing command: {str(exception)}")
            raise exception
libs/code_interpreter.py
haseeb-heaven/Open-Code-Interpreter
"""
This file contains the `Interpreter` class which is responsible for:

- Initializing the interpreter with the necessary configurations and arguments.
- Handling different modes of operation: Code, Script, and Command.
- Generating code based on the user's input and the selected mode.
- Executing the generated code if the user chooses to do so.
- Handling errors during code execution and attempting to resolve them by installing missing packages.
- Cleaning up response files after each iteration.
- Opening resource files such as graphs, charts, and tables generated by the code.
- Logging all important actions and errors for debugging and traceability.
"""

import os
import platform
import subprocess
import time
import webbrowser
from libs.code_interpreter import CodeInterpreter
from litellm import completion
from libs.logger import initialize_logger
from libs.markdown_code import display_code, display_markdown_message
from libs.package_installer import PackageInstaller
from libs.utility_manager import UtilityManager
from dotenv import load_dotenv
        
class Interpreter:
    logger = None
    client = None
    interpreter_version = "1.5"
    
    def __init__(self, args):
        self.args = args
        self.history = []
        self.utility_manager = UtilityManager()
        self.code_interpreter = CodeInterpreter()
        self.package_installer = PackageInstaller()
        self.logger = initialize_logger("logs/interpreter.log")
        self.client = None
        self.config_values = None
        self.system_message = ""
        self.initialize()

    def _open_resource_file(self,filename):
        try:
            if os.path.isfile(filename):
                if platform.system() == "Windows":
                    subprocess.call(['start', filename], shell=True)
                elif platform.system() == "Darwin":
                    subprocess.call(['open', filename])
                elif platform.system() == "Linux":
                    subprocess.call(['xdg-open', filename])
                self.logger.info(f"{filename} exists and opened successfully")
        except Exception as exception:
            display_markdown_message(f"Error in opening files: {str(exception)}")

    def _clean_responses(self):
        files_to_remove = ['graph.png', 'chart.png', 'table.md']
        for file in files_to_remove:
            try:
                if os.path.isfile(file):
                    os.remove(file)
                    self.logger.info(f"{file} removed successfully")
            except Exception as e:
                print(f"Error in removing {file}: {str(e)}")
    
    def _extract_content(self,output):
        try:
            return output['choices'][0]['message']['content']
        except (KeyError, TypeError) as e:
            self.logger.error(f"Error extracting content: {str(e)}")
            raise
    
    def initialize(self):
        self.INTERPRETER_LANGUAGE = self.args.lang if self.args.lang else 'python'
        self.SAVE_CODE = self.args.save_code
        self.EXECUTE_CODE = self.args.exec
        self.DISPLAY_CODE = self.args.display_code
        self.INTERPRRETER_MODEL = self.args.model if self.args.model else None
        self.logger.info(f"Interpreter args model selected is '{self.args.model}")
        self.logger.info(f"Interpreter model selected is '{self.INTERPRRETER_MODEL}")
        self.system_message = ""
        # Open file system_message.txt to a variable system_message
        try:
            with open('system/system_message.txt', 'r') as file:
                self.system_message = file.read()
                if self.system_message != "":
                    self.logger.info(f"System message read successfully")
        except Exception as exception:
            self.logger.error(f"Error occurred while reading system_message.txt: {str(exception)}")
            raise
        
        # Initialize client and mode.
        self.initialize_client()
        self.initialize_mode()
        
        try: # Make this as optional step to have readline history.
            self.utility_manager.initialize_readline_history()
        except:
            self.logger.error(f"Exception on initializing readline history")

    def initialize_client(self):
        load_dotenv()
        hf_model_name = ""
        self.logger.info("Initializing Client")
        
        self.logger.info(f"Interpreter model selected is '{self.INTERPRRETER_MODEL}")
        if self.INTERPRRETER_MODEL is None or self.INTERPRRETER_MODEL == "":
            self.logger.info("HF_MODEL is not provided, using default model.")
            self.INTERPRRETER_MODEL = self.INTERPRRETER_MODEL
            hf_model_name = self.INTERPRRETER_MODEL.strip().split("/")[-1]
            config_file_name = f"configs/gpt-3.5-turbo.config" # Setting default model to GPT 3.5 Turbo.
        else:
            config_file_name = f"configs/{self.INTERPRRETER_MODEL}.config"
        
        self.logger.info(f"Reading config file {config_file_name}")    
        self.config_values = self.utility_manager.read_config_file(config_file_name)
        self.INTERPRRETER_MODEL = str(self.config_values.get('HF_MODEL', self.INTERPRRETER_MODEL))       
        hf_model_name = self.INTERPRRETER_MODEL.strip().split("/")[-1]
        
        self.logger.info(f"Using model {hf_model_name}")
        
        if "gpt" in self.INTERPRRETER_MODEL:
            if os.getenv("OPENAI_API_KEY") is None:
                load_dotenv()
            if os.getenv("OPENAI_API_KEY") is None:
                # if there is no .env file, try to load from the current working directory
                load_dotenv(dotenv_path=os.path.join(os.getcwd(), ".env"))
                
            # Read the token from the .env file
            openai_key = os.getenv('OPENAI_API_KEY')
            if not openai_key:
                raise Exception("OpenAI Key not found in .env file.")
            elif not openai_key.startswith('sk-'):
                raise Exception("OpenAI token should start with 'sk-'. Please check your .env file.")
        
        elif "palm" in self.INTERPRRETER_MODEL:
            self.logger.info("User has agreed to terms and conditions of PALM-2")
            
            if os.getenv("PALM_API_KEY") is None:
                load_dotenv()
            if os.getenv("PALM_API_KEY") is None:
                # if there is no .env file, try to load from the current working directory
                load_dotenv(dotenv_path=os.path.join(os.getcwd(), ".env"))
            
            palm_token = os.getenv('PALM_API_KEY')
            # Validate the token
            if not palm_token:
                raise Exception("PALM token not found in .env file.")
            elif " " in palm_token or len(palm_token) <= 15:
                raise Exception("PALM API Key should have no spaces, length greater than 15. Please check your .env file.")
        
        else:
            if os.getenv("HUGGINGFACE_API_KEY") is None:
                load_dotenv()
            if os.getenv("HUGGINGFACE_API_KEY") is None:
                # if there is no .env file, try to load from the current working directory
                load_dotenv(dotenv_path=os.path.join(os.getcwd(), ".env"))
                
            # Read the token from the .env file
            openai_key = os.getenv('HUGGINGFACE_API_KEY')
            if not openai_key:
                raise Exception("HuggingFace token not found in .env file.")
            elif not openai_key.startswith('hf_'):
                raise Exception("HuggingFace token should start with 'hf_'. Please check your .env file.")

    def initialize_mode(self):
        self.CODE_MODE = True if self.args.mode == 'code' else False
        self.SCRIPT_MODE = True if self.args.mode == 'script' else False
        self.COMMAND_MODE = True if self.args.mode == 'command' else False
        if not self.SCRIPT_MODE and not self.COMMAND_MODE:
            self.CODE_MODE = True
    
    def get_prompt(self,message: str, chat_history: list[tuple[str, str]]) -> str:
        system_message = None
        
        if self.CODE_MODE:
            system_message = self.system_message
        elif self.SCRIPT_MODE:
            system_message = "Please generate a well-written script that is precise, easy to understand, and compatible with the current operating system."
        elif self.COMMAND_MODE:
            system_message = "Please generate a single line command that is precise, easy to understand, and compatible with the current operating system."
            
        messages = [
            {"role": "system", "content":system_message},
            {"role": "assistant", "content": "Please generate code wrapped inside triple backticks known as codeblock."},
            {"role": "user", "content": message}
        ]
        return messages
    
    def generate_text(self,message, chat_history: list[tuple[str, str]], temperature=0.1, max_tokens=1024,config_values=None):
        self.logger.debug("Generating code.")
        
        # Use the values from the config file if they are provided
        if config_values:
            temperature = float(config_values.get('temperature', temperature))
            max_tokens = int(config_values.get('max_tokens', max_tokens))
            api_base = str(config_values.get('api_base', None)) # Only for OpenAI.

        # Get the system prompt
        messages = self.get_prompt(message, chat_history)
        
         # Call the completion function
         
        
        if 'huggingface/' not in self.INTERPRRETER_MODEL and 'gpt' not in self.INTERPRRETER_MODEL and 'palm' not in self.INTERPRRETER_MODEL:
            self.INTERPRRETER_MODEL = 'huggingface/' + self.INTERPRRETER_MODEL

        # Check if the model is GPT 3.5/4
        if 'gpt' in self.INTERPRRETER_MODEL:
            self.logger.info("Model is GPT 3.5/4.")
            if api_base != 'None':
                # Set the custom language model provider
                custom_llm_provider = "openai"
                self.logger.info(f"Custom API mode selected for OpenAI, api_base={api_base}")
                response = completion(self.INTERPRRETER_MODEL, messages=messages, temperature=temperature, max_tokens=max_tokens, api_base=api_base, custom_llm_provider=custom_llm_provider)
            else:
                self.logger.info(f"Default API mode selected for OpenAI.")
                response = completion(self.INTERPRRETER_MODEL, messages=messages, temperature=temperature, max_tokens=max_tokens)
            self.logger.info("Response received from completion function.")
                
        # Check if the model is PALM-2
        elif 'palm' in self.INTERPRRETER_MODEL:
            self.logger.info("Model is PALM-2.")
            self.INTERPRRETER_MODEL = "palm/chat-bison"
            response = completion(self.INTERPRRETER_MODEL, messages=messages,temperature=temperature,max_tokens=max_tokens)
            self.logger.info("Response received from completion function.")
        
        # Check if model are from Hugging Face.
        else:
            self.logger.info("Model is from Hugging Face.")
            response = completion(self.INTERPRRETER_MODEL, messages=messages,temperature=temperature,max_tokens=max_tokens)
            self.logger.info("Response received from completion function.")
        
        self.logger.info(f"Generated text {response}")
        generated_text = self._extract_content(response)
        self.logger.info(f"Generated content {generated_text}")
        return generated_text

    def handle_code_mode(self, task, os_name):
        prompt = f"Generate the code in {self.INTERPRETER_LANGUAGE} language for this task '{task} for Operating System: {os_name}'."
        self.history.append((task, prompt))
        return prompt

    def handle_script_mode(self, task, os_name):
        language_map = {'macos': 'applescript', 'linux': 'bash', 'windows': 'powershell'}
        self.INTERPRETER_LANGUAGE = language_map.get(os_name.lower(), 'python')
        
        script_type = 'Apple script' if os_name.lower() == 'macos' else 'Bash Shell script' if os_name.lower() == 'linux' else 'Powershell script' if os_name.lower() == 'windows' else 'script'
        prompt = f"\nGenerate {script_type} for this prompt and make this script easy to read and understand for this task '{task} for Operating System is {os_name}'."
        return prompt

    def handle_command_mode(self, task, os_name):
        prompt = f"Generate the single terminal command for this task '{task} for Operating System is {os_name}'."
        return prompt

    def handle_mode(self, task, os_name):
        if self.CODE_MODE:
            return self.handle_code_mode(task, os_name)
        elif self.SCRIPT_MODE:
            return self.handle_script_mode(task, os_name)
        elif self.COMMAND_MODE:
            return self.handle_command_mode(task, os_name)

    def execute_code(self, extracted_code, os_name):
        execute = 'y' if self.EXECUTE_CODE else input("Execute the code? (Y/N): ")
        if execute.lower() == 'y':
            try:
                code_output, code_error = "", ""
                if self.SCRIPT_MODE:
                    code_output, code_error = self.code_interpreter.execute_script(extracted_code, os_type=os_name)
                elif self.COMMAND_MODE:
                    code_output, code_error = self.code_interpreter.execute_command(extracted_code)
                elif self.CODE_MODE:
                    code_output, code_error = self.code_interpreter.execute_code(extracted_code, language=self.INTERPRETER_LANGUAGE)
                return code_output, code_error
            except Exception as exception:
                self.logger.error(f"Error occurred while executing code: {str(exception)}")
                return None, str(exception)  # Return error message as second element of tuple
        else:
            return None, None  # Return None, None if user chooses not to execute the code

    def interpreter_main(self):
        
        self.logger.info(f"Code Interpreter - v{self.interpreter_version}")
        os_platform = self.utility_manager.get_os_platform()
        os_name = os_platform[0]
        command_mode = 'Code'
        mode = 'Script' if self.SCRIPT_MODE else 'Command' if self.COMMAND_MODE else 'Code'
        
        command_mode = mode
        start_sep = str(self.config_values.get('start_sep', '```'))
        end_sep = str(self.config_values.get('end_sep', '```'))
        skip_first_line = self.config_values.get('skip_first_line', 'False') == 'True'
        
        self.logger.info(f"Start separator: {start_sep}, End separator: {end_sep}, Skip first line: {skip_first_line}")
        
        # Display system and Assistant information.
        display_code(f"OS: '{os_name}', Language: '{self.INTERPRETER_LANGUAGE}', Mode: '{mode}' Model: '{self.INTERPRRETER_MODEL}'")
        display_markdown_message("Welcome to the **Interpreter**. I'm here to **assist** you with your everyday tasks. "
                                  "\nPlease enter your task and I'll do my best to help you out.")
        
        while True:
            try:
                
                task = input("> ")
                if task.lower() in ['exit', 'quit']:
                    break
                prompt = self.handle_mode(task, os_name)
                
                # Clean the responses
                self._clean_responses()
                
                # If graph were requested.
                if 'graph' in prompt.lower():
                    if self.INTERPRETER_LANGUAGE == 'python':
                        prompt += "\n" + "using Python use Matplotlib save the graph in file called 'graph.png'"
                    elif self.INTERPRETER_LANGUAGE == 'javascript':
                        prompt += "\n" + "using JavaScript use Chart.js save the graph in file called 'graph.png'"

                # if Chart were requested
                if 'chart' in prompt.lower() or 'plot' in prompt.lower():
                    if self.INTERPRETER_LANGUAGE == 'python':
                        prompt += "\n" + "using Python use Plotly save the chart in file called 'chart.png'"
                    elif self.INTERPRETER_LANGUAGE == 'javascript':
                        prompt += "\n" + "using JavaScript use Chart.js save the chart in file called 'chart.png'"

                # if Table were requested
                if 'table' in prompt.lower():
                    if self.INTERPRETER_LANGUAGE == 'python':
                        prompt += "\n" + "using Python use Pandas save the table in file called 'table.md'"
                    elif self.INTERPRETER_LANGUAGE == 'javascript':
                        prompt += "\n" + "using JavaScript use DataTables save the table in file called 'table.html'"
                    
                self.logger.info(f"Prompt: {prompt}")
                generated_output = self.generate_text(prompt, self.history, config_values=self.config_values)
                
                self.logger.info(f"Generated output type {type(generated_output)}")
                extracted_code = self.code_interpreter.extract_code(generated_output, start_sep, end_sep, skip_first_line,self.CODE_MODE)
                
                self.logger.info(f"Extracted code: {extracted_code[:50]}")

                
                if self.DISPLAY_CODE:
                    display_code(extracted_code)
                    self.logger.info("Code extracted successfully.")
                
                if extracted_code:
                    current_time = time.strftime("%Y_%m_%d-%H_%M_%S", time.localtime())
                    
                    if self.INTERPRETER_LANGUAGE == 'javascript' and self.SAVE_CODE and self.CODE_MODE:
                        self.code_interpreter.save_code(f"output/code_{current_time}.js", extracted_code)
                        self.logger.info(f"JavaScript code saved successfully.")
                    
                    elif self.INTERPRETER_LANGUAGE == 'python' and self.SAVE_CODE and self.CODE_MODE:
                        self.code_interpreter.save_code(f"output/code_{current_time}.py", extracted_code)
                        self.logger.info(f"Python code saved successfully.")
                    
                    elif self.SAVE_CODE and self.COMMAND_MODE:
                        self.code_interpreter.save_code(f"output/command_{current_time}.txt", extracted_code)
                        self.logger.info(f"Command saved successfully.")
  
                    elif self.SAVE_CODE and self.SCRIPT_MODE:
                        self.code_interpreter.save_code(f"output/script_{current_time}.txt", extracted_code)
                        self.logger.info(f"Script saved successfully.")
                  
                    # Execute the code if the user has selected.
                    code_output, code_error = self.execute_code(extracted_code, os_name)
                    
                    if code_output:
                        self.logger.info(f"{self.INTERPRETER_LANGUAGE} code executed successfully.")
                        display_code(code_output)
                        self.logger.info(f"Output: {code_output[:100]}")
                    elif code_error:
                        self.logger.info(f"Python code executed with error.")
                        display_markdown_message(f"Error: {code_error}")
                        
                    # install Package on error.
                    error_messages = ["ModuleNotFound", "ImportError", "No module named", "Cannot find module"]
                    if code_error is not None and any(error_message in code_error for error_message in error_messages):
                        package_name = self.package_installer.extract_package_name(code_error, self.INTERPRETER_LANGUAGE)
                        if package_name:
                            self.logger.info(f"Installing package {package_name} on interpreter {self.INTERPRETER_LANGUAGE}")
                            self.package_installer.install_package(package_name, self.INTERPRETER_LANGUAGE)

                    try:
                        # Check if graph.png exists and open it.
                        self._open_resource_file('graph.png')
                        
                        # Check if chart.png exists and open it.
                        self._open_resource_file('chart.png')
                        
                        # Check if table.md exists and open it.
                        self._open_resource_file('table.md')
                    except Exception as exception:
                        display_markdown_message(f"Error in opening resource files: {str(exception)}")
                
                self.utility_manager.save_history_json(task, command_mode, os_name, self.INTERPRETER_LANGUAGE, prompt, extracted_code, self.INTERPRRETER_MODEL)
                
            except Exception as exception:
                self.logger.error(f"An error occurred: {str(exception)}")
                raise
libs/interpreter_lib.py
haseeb-heaven/Open-Code-Interpreter
import logging
from logging.handlers import RotatingFileHandler

def initialize_logger(filename):
    # Define the logging format
    log_format = ("%(asctime)s [%(levelname)s] [%(filename)s:%(lineno)d] [%(funcName)s] - %(message)s")
    
    # Create a rotating file handler that will manage log file sizes and backups
    file_handler = RotatingFileHandler(filename, maxBytes=5*1024*1024)  # 5MB per file
    file_handler.setFormatter(logging.Formatter(log_format))
    
    # add logger for console
    console_handler = logging.StreamHandler()
    console_handler.setFormatter(logging.Formatter(log_format))
    # Create the logger
    logger = logging.getLogger()
    logger.setLevel(logging.INFO)
    
    # Check if the logger already has handlers, if not, add the file handler
    if not logger.handlers:
        logger.addHandler(file_handler)
    
    return logger
libs/logger.py
haseeb-heaven/Open-Code-Interpreter
import time
from rich import print as rich_print
from rich.markdown import Markdown
from rich.rule import Rule
from rich.syntax import Syntax
from pygments.formatters import TerminalFormatter

def display_markdown_message(message):
    """
    Display markdown message. Works with multiline strings with lots of indentation.
    Will automatically make single line > tags beautiful.
    """

    for line in message.split("\n"):
        line = line.strip()
        if line == "":
            print("")
        elif line == "---":
            rich_print(Rule(style="white"))
        else:
            rich_print(Markdown(line))

    if "\n" not in message and message.startswith(">"):
        # Aesthetic choice. For these tags, they need a space below them
        print("")
        

def display_code(code: list, language: str = "python"):
    try:
        syntax = Syntax(code, language, theme="monokai", line_numbers=True)
        rich_print(syntax,end="",flush=True)
    except Exception as exception:
        print(f"An error occurred: {exception}")
        

class CustomFormatter(TerminalFormatter):
    def format(self, tokensource, outfile):
        # call the parent method
        super().format(tokensource, outfile)
        # remove the trailing newline from the output file
        outfile.seek(outfile.tell() - 1)
        if outfile.read() == '\n':
            outfile.truncate()

from rich.console import Console

def display_code_stream(stream):
    """
    This function prints each token in the stream as a Python syntax highlighted code block.
    :param stream: The stream of text to be printed.
    """
    try:
        code = ""
        console = Console(record=True)  # Create a Console object that records print calls

        for output in stream:
            output_code = output.token.text
            output_code = output_code.replace("```","") if output_code else output_code
            output_code = output_code.replace("</s>","") if output_code else output_code
            if output_code  == '\n':
                highlighted_code = Syntax(code, "python", theme="monokai", line_numbers=False,word_wrap=True)
                console.print(highlighted_code)  # Print to the console object
                code = ""
            else:
                code += output_code + ""
            time.sleep(0.03)

        # Check if there is any remaining code that hasn't been printed
        if code:
            highlighted_code = Syntax(code, "python", theme="monokai", line_numbers=False,word_wrap=True)
            console.print(highlighted_code)  # Print to the console object
            code = ""
            
        return console.export_text().strip()  # Return the recorded text
    except Exception as exception:
        print(f"Error while printing as markdown: {exception}")
libs/markdown_code.py
haseeb-heaven/Open-Code-Interpreter
import subprocess
import re
import logging

import requests

class PackageInstaller:
    def __init__(self):
        self.pip_command = "pip"
        self.pip3_command = "pip3"
        self.npm_command = "npm"
        self.logger = logging.getLogger(__name__)
        handler = logging.StreamHandler()
        formatter = logging.Formatter('%(asctime)s [%(levelname)s] [%(name)s:%(lineno)d] [%(funcName)s] - %(message)s')
        handler.setFormatter(formatter)
        self.logger.addHandler(handler)
        self.logger.setLevel(logging.INFO)

    def install_package(self, package_name, language):
        if language == "python":
            if not self._check_package_exists_pip(package_name):
                exception = ValueError(f"Package {package_name} does not exist")
                self.logger.error(exception)
                raise exception
                
            if not self._is_package_installed(package_name,"pip"):
                try:
                    # Try to install the package using pip
                    self._install_package_with_pip(package_name)
                except subprocess.CalledProcessError:
                    try:
                        # If pip fails, try to install the package using pip3
                        self._install_package_with_pip3(package_name)
                    except subprocess.CalledProcessError as exception:
                        self.logger.error(f"Failed to install package with both pip and pip3: {package_name}")
                        raise exception
            else:
                self.logger.info(f"Package {package_name} is already installed")
        elif language == "javascript":
            try:
                if not self._check_package_exists_npm(package_name):
                    exception = ValueError(f"Package {package_name} does not exist")
                    self.logger.error(exception)
                    raise exception
            
                if not self._is_package_installed(package_name,"npm"):
                    try:
                        # Try to install the package using npm
                        self._install_package_with_npm(package_name)
                    except subprocess.CalledProcessError as exception:
                        raise exception
                    
            except subprocess.CalledProcessError as exception:
                self.logger.error(f"Failed to install package with npm: {package_name}")
                raise exception
        else:
            exception = ValueError("Invalid language selected.")
            self.logger.error(exception)
            raise exception
        
    def extract_package_name(self,error,language):
        if language == "python":
            return self._extract_python_package_name(error)
        elif language == "javascript":
            return self._extract_javascript_package_name(error)
        else:
            exception = ValueError("Invalid language selected.")
            self.logger.error(exception)
            raise exception

    def _install_package_with_pip(self, package_name):
        try:
            subprocess.check_call([self.pip_command, "install", package_name])
            self.logger.info(f"Successfully installed package with pip: {package_name}")
        except subprocess.CalledProcessError as exception:
            self.logger.error(f"Failed to install package with pip: {package_name}")
            raise exception

    def _install_package_with_pip3(self, package_name):
        try:
            subprocess.check_call([self.pip3_command, "install", package_name])
            self.logger.info(f"Successfully installed package with pip3: {package_name}")
        except subprocess.CalledProcessError as exception:
            self.logger.error(f"Failed to install package with pip3: {package_name}")
            raise exception
        
    def _install_package_with_npm(self, package_name):
        try:
            subprocess.check_call([self.npm_command, "install", package_name])
            self.logger.info(f"Successfully installed package with npm: {package_name}")
        except subprocess.CalledProcessError as exception:
            self.logger.error(f"Failed to install package with npm: {package_name}")
            raise exception
        
    def _is_package_installed(self, package_name, package_manager):
        if package_manager == "pip":
            try:
                subprocess.check_call([self.pip_command, "show", package_name])
                self.logger.info(f"Package {package_name} is installed")
                return True
            except subprocess.CalledProcessError:
                try:
                    subprocess.check_call([self.pip3_command, "show", package_name])
                    self.logger.info(f"Package {package_name} is installed")
                    return True
                except subprocess.CalledProcessError:
                    self.logger.info(f"Package {package_name} is not installed")
                    return False
        elif package_manager == "npm":
            try:
                subprocess.check_call([self.npm_command, "list", "-g", package_name])
                self.logger.info(f"Package {package_name} is installed")
                return True
            except subprocess.CalledProcessError:
                self.logger.info(f"Package {package_name} is not installed")
                return False
        else:
            exception = ValueError("Invalid package manager selected.")
            self.logger.error(exception)
            raise exception
    
    def _install_package_with_npm(self, package_name):
        try:
            subprocess.check_call([self.npm_command, "install", package_name])
            self.logger.info(f"Successfully installed package with npm: {package_name}")
        except subprocess.CalledProcessError as exception:
            self.logger.error(f"Failed to install package with npm: {package_name}")
            raise exception
        
    
    def _extract_python_package_name(self, error_message):
        # Regular expression pattern to match the error message
        pattern = r"ModuleNotFoundError: No module named '(\w+)'|ModuleNotFoundError: '(\w+)'"
        match = re.search(pattern, error_message)
        if match:
            # Extract the package name from the error message
            package_name = match.group(1) if match.group(1) else match.group(2)
            return package_name
        else:
            # If the package name could not be extracted, log an error and raise an exception
            exception = ValueError("Could not extract package name from error message")
            self.logger.error(exception)
            raise exception
    
    def _extract_javascript_package_name(self,error_message):
        try:
            lines = error_message.split('\n')
            for line in lines:
                if line.startswith("Error: Cannot find module"):
                    package_name = line.split("'")[1]
                    return package_name
            return None
        except Exception as exception:
            self.logger.error(f"Failed to extract package name from error message: {exception}")
            raise exception
        
    def _check_package_exists_pip(self,package_name):
        import requests
        from bs4 import BeautifulSoup
        
        try:
            # Send a GET request to the PyPI search page
            response = requests.get(f"https://pypi.org/search/?q={package_name}")
            response.raise_for_status()
            
            # Parse the HTML content of the page
            soup = BeautifulSoup(response.text, 'html.parser')
            
            # Search for the package name in the parsed HTML
            search_results = soup.find_all('span', class_='package-snippet__name')
            for result in search_results:
                if result.text.strip() == package_name:
                    return True
            
            # If the package name was not found in the search results, log an error and raise an exception
            exception = ValueError(f"Package {package_name} does not exist on PyPI")
            self.logger.error(exception)
            raise exception
        except requests.exceptions.RequestException as exception:
            raise exception
        
    def _check_package_exists_npm(self, package_name):
        try:
            response = requests.get(f"https://www.npmjs.com/package/{package_name}")
            if response.status_code == 200:
                self.logger.info(f"Package {package_name} exists on npm website")
                return True
            else:
                self.logger.info(f"Package {package_name} does not exist on npm website")
                return False
        except requests.exceptions.RequestException as exception:
            self.logger.error(f"Failed to check package existence on npm website: {exception}")
            raise exception

libs/package_installer.py
haseeb-heaven/Open-Code-Interpreter
import json
import os
from libs.logger import initialize_logger
import traceback

class UtilityManager:
    def __init__(self):
        try:
            if not os.path.exists('logs'):
                os.makedirs('logs')
            if not os.path.isfile('logs/interpreter.log'):
                open('logs/interpreter.log', 'w').close()
        except Exception as exception:
            self.logger.error(f"Error in UtilityManager initialization: {str(exception)}")
            raise
        self.logger = initialize_logger("logs/interpreter.log")

    def get_os_platform(self):
        try:
            import platform
            os_info = platform.uname()
            os_name = os_info.system

            os_name_mapping = {
                'Darwin': 'MacOS',
                'Linux': 'Linux',
                'Windows': 'Windows'
            }

            os_name = os_name_mapping.get(os_name, 'Other')

            self.logger.info(f"Operating System: {os_name} Version: {os_info.version}")
            return os_name, os_info.version
        except Exception as exception:
            self.logger.error(f"Error in getting OS platform: {str(exception)}")
            raise

    def save_history_json(self, task, mode, os_name, language, prompt, extracted_code, model_name, filename="history/history.json"):
        try:
            history_entry = {
                "Assistant": {
                    "Task": task,
                    "Mode": mode,
                    "OS": os_name,
                    "Language": language,
                    "Model": model_name
                },
                "User": prompt,
                "System": extracted_code
            }

            data = []
            if os.path.isfile(filename) and os.path.getsize(filename) > 0:
                with open(filename, "r") as history_file:  # Open the file in read mode
                    data = json.load(history_file)

            data.append(history_entry)

            with open(filename, "w") as history_file:
                json.dump(data, history_file)
        except Exception as exception:
            self.logger.error(f"Error in saving history to JSON: {str(exception)}")
            raise

    def initialize_readline_history(self):
        try:
            # Checking the OS type
            # If it's posix (Unix-like), import readline for handling lines from input
            # If it's not posix, import pyreadline as readline
            if os.name == 'posix':
                import readline
            else:
                import pyreadline as readline
                
            histfile = os.path.join(os.path.expanduser("~"), ".python_history")
            readline.read_history_file(histfile)
            
            # Save history to file on exit
            import atexit
            atexit.register(readline.write_history_file, histfile)
        except FileNotFoundError:
            pass
        except Exception as exception:
            self.logger.error(f"Error in initializing readline history: {str(exception)}")
            raise

    def read_config_file(self, filename=".config"):
        try:
            config_data = {}
            with open(filename, "r") as config_file:
                for line in config_file:
                    # Ignore comments and lines without an equals sign
                    if line.strip().startswith('#') or '=' not in line:
                        continue
                    key, value = line.strip().split("=")
                    config_data[key.strip()] = value.strip()
            return config_data
        except Exception as exception:
            self.logger.error(f"Error in reading config file: {str(exception)}")
            raise

libs/utility_manager.py
haseeb-heaven/Open-Code-Interpreter