Source code for catalyst.utils.sys

from typing import Any, Dict, List, Union
import json
import os
from pathlib import Path
import platform
import shutil
import subprocess
from subprocess import CalledProcessError
import sys
import warnings

from import SummaryWriter
from catalyst.settings import IS_HYDRA_AVAILABLE
from catalyst.utils.config import save_config
from catalyst.utils.misc import get_utcnow_time

    from omegaconf import DictConfig, OmegaConf

def _decode_dict(dictionary: Dict[str, Union[bytes, str]]) -> Dict[str, str]:
    Decode bytes values in the dictionary to UTF-8.

        dictionary: a dict

        Dict: decoded dict
    result = {
        k: v.decode("UTF-8") if type(v) == bytes else v
        for k, v in dictionary.items()
    return result

[docs]def get_environment_vars() -> Dict[str, Any]: """ Creates a dictionary with environment variables. Returns: Dict: environment variables """ result = { "python_version": sys.version, "conda_environment": os.environ.get("CONDA_DEFAULT_ENV", ""), "creation_time": get_utcnow_time(), "sysname": platform.uname()[0], "nodename": platform.uname()[1], "release": platform.uname()[2], "version": platform.uname()[3], "architecture": platform.uname()[4], "user": os.environ.get("USER", ""), "path": os.environ.get("PWD", ""), } with open(os.devnull, "w") as devnull: try: git_branch = ( subprocess.check_output( "git rev-parse --abbrev-ref HEAD".split(), shell=True, stderr=devnull, ) .strip() .decode("UTF-8") ) git_local_commit = subprocess.check_output( "git rev-parse HEAD".split(), shell=True, stderr=devnull ) git_origin_commit = subprocess.check_output( f"git rev-parse origin/{git_branch}".split(), shell=True, stderr=devnull, ) git = { "branch": git_branch, "local_commit": git_local_commit, "origin_commit": git_origin_commit, } result["git"] = _decode_dict(git) except (CalledProcessError, FileNotFoundError): pass result = _decode_dict(result) return result
[docs]def list_pip_packages() -> str: """ Lists pip installed packages. Returns: str: string with pip installed packages """ result = "" # TODO: Docs. Contribution is welcome # TODO: When catching exception, e has no attribute 'output' with open(os.devnull, "w") as devnull: try: result = ( subprocess.check_output("pip freeze".split(), stderr=devnull) .strip() .decode("UTF-8") ) except Exception: warnings.warn( "Failed to freeze pip packages. " # f"Pip Output: ```{e.output}```." "Continue experiment without pip packages dumping." ) pass # except FileNotFoundError: # pass # except subprocess.CalledProcessError as e: # raise Exception("Failed to list packages") from e return result
[docs]def list_conda_packages() -> str: """ Lists conda installed packages. Returns: str: list with conda installed packages """ result = "" conda_meta_path = Path(sys.prefix) / "conda-meta" # TODO: Docs. Contribution is welcome # TODO: When catching exception, e has no attribute 'output' if conda_meta_path.exists(): # We are currently in conda virtual env with open(os.devnull, "w") as devnull: try: result = ( subprocess.check_output( "conda list --export".split(), stderr=devnull ) .strip() .decode("UTF-8") ) except Exception: warnings.warn( "Running from conda env, " "but failed to list conda packages. " # f"Conda Output: ```{e.output}```." "Continue experiment without conda packages dumping." ) pass # except FileNotFoundError: # pass # except subprocess.CalledProcessError as e: # raise Exception( # f"Running from conda env, " # f"but failed to list conda packages. " # f"Conda Output: {e.output}" # ) from e return result
[docs]def dump_environment( experiment_config: Any, logdir: str, configs_path: List[str] = None, ) -> None: """ Saves config, environment variables and package list in JSON into logdir. Args: experiment_config: experiment config logdir: path to logdir configs_path: path(s) to config """ configs_path = configs_path or [] configs_path = [ Path(path) for path in configs_path if isinstance(path, str) ] config_dir = Path(logdir) / "configs" config_dir.mkdir(exist_ok=True, parents=True) if IS_HYDRA_AVAILABLE and isinstance(experiment_config, DictConfig): with open(config_dir / "config.yaml", "w") as f: f.write(OmegaConf.to_yaml(experiment_config, resolve=True)) experiment_config = OmegaConf.to_container( experiment_config, resolve=True ) environment = get_environment_vars() save_config(experiment_config, config_dir / "_config.json") save_config(environment, config_dir / "_environment.json") pip_pkg = list_pip_packages() (config_dir / "pip-packages.txt").write_text(pip_pkg) conda_pkg = list_conda_packages() if conda_pkg: (config_dir / "conda-packages.txt").write_text(conda_pkg) for path in configs_path: name: str = outpath = config_dir / name shutil.copyfile(path, outpath) config_str = json.dumps(experiment_config, indent=2, ensure_ascii=False) config_str = config_str.replace("\n", "\n\n") environment_str = json.dumps(environment, indent=2, ensure_ascii=False) environment_str = environment_str.replace("\n", "\n\n") pip_pkg = pip_pkg.replace("\n", "\n\n") conda_pkg = conda_pkg.replace("\n", "\n\n") with SummaryWriter(config_dir) as writer: writer.add_text("_config", config_str, 0) writer.add_text("_environment", environment_str, 0) writer.add_text("pip-packages", pip_pkg, 0) if conda_pkg: writer.add_text("conda-packages", conda_pkg, 0)
__all__ = [ "get_environment_vars", "list_conda_packages", "list_pip_packages", "dump_environment", ]