Environment

`pop_arg(arg_name, default_value=None)`

Get the specified args and remove them from argv

Source code in V3_1/src/super_gradients/common/environment/argparse_utils.py

def pop_arg(arg_name: str, default_value: Any = None) -> Any:
    """Get the specified args and remove them from argv"""

    parser = argparse.ArgumentParser()
    parser.add_argument(f"--{arg_name}", default=default_value)
    args, _ = parser.parse_known_args()

    # Remove the ddp args to not have a conflict with the use of hydra
    for val in filter(lambda x: x.startswith(f"--{arg_name}"), sys.argv):
        EXTRA_ARGS.append(val)
        sys.argv.remove(val)
    return vars(args)[arg_name]

`pop_local_rank()`

Pop the python arg "local-rank". If exists inform the user with a log, otherwise return -1.

Source code in V3_1/src/super_gradients/common/environment/argparse_utils.py

def pop_local_rank() -> int:
    """Pop the python arg "local-rank". If exists inform the user with a log, otherwise return -1."""
    local_rank = pop_arg("local_rank", default_value=-1)
    if local_rank != -1:
        logger.info("local_rank was automatically parsed from your config.")
    return local_rank

`add_params_to_cfg(cfg, params)`

Add parameters to an existing config

Parameters:

Name	Type	Description	Default
`cfg`	`DictConfig`	OmegaConf config	required
`params`	`List[str]`	List of parameters to add, in dotlist format (i.e. ["training_hyperparams.resume=True"])	required

Source code in V3_1/src/super_gradients/common/environment/cfg_utils.py

def add_params_to_cfg(cfg: DictConfig, params: List[str]):
    """Add parameters to an existing config

    :param cfg:     OmegaConf config
    :param params:  List of parameters to add, in dotlist format (i.e. ["training_hyperparams.resume=True"])"""
    new_cfg = OmegaConf.from_dotlist(params)
    override_cfg(cfg, new_cfg)

`load_arch_params(config_name, recipes_dir_path=None, overrides=None)`

Load a single arch_params file.

Parameters:

Name	Type	Description	Default
`config_name`	`str`	Name of the yaml to load (e.g. "resnet18_cifar_arch_params")	required
`recipes_dir_path`	`Optional[str]`	Optional. Main directory where every recipe are stored. (e.g. ../super_gradients/recipes) This directory should include a "arch_params" folder, which itself should include the config file named after config_name.	`None`
`overrides`	`Optional[list]`	List of hydra overrides for config file	`None`

Source code in V3_1/src/super_gradients/common/environment/cfg_utils.py

def load_arch_params(config_name: str, recipes_dir_path: Optional[str] = None, overrides: Optional[list] = None) -> DictConfig:
    """Load a single arch_params file.
    :param config_name:         Name of the yaml to load (e.g. "resnet18_cifar_arch_params")
    :param recipes_dir_path:    Optional. Main directory where every recipe are stored. (e.g. ../super_gradients/recipes)
                                This directory should include a "arch_params" folder,
                                which itself should include the config file named after config_name.
    :param overrides:           List of hydra overrides for config file
    """
    return load_recipe_from_subconfig(config_name=config_name, recipes_dir_path=recipes_dir_path, overrides=overrides, config_type="arch_params")

`load_dataset_params(config_name, recipes_dir_path=None, overrides=None)`

Load a single dataset_params file.

Parameters:

Name	Type	Description	Default
`config_name`	`str`	Name of the yaml to load (e.g. "cifar10_dataset_params")	required
`recipes_dir_path`	`Optional[str]`	Optional. Main directory where every recipe are stored. (e.g. ../super_gradients/recipes) This directory should include a "training_hyperparams" folder, which itself should include the config file named after config_name.	`None`
`overrides`	`Optional[list]`	List of hydra overrides for config file	`None`

Source code in V3_1/src/super_gradients/common/environment/cfg_utils.py

def load_dataset_params(config_name: str, recipes_dir_path: Optional[str] = None, overrides: Optional[list] = None) -> DictConfig:
    """Load a single dataset_params file.
    :param config_name:         Name of the yaml to load (e.g. "cifar10_dataset_params")
    :param recipes_dir_path:    Optional. Main directory where every recipe are stored. (e.g. ../super_gradients/recipes)
                                This directory should include a "training_hyperparams" folder,
                                which itself should include the config file named after config_name.
    :param overrides:           List of hydra overrides for config file
    """
    return load_recipe_from_subconfig(config_name=config_name, recipes_dir_path=recipes_dir_path, overrides=overrides, config_type="dataset_params")

`load_experiment_cfg(experiment_name, ckpt_root_dir=None)`

Load the hydra config associated to a specific experiment.

Background Information: every time an experiment is launched based on a recipe, all the hydra config params are stored in a hidden folder ".hydra". This hidden folder is used here to recreate the exact same config as the one that was used to launch the experiment (Also include hydra overrides).

The motivation is to be able to resume or evaluate an experiment with the exact same config as the one that was used when the experiment was initially started, regardless of any change that might have been introduced to the recipe, and also while using the same overrides that were used for that experiment.

Parameters:

Name	Type	Description	Default
`experiment_name`	`str`	Name of the experiment to resume	required
`ckpt_root_dir`	`str`	Directory including the checkpoints	`None`

Returns:

Type	Description
`DictConfig`	The config that was used for that experiment

Source code in V3_1/src/super_gradients/common/environment/cfg_utils.py

def load_experiment_cfg(experiment_name: str, ckpt_root_dir: str = None) -> DictConfig:
    """
    Load the hydra config associated to a specific experiment.

    Background Information: every time an experiment is launched based on a recipe, all the hydra config params are stored in a hidden folder ".hydra".
    This hidden folder is used here to recreate the exact same config as the one that was used to launch the experiment (Also include hydra overrides).

    The motivation is to be able to resume or evaluate an experiment with the exact same config as the one that was used when the experiment was
    initially started, regardless of any change that might have been introduced to the recipe, and also while using the same overrides that were used
    for that experiment.

    :param experiment_name:     Name of the experiment to resume
    :param ckpt_root_dir:       Directory including the checkpoints
    :return:                    The config that was used for that experiment
    """
    if not experiment_name:
        raise ValueError(f"experiment_name should be non empty string but got :{experiment_name}")

    checkpoints_dir_path = Path(get_checkpoints_dir_path(experiment_name, ckpt_root_dir))
    if not checkpoints_dir_path.exists():
        raise FileNotFoundError(f"Impossible to find checkpoint dir ({checkpoints_dir_path})")

    resume_dir = Path(checkpoints_dir_path) / ".hydra"
    if not resume_dir.exists():
        raise FileNotFoundError(f"The checkpoint directory {checkpoints_dir_path} does not include .hydra artifacts to resume the experiment.")

    # Load overrides that were used in previous run
    overrides_cfg = list(OmegaConf.load(resume_dir / "overrides.yaml"))

    cfg = load_recipe(config_name="config.yaml", recipes_dir_path=normalize_path(str(resume_dir)), overrides=overrides_cfg)
    return cfg

`load_recipe(config_name, recipes_dir_path=None, overrides=None)`

Load a single a file of the recipe directory.

Parameters:

Name	Type	Description	Default
`config_name`	`str`	Name of the yaml to load (e.g. "cifar10_resnet")	required
`recipes_dir_path`	`Optional[str]`	Optional. Main directory where every recipe are stored. (e.g. ../super_gradients/recipes) This directory should include a folder corresponding to the subconfig, which itself should include the config file named after config_name.	`None`
`overrides`	`Optional[list]`	List of hydra overrides for config file	`None`

Source code in V3_1/src/super_gradients/common/environment/cfg_utils.py

def load_recipe(config_name: str, recipes_dir_path: Optional[str] = None, overrides: Optional[list] = None) -> DictConfig:
    """Load a single a file of the recipe directory.

    :param config_name:         Name of the yaml to load (e.g. "cifar10_resnet")
    :param recipes_dir_path:    Optional. Main directory where every recipe are stored. (e.g. ../super_gradients/recipes)
                                This directory should include a folder corresponding to the subconfig, which itself should
                                include the config file named after config_name.
    :param overrides:           List of hydra overrides for config file
    """
    GlobalHydra.instance().clear()

    config_dir = recipes_dir_path or pkg_resources.resource_filename("super_gradients.recipes", "")

    with initialize_config_dir(config_dir=normalize_path(config_dir), version_base="1.2"):
        try:
            cfg = compose(config_name=normalize_path(config_name), overrides=overrides if overrides else [])
        except hydra.errors.MissingConfigException:
            raise RecipeNotFoundError(config_name=config_name, config_dir=config_dir, recipes_dir_path=recipes_dir_path)
    return cfg

`load_recipe_from_subconfig(config_name, config_type, recipes_dir_path=None, overrides=None)`

Load a single a file (e.g. "resnet18_cifar_arch_params") stored in a subconfig (e.g. "arch_param") of the recipe directory,.

Parameters:

Name	Type	Description	Default
`config_name`	`str`	Name of the yaml to load (e.g. "resnet18_cifar_arch_params")	required
`config_type`	`str`	Type of the subconfig (e.g. "arch_params")	required
`recipes_dir_path`	`Optional[str]`	Optional. Main directory where every recipe are stored. (e.g. ../super_gradients/recipes) This directory should include a folder corresponding to the subconfig, which itself should include the config file named after config_name.	`None`
`overrides`	`Optional[list]`	List of hydra overrides for config file	`None`

Source code in V3_1/src/super_gradients/common/environment/cfg_utils.py

def load_recipe_from_subconfig(config_name: str, config_type: str, recipes_dir_path: Optional[str] = None, overrides: Optional[list] = None) -> DictConfig:
    """Load a single a file (e.g. "resnet18_cifar_arch_params") stored in a subconfig (e.g. "arch_param") of the recipe directory,.

    :param config_name:         Name of the yaml to load (e.g. "resnet18_cifar_arch_params")
    :param config_type:         Type of the subconfig (e.g. "arch_params")
    :param recipes_dir_path:    Optional. Main directory where every recipe are stored. (e.g. ../super_gradients/recipes)
                                This directory should include a folder corresponding to the subconfig,
                                which itself should include the config file named after config_name.
    :param overrides:           List of hydra overrides for config file
    """

    try:
        cfg = load_recipe(config_name=os.path.join(config_type, config_name), recipes_dir_path=recipes_dir_path, overrides=overrides)
    except RecipeNotFoundError as e:
        postfix_err_msg = (
            f"Note: If your recipe is saved at '{os.path.join(e.config_dir, config_name.replace('.yaml', ''))}.yaml', you can load it with load_recipe(...).\n"
        )

        raise RecipeNotFoundError(
            config_name=config_name,
            config_dir=e.config_dir,
            config_type=config_type,
            recipes_dir_path=recipes_dir_path,
            postfix_err_msg=postfix_err_msg,
        )

    # Because of the way we load the subconfig, cfg will start with a single key corresponding to the type (arch_params, ...) and don't want that.
    cfg = cfg[config_type]

    return cfg

`load_training_hyperparams(config_name, recipes_dir_path=None, overrides=None)`

Load a single training_hyperparams file.

Parameters:

Name	Type	Description	Default
`config_name`	`str`	Name of the yaml to load (e.g. "cifar10_resnet_train_params")	required
`recipes_dir_path`	`Optional[str]`	Optional. Main directory where every recipe are stored. (e.g. ../super_gradients/recipes) This directory should include a "training_hyperparams" folder, which itself should include the config file named after config_name.	`None`
`overrides`	`Optional[list]`	List of hydra overrides for config file	`None`

Source code in V3_1/src/super_gradients/common/environment/cfg_utils.py

def load_training_hyperparams(config_name: str, recipes_dir_path: Optional[str] = None, overrides: Optional[list] = None) -> DictConfig:
    """Load a single training_hyperparams file.
    :param config_name:         Name of the yaml to load (e.g. "cifar10_resnet_train_params")
    :param recipes_dir_path:    Optional. Main directory where every recipe are stored. (e.g. ../super_gradients/recipes)
                                This directory should include a "training_hyperparams" folder,
                                which itself should include the config file named after config_name.
    :param overrides:           List of hydra overrides for config file
    """
    return load_recipe_from_subconfig(config_name=config_name, recipes_dir_path=recipes_dir_path, overrides=overrides, config_type="training_hyperparams")

`override_cfg(cfg, overrides)`

Override inplace a config with a list of hydra overrides

Parameters:

Name	Type	Description	Default
`cfg`	`DictConfig`	OmegaConf config	required
`overrides`	`Union[DictConfig, Dict[str, Any]]`	Dictionary like object that will be used to override cfg	required

Source code in V3_1/src/super_gradients/common/environment/cfg_utils.py

def override_cfg(cfg: DictConfig, overrides: Union[DictConfig, Dict[str, Any]]) -> None:
    """Override inplace a config with a list of hydra overrides
    :param cfg:         OmegaConf config
    :param overrides:   Dictionary like object that will be used to override cfg
    """
    with open_dict(cfg):  # This is required to add new fields to existing config
        cfg.merge_with(overrides)

`get_checkpoints_dir_path(experiment_name, ckpt_root_dir=None)`

Get the directory that includes all the checkpoints (and logs) of an experiment.

Parameters:

Name	Type	Description	Default
`experiment_name`	`str`	Name of the experiment.	required
`ckpt_root_dir`	`str`	Path to the directory where all the experiments are organised, each sub-folder representing a specific experiment. If None, SG will first check if a package named 'checkpoints' exists. If not, SG will look for the root of the project that includes the script that was launched. If not found, raise an error.	`None`

Returns:

Type	Description
`str`	Path of folder where the experiment checkpoints and logs will be stored.

Source code in V3_1/src/super_gradients/common/environment/checkpoints_dir_utils.py

def get_checkpoints_dir_path(experiment_name: str, ckpt_root_dir: str = None) -> str:
    """Get the directory that includes all the checkpoints (and logs) of an experiment.

    :param experiment_name:     Name of the experiment.
    :param ckpt_root_dir:       Path to the directory where all the experiments are organised, each sub-folder representing a specific experiment.
                                    If None, SG will first check if a package named 'checkpoints' exists.
                                    If not, SG will look for the root of the project that includes the script that was launched.
                                    If not found, raise an error.
    :return:                    Path of folder where the experiment checkpoints and logs will be stored.
    """
    ckpt_root_dir = ckpt_root_dir or PKG_CHECKPOINTS_DIR or get_project_checkpoints_dir_path()
    if ckpt_root_dir is None:
        raise ValueError("Illegal checkpoints directory: please set ckpt_root_dir")
    return os.path.join(ckpt_root_dir, experiment_name)

`get_ckpt_local_path(experiment_name, ckpt_name, external_checkpoint_path, ckpt_root_dir=None)`

Gets the local path to the checkpoint file, which will be: - By default: YOUR_REPO_ROOT/super_gradients/checkpoints/experiment_name/ckpt_name. - external_checkpoint_path when external_checkpoint_path != None - ckpt_root_dir/experiment_name/ckpt_name when ckpt_root_dir != None. - if the checkpoint file is remotely located: when overwrite_local_checkpoint=True then it will be saved in a temporary path which will be returned, otherwise it will be downloaded to YOUR_REPO_ROOT/super_gradients/checkpoints/experiment_name and overwrite YOUR_REPO_ROOT/super_gradients/checkpoints/experiment_name/ckpt_name if such file exists.

Parameters:

Name	Type	Description	Default
`experiment_name`	`str`	experiment name attr in trainer :param ckpt_name: checkpoint filename	required
`external_checkpoint_path`	`str`	full path to checkpoint file (that might be located outside of super_gradients/checkpoints directory)	required
`ckpt_root_dir`	`str`	Local root directory path where all experiment logging directories will reside. When None, it is assumed that pkg_resources.resource_filename( 'checkpoints', "") exists and will be used. :return: local path of the checkpoint file (Str)	`None`

Source code in V3_1/src/super_gradients/common/environment/checkpoints_dir_utils.py

def get_ckpt_local_path(experiment_name: str, ckpt_name: str, external_checkpoint_path: str, ckpt_root_dir: str = None) -> str:
    """
    Gets the local path to the checkpoint file, which will be:
        - By default: YOUR_REPO_ROOT/super_gradients/checkpoints/experiment_name/ckpt_name.
        - external_checkpoint_path when external_checkpoint_path != None
        - ckpt_root_dir/experiment_name/ckpt_name when ckpt_root_dir != None.
        - if the checkpoint file is remotely located:
            when overwrite_local_checkpoint=True then it will be saved in a temporary path which will be returned,
            otherwise it will be downloaded to YOUR_REPO_ROOT/super_gradients/checkpoints/experiment_name and overwrite
            YOUR_REPO_ROOT/super_gradients/checkpoints/experiment_name/ckpt_name if such file exists.


    :param experiment_name: experiment name attr in trainer :param ckpt_name: checkpoint filename
    :param external_checkpoint_path: full path to checkpoint file (that might be located outside of
    super_gradients/checkpoints directory)
    :param ckpt_root_dir: Local root directory path where all experiment
     logging directories will reside. When None, it is assumed that pkg_resources.resource_filename(
    'checkpoints', "") exists and will be used.

     :return: local path of the checkpoint file (Str)
    """
    if external_checkpoint_path:
        return external_checkpoint_path
    else:
        checkpoints_dir_path = get_checkpoints_dir_path(experiment_name, ckpt_root_dir)
        return os.path.join(checkpoints_dir_path, ckpt_name)

`get_project_checkpoints_dir_path()`

Get the checkpoints' directory that is at the root of the users project. Create it if it doesn't exist. Return None if root not found.

Source code in V3_1/src/super_gradients/common/environment/checkpoints_dir_utils.py

def get_project_checkpoints_dir_path() -> Optional[str]:
    """Get the checkpoints' directory that is at the root of the users project. Create it if it doesn't exist. Return None if root not found."""
    project_root_path = _get_project_root_path()
    if project_root_path is None:
        return None

    checkpoints_path = os.path.join(project_root_path, "checkpoints")
    if not os.path.exists(checkpoints_path):
        os.makedirs(checkpoints_path)
        logger.info(f'A checkpoints directory was just created at "{checkpoints_path}". To work with another directory, please set "ckpt_root_dir"')
    return checkpoints_path

`find_free_port()`

Find an available port of current machine/node. Note: there is still a chance the port could be taken by other processes.

Source code in V3_1/src/super_gradients/common/environment/ddp_utils.py

def find_free_port() -> int:
    """Find an available port of current machine/node.
    Note: there is still a chance the port could be taken by other processes."""

    with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as sock:
        # Binding to port 0 will cause the OS to find an available port for us
        sock.bind(("", 0))
        _ip, port = sock.getsockname()
    return port

`init_trainer()`

Initialize the super_gradients environment.

This function should be the first thing to be called by any code running super_gradients.

Source code in V3_1/src/super_gradients/common/environment/ddp_utils.py

def init_trainer():
    """
    Initialize the super_gradients environment.

    This function should be the first thing to be called by any code running super_gradients.
    """
    register_hydra_resolvers()
    pop_local_rank()

`is_distributed()`

Check if current process is a DDP subprocess.

Source code in V3_1/src/super_gradients/common/environment/ddp_utils.py

def is_distributed() -> bool:
    """Check if current process is a DDP subprocess."""
    return device_config.assigned_rank >= 0

`is_launched_using_sg()`

Check if the current process is a subprocess launched using SG restart_script_with_ddp

Source code in V3_1/src/super_gradients/common/environment/ddp_utils.py

def is_launched_using_sg():
    """Check if the current process is a subprocess launched using SG restart_script_with_ddp"""
    return os.environ.get("TORCHELASTIC_RUN_ID") == "sg_initiated"

`is_main_process()`

Check if current process is considered as the main process (i.e. is responsible for sanity check, atexit upload, ...). The definition ensures that 1 and only 1 process follows this condition, regardless of how the run was started.

The rule is as follow: - If not DDP: main process is current process - If DDP launched using SuperGradients: main process is the launching process (rank=-1) - If DDP launched with torch: main process is rank 0

Source code in V3_1/src/super_gradients/common/environment/ddp_utils.py

def is_main_process():
    """Check if current process is considered as the main process (i.e. is responsible for sanity check, atexit upload, ...).
    The definition ensures that 1 and only 1 process follows this condition, regardless of how the run was started.

    The rule is as follow:
        - If not DDP: main process is current process
        - If DDP launched using SuperGradients: main process is the launching process (rank=-1)
        - If DDP launched with torch: main process is rank 0
    """

    if not is_distributed():  # If no DDP, or DDP launching process
        return True
    elif (
        device_config.assigned_rank == 0 and not is_launched_using_sg()
    ):  # If DDP launched using torch.distributed.launch or torchrun, we need to run the check on rank 0
        return True
    else:
        return False

`multi_process_safe(func)`

A decorator for making sure a function runs only in main process. If not in DDP mode (local_rank = -1), the function will run. If in DDP mode, the function will run only in the main process (local_rank = 0) This works only for functions with no return value

Source code in V3_1/src/super_gradients/common/environment/ddp_utils.py

def multi_process_safe(func):
    """
    A decorator for making sure a function runs only in main process.
    If not in DDP mode (local_rank = -1), the function will run.
    If in DDP mode, the function will run only in the main process (local_rank = 0)
    This works only for functions with no return value
    """

    def do_nothing(*args, **kwargs):
        pass

    @wraps(func)
    def wrapper(*args, **kwargs):
        if device_config.assigned_rank <= 0:
            return func(*args, **kwargs)
        else:
            return do_nothing(*args, **kwargs)

    return wrapper

`EnvironmentVariables`

Class to dynamically get any environment variables.

Source code in V3_1/src/super_gradients/common/environment/env_variables.py

class EnvironmentVariables:
    """Class to dynamically get any environment variables."""

    # Infra

    @property
    def WANDB_BASE_URL(self) -> str:
        return os.getenv("WANDB_BASE_URL")

    @property
    def AWS_PROFILE(self) -> str:
        return os.getenv("AWS_PROFILE")

    # DDP
    @property
    def LOCAL_RANK(self) -> int:
        return int(os.getenv("LOCAL_RANK", -1))

    # Turn ON/OFF features
    @property
    def CRASH_HANDLER(self) -> str:
        return os.getenv("CRASH_HANDLER", "TRUE")

    @property
    def UPLOAD_LOGS(self) -> bool:
        return os.getenv("UPLOAD_LOGS", "TRUE") == "TRUE"

    @property
    def FILE_LOG_LEVEL(self) -> str:
        return os.getenv("FILE_LOG_LEVEL", default="DEBUG").upper()

    @property
    def CONSOLE_LOG_LEVEL(self) -> str:
        return os.getenv("CONSOLE_LOG_LEVEL", default="INFO").upper()

`get_cpu_percent()`

Average of all the CPU utilization.

Source code in V3_1/src/super_gradients/common/environment/monitoring/cpu.py

def get_cpu_percent() -> float:
    """Average of all the CPU utilization."""
    return psutil.cpu_percent(interval=None, percpu=False)

`GPUStatAggregatorIterator` `dataclass`

Iterator of multiple StatAggregator, that accumulate samples and aggregates them for each NVIDIA device.

Parameters:

Name	Type	Description	Default
`name`	`str`	Name of the statistic	required
`sampling_fn`		How the statistic is sampled	required
`aggregate_fn`		How the statistic samples are aggregated	required

Source code in V3_1/src/super_gradients/common/environment/monitoring/data_models.py

@dataclasses.dataclass
class GPUStatAggregatorIterator:
    """Iterator of multiple StatAggregator, that accumulate samples and aggregates them for each NVIDIA device.

    :param name:            Name of the statistic
    :param sampling_fn:     How the statistic is sampled
    :param aggregate_fn:    How the statistic samples are aggregated
    """

    name: str
    device_sampling_fn: Callable
    device_aggregate_fn: Callable
    _per_device_stat_aggregator: List[StatAggregator] = dataclasses.field(init=False)

    def __post_init__(self):
        """Initialize nvidia_management_lib and create a list of StatAggregator, one for each NVIDIA device."""
        init_nvidia_management_lib()
        self._per_device_stat_aggregator = [
            StatAggregator(name=f"{self.name}/device_{i}", sampling_fn=partial(self.device_sampling_fn, i), aggregate_fn=self.device_aggregate_fn)
            for i in range(count_gpus())
        ]

    def __iter__(self) -> Iterator[StatAggregator]:
        """Iterate over the StatAggregator of each node"""
        return iter(self._per_device_stat_aggregator)

`iter()`

Iterate over the StatAggregator of each node

Source code in V3_1/src/super_gradients/common/environment/monitoring/data_models.py

def __iter__(self) -> Iterator[StatAggregator]:
    """Iterate over the StatAggregator of each node"""
    return iter(self._per_device_stat_aggregator)

`__post_init__()`

Initialize nvidia_management_lib and create a list of StatAggregator, one for each NVIDIA device.

Source code in V3_1/src/super_gradients/common/environment/monitoring/data_models.py

def __post_init__(self):
    """Initialize nvidia_management_lib and create a list of StatAggregator, one for each NVIDIA device."""
    init_nvidia_management_lib()
    self._per_device_stat_aggregator = [
        StatAggregator(name=f"{self.name}/device_{i}", sampling_fn=partial(self.device_sampling_fn, i), aggregate_fn=self.device_aggregate_fn)
        for i in range(count_gpus())
    ]

`StatAggregator` `dataclass`

Accumulate statistics samples and aggregates them.

Parameters:

Name	Type	Description	Default
`name`	`str`	Name of the statistic	required
`sampling_fn`	`Callable`	How the statistic is sampled	required
`aggregate_fn`	`Callable[[List[Any], float], float]`	How the statistic samples are aggregated, has to take "samples: List[Any]" and "time: float" as parameters	required
`reset_callback_fn`	`Optional[Callable]`	Optional, can be used to reset any system metric	`None`

Source code in V3_1/src/super_gradients/common/environment/monitoring/data_models.py

@dataclasses.dataclass
class StatAggregator:
    """Accumulate statistics samples and aggregates them.

    :param name:                Name of the statistic
    :param sampling_fn:         How the statistic is sampled
    :param aggregate_fn:        How the statistic samples are aggregated, has to take "samples: List[Any]" and "time: float" as parameters
    :param reset_callback_fn:   Optional, can be used to reset any system metric
    """

    name: str
    sampling_fn: Callable
    aggregate_fn: Callable[[List[Any], float], float]
    reset_callback_fn: Optional[Callable] = None
    _samples: List = dataclasses.field(default_factory=list)
    _reset_time: float = None

    def sample(self):
        try:
            self._samples.append(self.sampling_fn())
        except Exception:
            pass

    def aggregate(self) -> Union[float, None]:
        if len(self._samples) > 0:
            time_diff = time.time() - self._reset_time
            return self.aggregate_fn(self._samples, time_diff)

    def reset(self):
        self._samples = []
        self._reset_time = time.time()
        if self.reset_callback_fn:
            self.reset_callback_fn()

`get_disk_usage_percent()`

Disk memory used in percent.

Source code in V3_1/src/super_gradients/common/environment/monitoring/disk.py

def get_disk_usage_percent() -> float:
    """Disk memory used in percent."""
    return psutil.disk_usage("/").percent

`get_io_read_mb()`

Number of MegaBytes read since import

Source code in V3_1/src/super_gradients/common/environment/monitoring/disk.py

def get_io_read_mb() -> float:
    """Number of MegaBytes read since import"""
    return bytes_to_megabytes(psutil.disk_io_counters().read_bytes - buffer_io_read_bytes)

`get_io_write_mb()`

Number of MegaBytes written since import

Source code in V3_1/src/super_gradients/common/environment/monitoring/disk.py

def get_io_write_mb() -> float:
    """Number of MegaBytes written since import"""
    return bytes_to_megabytes(psutil.disk_io_counters().write_bytes - buffer_io_write_bytes)

`reset_io_read()`

Reset the value of net_io_counters

Source code in V3_1/src/super_gradients/common/environment/monitoring/disk.py

def reset_io_read():
    """Reset the value of net_io_counters"""
    global buffer_io_read_bytes
    buffer_io_read_bytes = psutil.disk_io_counters().read_bytes

`reset_io_write()`

Reset the value of net_io_counters

Source code in V3_1/src/super_gradients/common/environment/monitoring/disk.py

def reset_io_write():
    """Reset the value of net_io_counters"""
    global buffer_io_write_bytes
    buffer_io_write_bytes = psutil.disk_io_counters().write_bytes