Repository: Lightning-AI/pytorch-lightning
File: pytorch_lightning/trainer/connectors/checkpoint_connector.py (Python)

Code before:
import pytorch_lightning
from pytorch_lightning.core.lightning import LightningModule
from pytorch_lightning.utilities import _OMEGACONF_AVAILABLE, DeviceType, rank_zero_info, rank_zero_warn
from pytorch_lightning.utilities.cloud_io import atomic_save, get_filesystem
from pytorch_lightning.utilities.cloud_io import load as pl_load
from pytorch_lightning.utilities.exceptions import MisconfigurationException
from pytorch_lightning.utilities.upgrade_checkpoint import KEYS_MAPPING as DEPRECATED_CHECKPOINT_KEYS

if _OMEGACONF_AVAILABLE:
    from omegaconf import Container


class CheckpointConnector:

    def __init__(self, trainer, resume_from_checkpoint: Optional[Union[str, Path]] = None):
        self.trainer = trainer
        self.resume_checkpoint_path = resume_from_checkpoint
        self.loaded_checkpoint = dict()
        # used to validate checkpointing logic
        self.has_trained = False
        self._load_optimizer_states = True

    @property
    def hpc_resume_path(self) -> Optional[str]:
        dir_path_hpc = str(self.trainer.weights_save_path)
        max_suffix = self.max_ckpt_in_folder(dir_path_hpc, "hpc_ckpt_")
        if max_suffix is not None:
            return f"{dir_path_hpc}/hpc_ckpt_{max_suffix}.ckpt"

    def resume_start(self) -> None:
        """
        Attempt to restore a checkpoint in this priority:

        1. from HPC weights if found
        2. from `resume_from_checkpoint` file if provided
        3. don't restore
        """
        self.resume_checkpoint_path = self.hpc_resume_path or self.resume_checkpoint_path
        checkpoint_path = self.resume_checkpoint_path
        if not checkpoint_path:
            return

        # clear cache before restore
        if self.trainer._device_type == DeviceType.GPU:
            torch.cuda.empty_cache()

        # Try to read the checkpoint file at `checkpoint_path`. If not exist, do not restore checkpoint.
        fs = get_filesystem(checkpoint_path)
        if not fs.exists(checkpoint_path):
            raise FileNotFoundError(f"Checkpoint at {checkpoint_path} not found. Aborting training.")


Review comment (suggestion):
```suggestion
        max_suffix = self.{latest|best}_ckpt_in_folder(dir_path_hpc, "hpc_ckpt_")
```
the max has confusing meaning

Code after:
import pytorch_lightning
from pytorch_lightning.core.lightning import LightningModule
from pytorch_lightning.utilities import _OMEGACONF_AVAILABLE, DeviceType, rank_zero_info, rank_zero_warn
from pytorch_lightning.utilities.cloud_io import atomic_save, get_filesystem
from pytorch_lightning.utilities.cloud_io import load as pl_load
from pytorch_lightning.utilities.exceptions import MisconfigurationException
from pytorch_lightning.utilities.upgrade_checkpoint import KEYS_MAPPING as DEPRECATED_CHECKPOINT_KEYS

if _OMEGACONF_AVAILABLE:
    from omegaconf import Container


class CheckpointConnector:

    def __init__(self, trainer, resume_from_checkpoint: Optional[Union[str, Path]] = None):
        self.trainer = trainer
        self.resume_checkpoint_path = resume_from_checkpoint
        # used to validate checkpointing logic
        self.has_trained = False

        self._loaded_checkpoint = dict()
        # FIXME: remove in https://github.com/PyTorchLightning/pytorch-lightning/pull/7652
        self._load_optimizer_states = True

    @property
    def hpc_resume_path(self) -> Optional[str]:
        dir_path_hpc = str(self.trainer.weights_save_path)
        max_version = self.max_ckpt_version_in_folder(dir_path_hpc, "hpc_ckpt_")
        if max_version is not None:
            return f"{dir_path_hpc}/hpc_ckpt_{max_version}.ckpt"

    def resume_start(self) -> None:
        """
        Attempts to pre-load the checkpoint file to memory, with the source path determined in this priority:

        1. from HPC weights if found
        2. from `resume_from_checkpoint` file if provided
        3. don't restore

        Raises:
            FileNotFoundError: If the path to the checkpoint file is provided but the file does not exist.
        """
        self.resume_checkpoint_path = self.hpc_resume_path or self.resume_checkpoint_path
        checkpoint_path = self.resume_checkpoint_path
        if not checkpoint_path:
            return

        # clear cache before restore
        if self.trainer._device_type == DeviceType.GPU:
            torch.cuda.empty_cache()