Fix path processing (Significant-Gravitas#5032)

* Fix and clean up path processing in logs module

* Fix path processing throughout the project

* Fix plugins test

* Fix borky pytest vs mkdir(exist_ok=True)

* Update docs and gitignore for new workspace location

* Fix borky pytest vol.2

* ok james

.gitignore

@@ -1,7 +1,7 @@
 ## Original ignores
 autogpt/keys.py
 autogpt/*.json
-**/auto_gpt_workspace/*
+auto_gpt_workspace/*
 *.mpeg
 .env
 azure.yaml

autogpt/cli.py

@@ -1,4 +1,5 @@
 """Main script for the autogpt package."""
+from pathlib import Path
 from typing import Optional
 
 import click
@@ -115,24 +116,25 @@ def main(
 
     if ctx.invoked_subcommand is None:
         run_auto_gpt(
-            continuous,
-            continuous_limit,
-            ai_settings,
-            prompt_settings,
-            skip_reprompt,
-            speak,
-            debug,
-            gpt3only,
-            gpt4only,
-            memory_type,
-            browser_name,
-            allow_downloads,
-            skip_news,
-            workspace_directory,
-            install_plugin_deps,
-            ai_name,
-            ai_role,
-            ai_goal,
+            continuous=continuous,
+            continuous_limit=continuous_limit,
+            ai_settings=ai_settings,
+            prompt_settings=prompt_settings,
+            skip_reprompt=skip_reprompt,
+            speak=speak,
+            debug=debug,
+            gpt3only=gpt3only,
+            gpt4only=gpt4only,
+            memory_type=memory_type,
+            browser_name=browser_name,
+            allow_downloads=allow_downloads,
+            skip_news=skip_news,
+            working_directory=Path(__file__).parent.parent,  # TODO: make this an option
+            workspace_directory=workspace_directory,
+            install_plugin_deps=install_plugin_deps,
+            ai_name=ai_name,
+            ai_role=ai_role,
+            ai_goals=ai_goal,
         )
 
 

autogpt/commands/execute_code.py

@@ -150,7 +150,7 @@ def execute_python_file(filename: str, agent: Agent) -> str:
                 file_path.relative_to(agent.workspace.root).as_posix(),
             ],
             volumes={
-                agent.config.workspace_path: {
+                str(agent.config.workspace_path): {
                     "bind": "/workspace",
                     "mode": "rw",
                 }

autogpt/commands/image_gen.py

@@ -37,7 +37,7 @@ def generate_image(prompt: str, agent: Agent, size: int = 256) -> str:
     Returns:
         str: The filename of the image
     """
-    filename = f"{agent.config.workspace_path}/{str(uuid.uuid4())}.jpg"
+    filename = agent.config.workspace_path / f"{str(uuid.uuid4())}.jpg"
 
     # DALL-E
     if agent.config.image_provider == "dalle":

autogpt/config/ai_config.py

@@ -4,7 +4,6 @@
 """
 from __future__ import annotations
 
-import os
 import platform
 from pathlib import Path
 from typing import TYPE_CHECKING, Optional
@@ -16,9 +15,6 @@
     from autogpt.models.command_registry import CommandRegistry
     from autogpt.prompts.generator import PromptGenerator
 
-# Soon this will go in a folder where it remembers more stuff about the run(s)
-SAVE_FILE = str(Path(os.getcwd()) / "ai_settings.yaml")
-
 
 class AIConfig:
     """
@@ -57,14 +53,13 @@ def __init__(
         self.command_registry: CommandRegistry | None = None
 
     @staticmethod
-    def load(ai_settings_file: str = SAVE_FILE) -> "AIConfig":
+    def load(ai_settings_file: str | Path) -> "AIConfig":
         """
         Returns class object with parameters (ai_name, ai_role, ai_goals, api_budget)
         loaded from yaml file if yaml file exists, else returns class with no parameters.
 
         Parameters:
-            ai_settings_file (int): The path to the config yaml file.
-              DEFAULT: "../ai_settings.yaml"
+            ai_settings_file (Path): The path to the config yaml file.
 
         Returns:
             cls (object): An instance of given cls object
@@ -85,16 +80,15 @@ def load(ai_settings_file: str = SAVE_FILE) -> "AIConfig":
             for goal in config_params.get("ai_goals", [])
         ]
         api_budget = config_params.get("api_budget", 0.0)
-        # type: Type[AIConfig]
+
         return AIConfig(ai_name, ai_role, ai_goals, api_budget)
 
-    def save(self, ai_settings_file: str = SAVE_FILE) -> None:
+    def save(self, ai_settings_file: str | Path) -> None:
         """
         Saves the class parameters to the specified file yaml file path as a yaml file.
 
         Parameters:
-            ai_settings_file(str): The path to the config yaml file.
-              DEFAULT: "../ai_settings.yaml"
+            ai_settings_file (Path): The path to the config yaml file.
 
         Returns:
             None

autogpt/config/config.py

@@ -4,6 +4,7 @@
 import contextlib
 import os
 import re
+from pathlib import Path
 from typing import Any, Dict, Optional, Union
 
 import yaml
@@ -14,10 +15,8 @@
 from autogpt.core.configuration.schema import Configurable, SystemSettings
 from autogpt.plugins.plugins_config import PluginsConfig
 
-AZURE_CONFIG_FILE = os.path.join(os.path.dirname(__file__), "../..", "azure.yaml")
-PLUGINS_CONFIG_FILE = os.path.join(
-    os.path.dirname(__file__), "../..", "plugins_config.yaml"
-)
+AZURE_CONFIG_FILE = "azure.yaml"
+PLUGINS_CONFIG_FILE = "plugins_config.yaml"
 GPT_4_MODEL = "gpt-4"
 GPT_3_MODEL = "gpt-3.5-turbo"
 
@@ -47,7 +46,8 @@ class Config(SystemSettings, arbitrary_types_allowed=True):
     # Paths
     ai_settings_file: str = "ai_settings.yaml"
     prompt_settings_file: str = "prompt_settings.yaml"
-    workspace_path: Optional[str] = None
+    workdir: Path = None
+    workspace_path: Optional[Path] = None
     file_logger_path: Optional[str] = None
     # Model configuration
     fast_llm: str = "gpt-3.5-turbo"
@@ -210,9 +210,10 @@ class ConfigBuilder(Configurable[Config]):
     default_settings = Config()
 
     @classmethod
-    def build_config_from_env(cls) -> Config:
+    def build_config_from_env(cls, workdir: Path) -> Config:
         """Initialize the Config class"""
         config_dict = {
+            "workdir": workdir,
             "authorise_key": os.getenv("AUTHORISE_COMMAND_KEY"),
             "exit_key": os.getenv("EXIT_KEY"),
             "plain_output": os.getenv("PLAIN_OUTPUT", "False") == "True",
@@ -299,7 +300,9 @@ def build_config_from_env(cls) -> Config:
             config_dict["temperature"] = float(os.getenv("TEMPERATURE"))
 
         if config_dict["use_azure"]:
-            azure_config = cls.load_azure_config(config_dict["azure_config_file"])
+            azure_config = cls.load_azure_config(
+                workdir / config_dict["azure_config_file"]
+            )
             config_dict.update(azure_config)
 
         elif os.getenv("OPENAI_API_BASE_URL"):
@@ -318,21 +321,21 @@ def build_config_from_env(cls) -> Config:
         # Set secondary config variables (that depend on other config variables)
 
         config.plugins_config = PluginsConfig.load_config(
-            config.plugins_config_file,
+            config.workdir / config.plugins_config_file,
             config.plugins_denylist,
             config.plugins_allowlist,
         )
 
         return config
 
     @classmethod
-    def load_azure_config(cls, config_file: str = AZURE_CONFIG_FILE) -> Dict[str, str]:
+    def load_azure_config(cls, config_file: Path) -> Dict[str, str]:
         """
         Loads the configuration parameters for Azure hosting from the specified file
           path as a yaml file.
 
         Parameters:
-            config_file(str): The path to the config yaml file. DEFAULT: "../azure.yaml"
+            config_file (Path): The path to the config yaml file.
 
         Returns:
             Dict

autogpt/logs/handlers.py

@@ -2,6 +2,7 @@
 import logging
 import random
 import time
+from pathlib import Path
 
 
 class ConsoleHandler(logging.StreamHandler):
@@ -38,7 +39,7 @@ def emit(self, record: logging.LogRecord):
 
 
 class JsonFileHandler(logging.FileHandler):
-    def __init__(self, filename: str, mode="a", encoding=None, delay=False):
+    def __init__(self, filename: str | Path, mode="a", encoding=None, delay=False):
         super().__init__(filename, mode, encoding, delay)
 
     def emit(self, record: logging.LogRecord):

autogpt/logs/log_cycle.py

@@ -1,5 +1,6 @@
 import json
 import os
+from pathlib import Path
 from typing import Any, Dict, Union
 
 from .logger import logger
@@ -23,38 +24,33 @@ class LogCycleHandler:
     def __init__(self):
         self.log_count_within_cycle = 0
 
-    @staticmethod
-    def create_directory_if_not_exists(directory_path: str) -> None:
-        if not os.path.exists(directory_path):
-            os.makedirs(directory_path, exist_ok=True)
-
-    def create_outer_directory(self, ai_name: str, created_at: str) -> str:
-        log_directory = logger.get_log_directory()
-
+    def create_outer_directory(self, ai_name: str, created_at: str) -> Path:
         if os.environ.get("OVERWRITE_DEBUG") == "1":
             outer_folder_name = "auto_gpt"
         else:
             ai_name_short = self.get_agent_short_name(ai_name)
             outer_folder_name = f"{created_at}_{ai_name_short}"
 
-        outer_folder_path = os.path.join(log_directory, "DEBUG", outer_folder_name)
-        self.create_directory_if_not_exists(outer_folder_path)
+        outer_folder_path = logger.log_dir / "DEBUG" / outer_folder_name
+        if not outer_folder_path.exists():
+            outer_folder_path.mkdir(parents=True)
 
         return outer_folder_path
 
     def get_agent_short_name(self, ai_name: str) -> str:
         return ai_name[:15].rstrip() if ai_name else DEFAULT_PREFIX
 
-    def create_inner_directory(self, outer_folder_path: str, cycle_count: int) -> str:
+    def create_inner_directory(self, outer_folder_path: Path, cycle_count: int) -> Path:
         nested_folder_name = str(cycle_count).zfill(3)
-        nested_folder_path = os.path.join(outer_folder_path, nested_folder_name)
-        self.create_directory_if_not_exists(nested_folder_path)
+        nested_folder_path = outer_folder_path / nested_folder_name
+        if not nested_folder_path.exists():
+            nested_folder_path.mkdir()
 
         return nested_folder_path
 
     def create_nested_directory(
         self, ai_name: str, created_at: str, cycle_count: int
-    ) -> str:
+    ) -> Path:
         outer_folder_path = self.create_outer_directory(ai_name, created_at)
         nested_folder_path = self.create_inner_directory(outer_folder_path, cycle_count)
 
@@ -75,14 +71,10 @@ def log_cycle(
             data (Any): The data to be logged.
             file_name (str): The name of the file to save the logged data.
         """
-        nested_folder_path = self.create_nested_directory(
-            ai_name, created_at, cycle_count
-        )
+        cycle_log_dir = self.create_nested_directory(ai_name, created_at, cycle_count)
 
         json_data = json.dumps(data, ensure_ascii=False, indent=4)
-        log_file_path = os.path.join(
-            nested_folder_path, f"{self.log_count_within_cycle}_{file_name}"
-        )
+        log_file_path = cycle_log_dir / f"{self.log_count_within_cycle}_{file_name}"
 
         logger.log_json(json_data, log_file_path)
         self.log_count_within_cycle += 1

autogpt/logs/logger.py

@@ -2,7 +2,7 @@
 from __future__ import annotations
 
 import logging
-import os
+from pathlib import Path
 from typing import TYPE_CHECKING, Any, Optional
 
 from colorama import Fore
@@ -25,10 +25,10 @@ class Logger(metaclass=Singleton):
 
     def __init__(self):
         # create log directory if it doesn't exist
-        this_files_dir_path = os.path.dirname(__file__)
-        log_dir = os.path.join(this_files_dir_path, "../logs")
-        if not os.path.exists(log_dir):
-            os.makedirs(log_dir)
+        # TODO: use workdir from config
+        self.log_dir = Path(__file__).parent.parent.parent / "logs"
+        if not self.log_dir.exists():
+            self.log_dir.mkdir()
 
         log_file = "activity.log"
         error_file = "error.log"
@@ -46,19 +46,15 @@ def __init__(self):
         self.console_handler.setFormatter(console_formatter)
 
         # Info handler in activity.log
-        self.file_handler = logging.FileHandler(
-            os.path.join(log_dir, log_file), "a", "utf-8"
-        )
+        self.file_handler = logging.FileHandler(self.log_dir / log_file, "a", "utf-8")
         self.file_handler.setLevel(logging.DEBUG)
         info_formatter = AutoGptFormatter(
             "%(asctime)s %(levelname)s %(title)s %(message_no_color)s"
         )
         self.file_handler.setFormatter(info_formatter)
 
         # Error handler error.log
-        error_handler = logging.FileHandler(
-            os.path.join(log_dir, error_file), "a", "utf-8"
-        )
+        error_handler = logging.FileHandler(self.log_dir / error_file, "a", "utf-8")
         error_handler.setLevel(logging.ERROR)
         error_formatter = AutoGptFormatter(
             "%(asctime)s %(levelname)s %(module)s:%(funcName)s:%(lineno)d %(title)s"
@@ -179,13 +175,9 @@ def double_check(self, additionalText: Optional[str] = None) -> None:
 
         self.typewriter_log("DOUBLE CHECK CONFIGURATION", Fore.YELLOW, additionalText)
 
-    def log_json(self, data: Any, file_name: str) -> None:
-        # Define log directory
-        this_files_dir_path = os.path.dirname(__file__)
-        log_dir = os.path.join(this_files_dir_path, "../logs")
-
+    def log_json(self, data: Any, file_name: str | Path) -> None:
         # Create a handler for JSON files
-        json_file_path = os.path.join(log_dir, file_name)
+        json_file_path = self.log_dir / file_name
         json_data_handler = JsonFileHandler(json_file_path)
         json_data_handler.setFormatter(JsonFormatter())
 
@@ -194,10 +186,5 @@ def log_json(self, data: Any, file_name: str) -> None:
         self.json_logger.debug(data)
         self.json_logger.removeHandler(json_data_handler)
 
-    def get_log_directory(self) -> str:
-        this_files_dir_path = os.path.dirname(__file__)
-        log_dir = os.path.join(this_files_dir_path, "../../logs")
-        return os.path.abspath(log_dir)
-
 
 logger = Logger()