o }oi@sddlZddlZddlZddlmZddlmZddlmZm Z m Z m Z m Z m Z mZddlZddlmZddlZddlmZddlmZddlmZddlmZdd lmZdd lmZdd l m!Z!dd l"m#Z#Gd ddeZdS)N) timedelta)Path)AnyDictIterableListLiteralOptionalUnion)proxy)ModelCheckpoint)_is_local_file_protocol)rank_zero_info) ckpt_to_dir)TrainerContext)logging)AppStatecseZdZdZdZ             dYd eed ed eeee d fd e dedede dee dee deedeededeffdd Z fddZ ddZddZfdd Zd!eeefd"dffd#d$ ZdZfd%d& Zfd'd(Zd)ed"dfd*d+Zd[d.d/Zed0eeefd"efd1d2Zed0eeefd"efd3d4Zed\d0eeefd"dfd5d6Zed\d0eeefd"dfd7d8Zd]d)ed,d-d9ed"efd:d;Zd,d Z!d\d,dsz?ModelCheckpoint.nemo_topk_check_previous_run..i -last.ckpt-lastr6z[A-z]rFTkeyreverserzNumber of models to delete: r5zRemoved checkpoint: ) best_k_modelskth_best_model_pathbest_model_scorebest_model_pathAttributeErrorlist_saved_checkpoint_pathsrKfindrrDresearchstartfloatrsortedgetrmaxrr@_has_ema_ckptsrangepop_del_model_without_trainer_fsrG_ema_format_filepath) r+ checkpoints checkpointindexmatchvalue_reversertmodels_to_delete ema_enabled_modelr%rlr/rAsZ             z,ModelCheckpoint.nemo_topk_check_previous_runcsdtdtffdd fddjD_tjdkrHjdk}tjjj|d }|d _jj_ |d_ jj _ d Sd _d _ d _ d _ d S) a Removes invalid (incomplete or non-existing) checkpoints from the list of top-k checkpoints. This function is necessary when checkpointing might have been abruptly interrupted, leaving behind incomplete or corrupted checkpoints. The invalid checkpoints are identified by checking if their corresponding directory exists and if the checkpoint is not unfinished. After removing invalid entries, the method updates the best-k models based on the existing, valid checkpoints. Attributes Updated: - `best_k_models`: A dictionary of valid checkpoints from top-k models. - `best_model_path`: Path to the best model based on the current sorting order. - `best_model_score`: The score associated with the best model. - `kth_best_model_path`: Path to the kth best model. - `kth_value`: The score associated with the kth best model. ckpt_pathreturncs"tj|d}|o| S)N.ckpt)osrkisdir removesuffixis_checkpoint_unfinished)rrGrlr%r/ __is_ckpt_ok"szGModelCheckpoint._remove_invalid_entries_from_topk..__is_ckpt_okcsi|] \}}|r||qSr%r%)rjkv)_ModelCheckpoint__is_ckpt_okr%r/ &szEModelCheckpoint._remove_invalid_entries_from_topk..rrrqr5r$N) rKboolrtitemsrDrrrru kth_valuerwrv)r+ reverse_arrbest_k_models_arrr%)rr+r/!_remove_invalid_entries_from_topks    z1ModelCheckpoint._remove_invalid_entries_from_topkcs"t}|jdkr|j|d<|S)a Returns the state dictionary of the model. This function adds additional logic to handle the case when using symlinks. If the model is configured to save the last checkpoint as a symlink, the path to the last checkpoint is updated in the returned state dictionary to avoid off-by-one errors in the checkpointing system. Returns: Dict[str, Any]: The state dictionary of the model, including any necessary modifications for symlinks. rlast_model_path)r) state_dictrr&)r+stater-r%r/r4s  zModelCheckpoint.state_dictrrcst||dS)ay Loads the state dictionary into the model and removes invalid entries from the top-k checkpoints. This method ensures that after loading the model state, any invalid (incomplete or missing) checkpoints are removed from the top-k models list. Args: state_dict (Dict[str, Any]): The state dictionary to load into the model. N)r)load_state_dictr)r+rr-r%r/rEs zModelCheckpoint.load_state_dictcsjddlm}|rtdt|jtj rtj t |j dd|_ tj|g|Ri|dS)aQ Initializes the model and removes any unfinished checkpoints before training. This method is responsible for ensuring that unfinished checkpoints are removed prior to starting the training. It also synchronizes all ranks in a distributed setting to ensure that unfinished checkpoints are removed across all ranks. Args: trainer: The trainer instance used for training. *args: Additional arguments passed to the parent setup method. **kwargs: Additional keyword arguments passed to the parent setup method. rr2z)Removing unfinished checkpoints if any... async_saveFN)r=r3rr@r _remove_unfinished_checkpointsdirpathrTrUrVrWgetattrstrategyrr)setup)r+rYargsr,r3r-r%r/rRs     zModelCheckpoint.setupcsddlm}|jr dS|jrq|jdkrqd}t|jtr%|j|jdkr%d}t|jtr5|j|jdkr5d}|rX| |}|j | ||j krQt d|j dnt|||jrq|jsq|rqt|jt|j dd gd t||dS) a Handles actions to be performed when training ends, such as saving the last checkpoint. This method ensures that the last checkpoint is saved if needed, particularly when validation steps aren't always run based on the interval. It also manages saving the training context to disk, if configured. Args: trainer: The trainer instance used for training. pl_module: The model being trained. rr2NFTzLast checkpoint z already savedcontextr yaml_attrs)r=r3 fast_dev_runrval_check_interval isinstancer global_stepint_monitor_candidatesrformat_checkpoint_nameCHECKPOINT_NAME_LASTrr@r)_save_last_checkpointr#r"r from_trainerio_dumpr on_train_end)r+rYrZr3should_save_last_checkpointmonitor_candidatesr-r%r/rks&   zModelCheckpoint.on_train_endfilepathcCs|ddlm}t|}|r0zt|}tj|ddtd|Wn td|dYtj r._cbr%)r+rYrrr r%r r/r|sz6ModelCheckpoint._get_finalize_save_checkpoint_callbackc s|jr|st|jdkr|jg|jd|dS|j|ddz t||WntyI}zt d|d|WYd}~nd}~ww| |}|dur| |}z t||Wnty~}zt d|d|WYd}~nd}~ww|j |dddS) zPerforms checkpoint removal. With async save, `self._remove_checkpoint` is called before the checkpoint is actually finished so we can't remove it. Instead we add it to `self.deferred_ckpts_to_remove` for future removal. rr5NTrzJError removing checkpoint, common if doing manual cleanup and restarting: z: r) rrDr'rrr)r Exceptionrwarningrrr)r+rYrrerr-r%r/rs4    z"ModelCheckpoint._remove_checkpointcCs||jd|jS)zFormats given path for EMA checkpoint Args: filepath (str): filepath Returns: str: EMA-formatted filepath r)rFILE_EXTENSIONr+rr%r%r/r z$ModelCheckpoint._ema_format_filepathrcstfdd|DS)zCheckes whether filepaths are EMA-formatted Args: checkpoints (Iterable[Path]): paths to check Returns: bool: True indicates path is EMA-formatted. c3s|]}|VqdSrgrh)rjrrlr%r/rmsz1ModelCheckpoint._has_ema_ckpts..)any)r+rr%rlr/rrzModelCheckpoint._has_ema_ckptscCst|d|jS)zCheckes whether filepaths are EMA-formatted Args: filepath (Union[Path, str]): path to check Returns: bool: True indicates path is EMA-formatted. r)rKrrrr%r%r/rirz ModelCheckpoint._is_ema_filepathcs\ddtjdD}|rtfdd|SddtjdD}tfdd|S) a Retrieves a list of saved checkpoint paths while filtering out unfinished checkpoints. - If distributed checkpoints (directories) exist, return only those. - Otherwise, return individual checkpoint files with a .ckpt extension. - Filters out any checkpoints that are marked as unfinished. Returns: Iterable[Path]: An iterable containing valid checkpoint paths. cSsg|]}|r|qSr%)rFrjdr%r%r/ sz;ModelCheckpoint._saved_checkpoint_paths..*c | Srgrprlr%r/ z9ModelCheckpoint._saved_checkpoint_paths..cSsg|]}|qSr%r%rjfr%r%r/rs*.ckptcrrgrrrlr%r/rr)rrrEfilterrglob)r+dist_checkpointscheckpoint_filesr%rlr/rzs z'ModelCheckpoint._saved_checkpoint_pathscheckpoint_dirc Csddlm}|s tdt|}dd|dtjD}dd|dD}|D]}t|}||vrCt d |t |q+d d|dD}|D]}t|}||vrht d |t |qP|D]}t |qkd S) aB Removes all unfinished checkpoints and their associated marker files from the filesystem. - Ensures this function runs only on rank 0. - Deletes individual unfinished checkpoint files. - Removes directories corresponding to unfinished distributed checkpoints. - Deletes the marker files indicating unfinished checkpoints. Args: checkpoint_dir (Union[Path, str]): Path to the directory containing checkpoints. Raises: AssertionError: If the function is called from a non-rank 0 process. rr2z8_remove_unfinished_checkpoints should run only on rank 0cSh|] }|r|qSr%)is_fileresolverr%r%r/ s zAModelCheckpoint._remove_unfinished_checkpoints..rcSsh|]}|qSr%)r)rr%r%r/r*sr!z Removing unfinished checkpoint: cSr'r%)rFr)rr%r%r/r*!sz%Removing unfinished dist checkpoint: N)r=r3AssertionErrorrrEr rr#rrrrremoverIr) r&r3existing_marker_filepathscheckpoint_filepathsrpossible_marker_path all_dirpaths ckpt_dirpathrr%r%r/rs0      z.ModelCheckpoint._remove_unfinished_checkpointspreviouscurrentcCs||krdSt|s dSt|}|jdurt|jnd}|dur8||kr8t|dr6|jdr6ndS|jdurEt|j dt|j}||j vS)aChecks if the previous checkpoint should be deleted. A checkpoint won't be deleted if any of the cases apply: - The previous checkpoint is the same as the current checkpoint (means the old was already overwritten by new) - The previous checkpoint is not in the current checkpoint directory and the filesystem is local - The previous checkpoint is the checkpoint the Trainer resumed from and the filesystem is local and the resumed from checkpoint is not the last checkpoint FTNrnz.dirpath is None.) r rabsoluterrKrrMrrr.r)r+rYr2r3 resume_pathrr%r%r/_should_remove_checkpoint,s   z)ModelCheckpoint._should_remove_checkpoint) rTTrFrNNNFFTT)rN)rYr)F)T)/__name__ __module__ __qualname____doc__rr rKrr rrrr*rXrArrrrrrrrr staticmethodrrrrrrrTTensorrrrrrrrrripropertyrzrr6 __classcell__r%r%r-r/r $s$     1ME & $    M $%   "0r )$rr|rIdatetimerpathlibrtypingrrrrrr r lightninglightning.pytorchpytorchplrT_weakrefr ,lightning.pytorch.callbacks.model_checkpointr PTLModelCheckpointr lightning.pytorch.utilitiesrnemo.lightning.ckpt_utilsrnemo.lightning.io.plr nemo.utilsrnemo.utils.app_staterr%r%r%r/s$  $