Source code for openghg.util._user

import logging
import os
import platform
from pathlib import Path
from typing import Dict
import uuid
import toml

logger = logging.getLogger("openghg.util")
logger.setLevel(logging.DEBUG)  # Have to set level for logger as well as handler


# @lru_cache
def get_user_id() -> str:
    """Return the user's ID

    Returns:
        str: User ID
    """
    config = read_local_config()
    uid = str(config.get("user_id", "NA"))
    return uid


def get_default_objectstore_path() -> Path:
    """Returns the default object store path in the user's home directory

    Returns:
        Path: Object store path in ~/openghg_store
    """
    return Path.home().joinpath("openghg_store").absolute()


# @lru_cache
def get_user_config_path() -> Path:
    """Checks if a config file has already been create for
    OpenGHG to use. This file is created in the user's home directory
    in  ~/.config/openghg/user.conf on Linux / macOS or
    in LOCALAPPDATA/openghg/openghg.conf on Windows.

    Returns:
        pathlib.Path: Path to user config file
    """
    user_platform = platform.system()

    openghg_config_filename = "openghg.conf"

    if user_platform == "Windows":
        appdata_path = os.getenv("LOCALAPPDATA")
        if appdata_path is None:
            raise ValueError("Unable to read LOCALAPPDATA environment variable.")

        config_path = Path(appdata_path).joinpath("openghg", openghg_config_filename)
    elif user_platform in ("Linux", "Darwin"):
        config_path = Path.home().joinpath(".config", "openghg", openghg_config_filename)
    else:
        raise ValueError(f"Unknown platform: {user_platform}")

    return config_path


def create_config(silent: bool = False) -> None:
    """Creates a user config.

    Returns:
        None
    """
    if not silent:
        print("\nOpenGHG configuration")
        print("---------------------\n")

    user_config_path = get_user_config_path()

    updated = False
    # If the config file exists we might need to update it due to the introduction
    # of the user ID
    if user_config_path.exists():
        if silent:
            logger.error("Error: cannot overwrite an existing configuration. Please run interactively.")
            return

        logger.info(f"User config exists at {str(user_config_path)}, checking...")

        config = toml.loads(user_config_path.read_text())

        objstore_path_config = Path(config["object_store"]["local_store"])

        logger.info(f"Current object store path: {objstore_path_config}")
        update_input = input("Would you like to update the path? (y/n): ")
        if update_input.lower() in ("y", "yes"):
            new_path_input = input("Enter new path for object store: ")
            new_path = Path(new_path_input).expanduser().resolve()

            config["object_store"]["local_store"] = str(new_path)
            updated = True
        else:
            logger.info("Matching object store path, nothing to do.\n")

        # Some users may not have a user ID if they've used previous versions of OpenGHG
        # Or if they UUID isn't valid a new one is created, the value of the UUID
        # doesn't matter at the moment, it's not used for anything very important
        try:
            user_id = config["user_id"]
            uuid.UUID(user_id, version=4)
        except (KeyError, ValueError):
            config["user_id"] = str(uuid.uuid4())
            updated = True
    else:
        updated = True
        default_objstore_path = get_default_objectstore_path()

        if silent:
            obj_store_path = default_objstore_path
        else:
            obj_store_input = input(f"Enter path for object store (default {default_objstore_path}): ")
            obj_store_path = Path(obj_store_input).expanduser().resolve()

        user_config_path.parent.mkdir(parents=True, exist_ok=True)

        user_id = str(uuid.uuid4())
        config = {"object_store": {"local_store": str(obj_store_path)}, "user_id": user_id}

        obj_store_path.mkdir(exist_ok=True)

        logger.info(f"Creating config at {str(user_config_path)}\n")

    if updated:
        logger.info(f"Configuration written to {user_config_path}")
        user_config_path.write_text(toml.dumps(config))
    else:
        logger.info("Configuration unchanged.")


# @lru_cache
def read_local_config() -> Dict:
    """Reads the local config file.

    Returns:
        dict: OpenGHG configurations
    """
    config_path = get_user_config_path()

    if not config_path.exists():
        raise FileNotFoundError(
            "Unable to read configuration file, please see the installation instructions or run openghg --quickstart"
        )

    config: Dict = toml.loads(config_path.read_text())
    return config


def check_config() -> None:
    """Check that the user config file is valid and the paths
    given in it exist.

    Returns:
        bool
    """
    config_path = get_user_config_path()

    if not config_path.exists():
        logger.warning("Configuration file does not exist. Please create it by running openghg --quickstart.")

    config = read_local_config()
    uid = config["user_id"]
    object_stores = config["object_store"]

    try:
        uuid.UUID(uid, version=4)
    except ValueError:
        logger.exception("Invalid user ID. Please re-run quickstart to setup a valid config file.")
        raise

    for path in object_stores.values():
        if not Path(path).exists():
            logger.info(f"{path} does not exist but will be created.")