skinmap_runtime/core/utils/utils.py

import os

import numpy as np
import torch
import torch.distributed as dist
from loguru import logger


def fix_random_seeds(seed=42):
    """
    Fix random seeds.
    """
    torch.manual_seed(seed)
    torch.cuda.manual_seed_all(seed)
    np.random.seed(seed)


def flatten(t):
    return t.reshape(t.shape[0], -1)


def set_requires_grad(model, val):
    for p in model.parameters():
        p.requires_grad = val


def has_batchnorms(model: torch.nn.Module):
    bn_types = (
        torch.nn.BatchNorm1d,
        torch.nn.BatchNorm2d,
        torch.nn.BatchNorm3d,
        torch.nn.SyncBatchNorm,
    )
    for _, module in model.named_modules():
        if isinstance(module, bn_types):
            return True
    return False


def compare_models(model_1, model_2, log=False):
    models_differ = 0
    for key_item_1, key_item_2 in zip(
        model_1.state_dict().items(), model_2.state_dict().items()
    ):
        if torch.equal(key_item_1[1], key_item_2[1]):
            pass
        else:
            models_differ += 1
            if log and (key_item_1[0] == key_item_2[0]):
                logger.error("Mismatch found at", key_item_1[0])
    return models_differ


def clip_gradients(model, clip):
    norms = []
    for name, p in model.named_parameters():
        if p.grad is not None:
            param_norm = p.grad.data.norm(2)
            norms.append(param_norm.item())
            clip_coef = clip / (param_norm + 1e-6)
            if clip_coef < 1:
                p.grad.data.mul_(clip_coef)
    return norms


def init_distributed_mode():
    # launched with torch.distributed.launch
    if "RANK" in os.environ and "WORLD_SIZE" in os.environ:
        rank = int(os.environ["RANK"])
        world_size = int(os.environ["WORLD_SIZE"])
        gpu = int(os.environ["LOCAL_RANK"])
    # launched with submitit on a slurm cluster
    elif "SLURM_PROCID" in os.environ:
        rank = int(os.environ["SLURM_PROCID"])
        world_size = int(os.environ["WORLD_SIZE"])
        gpu = rank % torch.cuda.device_count()
    # launched naively with `python main_XXX.py`
    # we manually add MASTER_ADDR and MASTER_PORT to env variables
    elif torch.cuda.is_available():
        logger.debug("Will run the code on one GPU.")
        rank, gpu, world_size = 0, 0, 1
        os.environ["MASTER_ADDR"] = "127.0.0.1"
        os.environ["MASTER_PORT"] = "29500"
        os.environ["LOCAL_RANK"] = "0"
    else:
        # if there is no GPU available we don't do anything
        return

    dist.init_process_group(
        backend="nccl",
        init_method="env://",
        world_size=world_size,
        rank=rank,
    )

    torch.cuda.set_device(gpu)
    logger.debug(f"STARTUP: Distributed init (rank {rank}): env://", flush=True)
    dist.barrier()
    setup_for_distributed(rank == 0)


def cleanup():
    if is_dist_avail_and_initialized():
        dist.destroy_process_group()


def setup_for_distributed(is_master):
    """
    This function disables printing when not in master process
    """
    import builtins as __builtin__

    builtin_print = __builtin__.print

    def print(*args, **kwargs):
        force = kwargs.pop("force", False)
        if is_master or force:
            builtin_print(*args, **kwargs)

    __builtin__.print = print


def get_rank():
    if not is_dist_avail_and_initialized():
        return 0
    return dist.get_rank()


def is_main_process():
    return get_rank() == 0


def is_dist_avail_and_initialized():
    if not dist.is_available():
        return False
    if not dist.is_initialized():
        return False
    return True


def get_world_size():
    if not is_dist_avail_and_initialized():
        return 1
    return dist.get_world_size()


def restart_from_checkpoint(ckp_path, run_variables=None, **kwargs):
    if not os.path.isfile(ckp_path):
        logger.info("Pre-trained weights not found. Training from scratch.")
        return
    logger.info("Found checkpoint at {}".format(ckp_path))

    # open checkpoint file
    checkpoint = torch.load(ckp_path, map_location="cpu")

    # key is what to look for in the checkpoint file
    # value is the object to load
    # example: {'state_dict': model}
    for key, value in kwargs.items():
        if key in checkpoint and value is not None:
            try:
                msg = value.load_state_dict(checkpoint[key], strict=False)
                if len(msg.missing_keys) > 0:
                    k = next(iter(checkpoint[key]))
                    if "module." in k:
                        logger.debug(
                            f"=> Found `module` in {key}, trying to transform."
                        )
                        transf_state_dict = OrderedDict()
                        for k, v in checkpoint[key].items():
                            # remove the module from the key
                            # this is caused by the distributed training
                            k = k.replace("module.", "")
                            transf_state_dict[k] = v
                        msg = value.load_state_dict(transf_state_dict, strict=False)
                logger.debug(
                    "=> loaded '{}' from checkpoint '{}' with msg {}".format(
                        key, ckp_path, msg
                    )
                )
            except TypeError:
                try:
                    msg = value.load_state_dict(checkpoint[key])
                    logger.debug(
                        "=> loaded '{}' from checkpoint: '{}'".format(key, ckp_path)
                    )
                except ValueError:
                    logger.error(
                        "=> failed to load '{}' from checkpoint: '{}'".format(
                            key, ckp_path
                        )
                    )
        else:
            logger.error(
                "=> key '{}' not found in checkpoint: '{}'".format(key, ckp_path)
            )

    # reload variable important for the run
    if run_variables is not None:
        for var_name in run_variables:
            if var_name in checkpoint:
                run_variables[var_name] = checkpoint[var_name]


def save_checkpoint(run_dir, save_dict, epoch, save_best=False):
    if is_main_process():
        (run_dir / "checkpoints").mkdir(parents=True, exist_ok=True)
        filename = str(run_dir / "checkpoints" / "checkpoint-epoch{}.pth".format(epoch))
        torch.save(save_dict, filename)
        logger.info("Saving checkpoint: {} ...".format(filename))
        if save_best:
            best_path = str(run_dir / "checkpoints" / "model_best.pth")
            torch.save(save_dict, best_path)
            logger.info("Saving current best: model_best.pth ...")


class EarlyStopping:
    """
    Early stopping to stop the training when the loss does not improve after
    certain epochs.
    """

    def __init__(self, patience=5, min_delta=0, log_messages: bool = True):
        """
        :param patience: how many epochs to wait before stopping when loss is
               not improving
        :param min_delta: minimum difference between new loss and old loss for
               new loss to be considered as an improvement
        """
        self.patience = patience
        self.min_delta = min_delta
        self.log_messages = log_messages
        self.counter = 0
        self.best_loss = None
        self.early_stop = False

    def __call__(self, val_loss):
        if self.best_loss == None:
            self.best_loss = val_loss
        elif self.best_loss - val_loss > self.min_delta:
            self.best_loss = val_loss
            # reset counter if validation loss improves
            self.counter = 0
        elif self.best_loss - val_loss < self.min_delta:
            self.counter += 1
            if self.log_messages:
                logger.debug(
                    f"Early stopping counter {self.counter} of {self.patience}"
                )
            if self.counter >= self.patience:
                if self.log_messages:
                    logger.info("EarlyStopping, evaluation did not decrease.")
                self.early_stop = True


def p_value_stars(p: float, latex: bool = True) -> str:
    stars = "{}"
    if latex:
        stars = "{{{0}}}"
    if p < 0.001:
        return stars.format("***")
    elif p < 0.01:
        return stars.format("**")
    elif p < 0.05:
        return stars.format("*")
    else:
        return ""


def latex_median_quantile(arr: np.ndarray) -> str:
    median = np.median(arr)
    q_05 = np.quantile(arr, q=0.05)
    q_95 = np.quantile(arr, q=0.95)

    diff_05 = "{" + "{0:+.1f}".format(q_05 - median) + "}"
    diff_95 = "{" + "{0:+.1f}".format(q_95 - median) + "}"

    return f"{median:.1f}^{diff_95}_{diff_05}"