oxbrbtx/PrimAITE
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

#1647 - Added _PrimaitePaths class that manages all the primaite locations using PlayformDirs. This class now creates new primaite locations for each version of primaite.

Browse Source 
- Rolled the _PrimaitePaths class out throughout the code base.
- Updated the docs to reference the new version paths.
- Updated the author from qinetiq to dstl
- Bumped version number to 2.0.0rc2
This commit is contained in:
Chris McCarthy
2023-07-21 14:00:50 +01:00
parent de8f847afd
commit 534c066170
19 changed files with 192 additions and 185 deletions
Download Patch File Download Diff File Expand all files Collapse all files

2
src/primaite/VERSION
View File

@@ -1 +1 @@
2.0.0rc1
2.0.0rc2

164
src/primaite/__init__.py
View File

@@ -1,23 +1,126 @@
# Crown Owned Copyright (C) Dstl 2023. DEFCON 703. Shared in confidence.
import logging
import logging.config
import shutil
import sys
from bisect import bisect
from logging import Formatter, Logger, LogRecord, StreamHandler
from logging.handlers import RotatingFileHandler
from pathlib import Path
from typing import Any, Dict, Final
from typing import Any, Dict, Final, List
import pkg_resources
import yaml
from platformdirs import PlatformDirs
_PLATFORM_DIRS: Final[PlatformDirs] = PlatformDirs(appname="primaite")
"""An instance of `PlatformDirs` set with appname='primaite'."""
with open(Path(__file__).parent.resolve() / "VERSION", "r") as file:
__version__ = file.readline().strip()
class _PrimaitePaths:
"""
A Primaite paths class that leverages PlatformDirs.
The PlatformDirs appname is 'primaite' and the version is ``primaite.__version__`.
"""
def __init__(self):
self._dirs: Final[PlatformDirs] = PlatformDirs(appname="primaite", version=__version__)
def _get_dirs_properties(self) -> List[str]:
class_items = self.__class__.__dict__.items()
return [k for k, v in class_items if isinstance(v, property)]
def mkdirs(self):
"""
Creates all Primaite directories.
Does this by retrieving all properties in the PrimaiteDirs class and calls each one.
"""
for p in self._get_dirs_properties():
getattr(self, p)
@property
def user_home_path(self) -> Path:
"""The PrimAITE user home path."""
path = Path.home() / "primaite" / __version__
path.mkdir(exist_ok=True, parents=True)
return path
@property
def user_sessions_path(self) -> Path:
"""The PrimAITE user sessions path."""
path = self.user_home_path / "sessions"
path.mkdir(exist_ok=True, parents=True)
return path
@property
def user_config_path(self) -> Path:
"""The PrimAITE user config path."""
path = self.user_home_path / "config"
path.mkdir(exist_ok=True, parents=True)
return path
@property
def user_notebooks_path(self) -> Path:
"""The PrimAITE user notebooks path."""
path = self.user_home_path / "notebooks"
path.mkdir(exist_ok=True, parents=True)
return path
@property
def app_home_path(self) -> Path:
"""The PrimAITE app home path."""
path = self._dirs.user_data_path
path.mkdir(exist_ok=True, parents=True)
return path
@property
def app_config_dir_path(self) -> Path:
"""The PrimAITE app config directory path."""
path = self._dirs.user_config_path
path.mkdir(exist_ok=True, parents=True)
return path
@property
def app_config_file_path(self) -> Path:
"""The PrimAITE app config file path."""
return self.app_config_dir_path / "primaite_config.yaml"
@property
def app_log_dir_path(self) -> Path:
"""The PrimAITE app log directory path."""
if sys.platform == "win32":
path = self.app_home_path / "logs"
else:
path = self._dirs.user_log_path
path.mkdir(exist_ok=True, parents=True)
return path
@property
def app_log_file_path(self) -> Path:
"""The PrimAITE app log file path."""
return self.app_log_dir_path / "primaite.log"
def __repr__(self):
properties_str = ", ".join([f"{p}='{getattr(self, p)}'" for p in self._get_dirs_properties()])
return f"{self.__class__.__name__}({properties_str})"
PRIMAITE_PATHS: Final[_PrimaitePaths] = _PrimaitePaths()
def _host_primaite_config():
if not PRIMAITE_PATHS.app_config_file_path.exists():
pkg_config_path = Path(pkg_resources.resource_filename("primaite", "setup/_package_data/primaite_config.yaml"))
shutil.copy2(pkg_config_path, PRIMAITE_PATHS.app_config_file_path)
_host_primaite_config()
def _get_primaite_config() -> Dict:
config_path = _PLATFORM_DIRS.user_config_path / "primaite_config.yaml"
config_path = PRIMAITE_PATHS.app_config_file_path
if not config_path.exists():
config_path = Path(pkg_resources.resource_filename("primaite", "setup/_package_data/primaite_config.yaml"))
with open(config_path, "r") as file:
@@ -36,35 +139,7 @@ def _get_primaite_config() -> Dict:
_PRIMAITE_CONFIG = _get_primaite_config()
_USER_DIRS: Final[Path] = Path.home() / "primaite"
"""The users home space for PrimAITE which is located at: ~/primaite."""
NOTEBOOKS_DIR: Final[Path] = _USER_DIRS / "notebooks"
"""
The path to the users notebooks directory as an instance of `Path` or
`PosixPath`, depending on the OS.
Users notebooks are stored at: ``~/primaite/notebooks``.
"""
USERS_CONFIG_DIR: Final[Path] = _USER_DIRS / "config"
"""
The path to the users config directory as an instance of `Path` or
`PosixPath`, depending on the OS.
Users config files are stored at: ``~/primaite/config``.
"""
SESSIONS_DIR: Final[Path] = _USER_DIRS / "sessions"
"""
The path to the users PrimAITE Sessions directory as an instance of `Path` or
`PosixPath`, depending on the OS.
Users PrimAITE Sessions are stored at: ``~/primaite/sessions``.
"""
# region Setup Logging
class _LevelFormatter(Formatter):
"""
A custom level-specific formatter.
@@ -87,14 +162,6 @@ class _LevelFormatter(Formatter):
return formatter.format(record)
def _log_dir() -> Path:
if sys.platform == "win32":
dir_path = _PLATFORM_DIRS.user_data_path / "logs"
else:
dir_path = _PLATFORM_DIRS.user_log_path
return dir_path
_LEVEL_FORMATTER: Final[_LevelFormatter] = _LevelFormatter(
{
logging.DEBUG: _PRIMAITE_CONFIG["logging"]["logger_format"]["DEBUG"],
@@ -105,18 +172,10 @@ _LEVEL_FORMATTER: Final[_LevelFormatter] = _LevelFormatter(
}
)
LOG_DIR: Final[Path] = _log_dir()
"""The path to the app log directory as an instance of `Path` or `PosixPath`, depending on the OS."""
LOG_DIR.mkdir(exist_ok=True, parents=True)
LOG_PATH: Final[Path] = LOG_DIR / "primaite.log"
"""The primaite.log file path as an instance of `Path` or `PosixPath`, depending on the OS."""
_STREAM_HANDLER: Final[StreamHandler] = StreamHandler()
_FILE_HANDLER: Final[RotatingFileHandler] = RotatingFileHandler(
filename=LOG_PATH,
filename=PRIMAITE_PATHS.app_log_file_path,
maxBytes=10485760, # 10MB
backupCount=9, # Max 100MB of logs
encoding="utf8",
@@ -146,10 +205,3 @@ def getLogger(name: str) -> Logger: # noqa
logger.setLevel(_PRIMAITE_CONFIG["log_level"])
return logger
# endregion
with open(Path(__file__).parent.resolve() / "VERSION", "r") as file:
__version__ = file.readline().strip()

4
src/primaite/agents/agent_abc.py
View File

@@ -10,7 +10,7 @@ from typing import Any, Dict, Optional, Union
from uuid import uuid4
import primaite
from primaite import getLogger, SESSIONS_DIR
from primaite import getLogger, PRIMAITE_PATHS
from primaite.config import lay_down_config, training_config
from primaite.config.training_config import TrainingConfig
from primaite.data_viz.session_plots import plot_av_reward_per_episode
@@ -32,7 +32,7 @@ def get_session_path(session_timestamp: datetime) -> Path:
"""
date_dir = session_timestamp.strftime("%Y-%m-%d")
session_path = session_timestamp.strftime("%Y-%m-%d_%H-%M-%S")
session_path = SESSIONS_DIR / date_dir / session_path
session_path = PRIMAITE_PATHS.user_sessions_path / date_dir / session_path
session_path.mkdir(exist_ok=True, parents=True)
return session_path

41
src/primaite/cli.py
View File

@@ -2,17 +2,14 @@
"""Provides a CLI using Typer as an entry point."""
import logging
import os
import shutil
from enum import Enum
from pathlib import Path
from typing import Optional
import pkg_resources
import typer
import yaml
from platformdirs import PlatformDirs
from typing_extensions import Annotated
from primaite import PRIMAITE_PATHS
from primaite.data_viz import PlotlyTemplate
app = typer.Typer()
@@ -47,10 +44,10 @@ def logs(last_n: Annotated[int, typer.Option("-n")]) -> None:
"""
import re
from primaite import LOG_PATH
from primaite import PRIMAITE_PATHS
if os.path.isfile(LOG_PATH):
with open(LOG_PATH) as file:
if os.path.isfile(PRIMAITE_PATHS.app_log_file_path):
with open(PRIMAITE_PATHS.app_log_file_path) as file:
lines = file.readlines()
for line in lines[-last_n:]:
print(re.sub(r"\n*", "", line))
@@ -70,16 +67,13 @@ def log_level(level: Annotated[Optional[_LogLevel], typer.Argument()] = None) ->
For example, to set the to debug, call: primaite log-level DEBUG
"""
app_dirs = PlatformDirs(appname="primaite")
app_dirs.user_config_path.mkdir(exist_ok=True, parents=True)
user_config_path = app_dirs.user_config_path / "primaite_config.yaml"
if user_config_path.exists():
with open(user_config_path, "r") as file:
if PRIMAITE_PATHS.app_config_file_path.exists():
with open(PRIMAITE_PATHS.app_config_file_path, "r") as file:
primaite_config = yaml.safe_load(file)
if level:
primaite_config["logging"]["log_level"] = level.value
with open(user_config_path, "w") as file:
with open(PRIMAITE_PATHS.app_config_file_path, "w") as file:
yaml.dump(primaite_config, file)
print(f"PrimAITE Log Level: {level}")
else:
@@ -118,16 +112,8 @@ def setup(overwrite_existing: bool = True) -> None:
WARNING: All user-data will be lost.
"""
# Does this way to avoid using PrimAITE package before config is loaded
app_dirs = PlatformDirs(appname="primaite")
app_dirs.user_config_path.mkdir(exist_ok=True, parents=True)
user_config_path = app_dirs.user_config_path / "primaite_config.yaml"
pkg_config_path = Path(pkg_resources.resource_filename("primaite", "setup/_package_data/primaite_config.yaml"))
shutil.copy2(pkg_config_path, user_config_path)
from primaite import getLogger
from primaite.setup import old_installation_clean_up, reset_demo_notebooks, reset_example_configs, setup_app_dirs
from primaite.setup import old_installation_clean_up, reset_demo_notebooks, reset_example_configs
_LOGGER = getLogger(__name__)
@@ -136,7 +122,7 @@ def setup(overwrite_existing: bool = True) -> None:
_LOGGER.info("Building primaite_config.yaml...")
_LOGGER.info("Building the PrimAITE app directories...")
setup_app_dirs.run()
PRIMAITE_PATHS.mkdirs()
_LOGGER.info("Rebuilding the demo notebooks...")
reset_demo_notebooks.run(overwrite_existing=True)
@@ -195,16 +181,13 @@ def plotly_template(template: Annotated[Optional[PlotlyTemplate], typer.Argument
For example, to set as plotly_dark, call: primaite plotly-template PLOTLY_DARK
"""
app_dirs = PlatformDirs(appname="primaite")
app_dirs.user_config_path.mkdir(exist_ok=True, parents=True)
user_config_path = app_dirs.user_config_path / "primaite_config.yaml"
if user_config_path.exists():
with open(user_config_path, "r") as file:
if PRIMAITE_PATHS.app_config_file_path.exists():
with open(PRIMAITE_PATHS.app_config_file_path, "r") as file:
primaite_config = yaml.safe_load(file)
if template:
primaite_config["session"]["outputs"]["plots"]["template"] = template.value
with open(user_config_path, "w") as file:
with open(PRIMAITE_PATHS.app_config_file_path, "w") as file:
yaml.dump(primaite_config, file)
print(f"PrimAITE plotly template: {template.value}")
else:

4
src/primaite/config/lay_down_config.py
View File

@@ -5,11 +5,11 @@ from typing import Any, Dict, Final, Union
import yaml
from primaite import getLogger, USERS_CONFIG_DIR
from primaite import getLogger, PRIMAITE_PATHS
_LOGGER: Logger = getLogger(__name__)
_EXAMPLE_LAY_DOWN: Final[Path] = USERS_CONFIG_DIR / "example_config" / "lay_down"
_EXAMPLE_LAY_DOWN: Final[Path] = PRIMAITE_PATHS.user_config_path / "example_config" / "lay_down"
def convert_legacy_lay_down_config_dict(legacy_config_dict: Dict[str, Any]) -> Dict[str, Any]:

4
src/primaite/config/training_config.py
View File

@@ -8,7 +8,7 @@ from typing import Any, Dict, Final, Optional, Union
import yaml
from primaite import getLogger, USERS_CONFIG_DIR
from primaite import getLogger, PRIMAITE_PATHS
from primaite.common.enums import (
ActionType,
AgentFramework,
@@ -22,7 +22,7 @@ from primaite.common.enums import (
_LOGGER: Logger = getLogger(__name__)
_EXAMPLE_TRAINING: Final[Path] = USERS_CONFIG_DIR / "example_config" / "training"
_EXAMPLE_TRAINING: Final[Path] = PRIMAITE_PATHS.user_config_path / "example_config" / "training"
def main_training_config_path() -> Path:

5
src/primaite/data_viz/session_plots.py
View File

@@ -7,13 +7,12 @@ import polars as pl
import yaml
from plotly.graph_objs import Figure
from primaite import _PLATFORM_DIRS
from primaite import PRIMAITE_PATHS
def get_plotly_config() -> Dict:
"""Get the plotly config from primaite_config.yaml."""
user_config_path = _PLATFORM_DIRS.user_config_path / "primaite_config.yaml"
with open(user_config_path, "r") as file:
with open(PRIMAITE_PATHS.app_config_file_path, "r") as file:
primaite_config = yaml.safe_load(file)
return primaite_config["session"]["outputs"]["plots"]

1
src/primaite/exceptions.py
View File

@@ -1,3 +1,4 @@
# Crown Owned Copyright (C) Dstl 2023. DEFCON 703. Shared in confidence.
class PrimaiteError(Exception):
"""The root PrimAITe Error."""

4
src/primaite/notebooks/__init__.py
View File

@@ -7,7 +7,7 @@ import subprocess
import sys
from logging import Logger
from primaite import getLogger, NOTEBOOKS_DIR
from primaite import getLogger, PRIMAITE_PATHS
_LOGGER: Logger = getLogger(__name__)
@@ -26,7 +26,7 @@ def start_jupyter_session() -> None:
jupyter_cmd = "jupyter lab"
working_dir = os.getcwd()
os.chdir(NOTEBOOKS_DIR)
os.chdir(PRIMAITE_PATHS.user_notebooks_path)
subprocess.Popen(jupyter_cmd)
os.chdir(working_dir)
else:

4
src/primaite/setup/reset_demo_notebooks.py
View File

@@ -7,7 +7,7 @@ from pathlib import Path
import pkg_resources
from primaite import getLogger, NOTEBOOKS_DIR
from primaite import getLogger, PRIMAITE_PATHS
_LOGGER: Logger = getLogger(__name__)
@@ -23,7 +23,7 @@ def run(overwrite_existing: bool = True) -> None:
for file in files:
fp = os.path.join(subdir, file)
path_split = os.path.relpath(fp, notebooks_package_data_root).split(os.sep)
target_fp = NOTEBOOKS_DIR / Path(*path_split)
target_fp = PRIMAITE_PATHS.user_notebooks_path / Path(*path_split)
target_fp.parent.mkdir(exist_ok=True, parents=True)
copy_file = not target_fp.is_file()

4
src/primaite/setup/reset_example_configs.py
View File

@@ -6,7 +6,7 @@ from pathlib import Path
import pkg_resources
from primaite import getLogger, USERS_CONFIG_DIR
from primaite import getLogger, PRIMAITE_PATHS
_LOGGER = getLogger(__name__)
@@ -23,7 +23,7 @@ def run(overwrite_existing: bool = True) -> None:
for file in files:
fp = os.path.join(subdir, file)
path_split = os.path.relpath(fp, configs_package_data_root).split(os.sep)
target_fp = USERS_CONFIG_DIR / "example_config" / Path(*path_split)
target_fp = PRIMAITE_PATHS.user_config_path / "example_config" / Path(*path_split)
target_fp.parent.mkdir(exist_ok=True, parents=True)
copy_file = not target_fp.is_file()

29
src/primaite/setup/setup_app_dirs.py
View File

@@ -1,29 +0,0 @@
# Crown Owned Copyright (C) Dstl 2023. DEFCON 703. Shared in confidence.
from logging import Logger
from primaite import _USER_DIRS, getLogger, LOG_DIR, NOTEBOOKS_DIR
_LOGGER: Logger = getLogger(__name__)
def run() -> None:
"""
Handles creation of application directories and user directories.
Uses `platformdirs.PlatformDirs` and `pathlib.Path` to create the required
app directories in the correct locations based on the users OS.
"""
app_dirs = [
_USER_DIRS,
NOTEBOOKS_DIR,
LOG_DIR,
]
for app_dir in app_dirs:
if not app_dir.is_dir():
app_dir.mkdir(parents=True, exist_ok=True)
_LOGGER.info(f"Created directory: {app_dir}")
if __name__ == "__main__":
run()