Repository: Lightning-AI/pytorch-lightning
File: pytorch_lightning/callbacks/model_checkpoint.py (Python)

Code before:

    def _get_metric_interpolated_filepath_name(self, monitor_candidates: Dict[str, Any], epoch: int, step: int) -> str:
        original_filepath = self.format_checkpoint_name(epoch, step, monitor_candidates)
        v0_filepath = self.format_checkpoint_name(epoch, step, monitor_candidates, ver=0)
        if not self._fs.exists(original_filepath) and not self._fs.exists(v0_filepath):
            # first time
            return original_filepath
        elif self._fs.exists(original_filepath) and not self._fs.exists(v0_filepath):
            # re-run special case, rename file to file-v0
            self._fs.move(original_filepath, v0_filepath)
            if original_filepath in self.best_k_models:
                self.best_k_models[v0_filepath] = self.best_k_models.pop(original_filepath)
            if original_filepath == self.kth_best_model_path:
                self.kth_best_model_path = v0_filepath
            if original_filepath == self.best_model_path:
                self.best_model_path = v0_filepath
        # re-run, use the vesion suffix
        version = 0
        filepath = v0_filepath
        while self._fs.exists(filepath):
            version += 1
            filepath = self.format_checkpoint_name(epoch, step, monitor_candidates, ver=version)
        return filepath

    def _monitor_candidates(self, trainer):
        monitor_candidates = deepcopy(trainer.logger_connector.logged_metrics)
        monitor_candidates.update(trainer.logger_connector.callback_metrics)
        monitor_candidates.update(trainer.logger_connector.progress_bar_metrics)
        monitor_candidates.update({"step": trainer.global_step, "epoch": trainer.current_epoch})
        return monitor_candidates

    def _save_last_checkpoint(self, trainer, pl_module, monitor_candidates, filepath):
        should_save_last = self.monitor is None or self.save_last
        if not should_save_last:
            return

        last_filepath = filepath

        # when user ALSO asked for the 'last.ckpt' change the name
        if self.save_last:
            last_filepath = self._format_checkpoint_name(
                self.CHECKPOINT_NAME_LAST,
                trainer.current_epoch,
                trainer.global_step,
                monitor_candidates,
                prefix=self.prefix,
            )
            last_filepath = os.path.join(self.dirpath, f"{last_filepath}{self.FILE_EXTENSION}")

        self._save_model(last_filepath, trainer, pl_module)
        if (

Review comment (suggestion):
logged_metrics and progress_bar_metrics should not be needed here as they are already included in `callback_metrics`.

Code after:
        return monitor_candidates

    def _save_last_checkpoint(self, trainer, pl_module, monitor_candidates):
        should_save_last = self.monitor is None or self.save_last
        if not should_save_last:
            return

        # when user ALSO asked for the 'last.ckpt' change the name
        if self.save_last:
            last_filepath = self._format_checkpoint_name(
                self.CHECKPOINT_NAME_LAST,
                trainer.current_epoch,
                trainer.global_step,
                monitor_candidates,
                prefix=self.prefix,
            )
            last_filepath = os.path.join(self.dirpath, f"{last_filepath}{self.FILE_EXTENSION}")
        else:
            last_filepath = self._get_metric_interpolated_filepath_name(
                monitor_candidates, trainer.current_epoch, trainer.global_step
            )

        accelerator_backend = trainer.accelerator_backend

        if accelerator_backend is not None and accelerator_backend.rpc_enabled:
            # RPCPlugin manages saving all model states
            accelerator_backend.ddp_plugin.rpc_save_model(self._save_model, last_filepath, trainer, pl_module)
        else:
            self._save_model(last_filepath, trainer, pl_module)
        if (
            self.last_model_path
            and self.last_model_path != last_filepath
            and (self.save_top_k != -1 or self.save_last)
            and trainer.is_global_zero
        ):
            self._del_model(self.last_model_path)
        self.last_model_path = last_filepath

        if self.monitor is None:
            self.best_model_path = self.last_model_path

    def _save_top_k_checkpoints(self, trainer, pl_module, metrics):
        current = metrics.get(self.monitor)
        epoch = metrics.get("epoch")
        step = metrics.get("step")

        if self.check_monitor_top_k(current):
            self._update_best_and_save(current, epoch, step, trainer, pl_module, metrics)
        elif self.verbose:
            rank_zero_info(
                f"Epoch {epoch:d}, step {step:d}: {self.monitor} was not in top {self.save_top_k}"