o
    vi                     @   st   d dl Z d dlmZmZ d dlmZ d dlZd dlmZm	Z	 d dl
mZ d dlmZ d dlmZ G dd	 d	eZdS )
    N)ABCabstractmethod)Optional)CheckpointConfig	JobConfig)ImaginaireModel)callback)easy_ioc                   @   s  e Zd ZdZ	d$dededeej fddZ	e
ded	ejjd
ejjjdejjdeddfddZe
			d%ded	eejj d
eejjj deejj def
ddZedd Zedd Zedd Zedd Zd&ddZdedB fddZdeddfdd Zd!eddfd"d#ZdS )'AbstractCheckpointerz^The checkpointer class. Supports checkpoint saving/loading to both local disk or object store.Nconfig_checkpoint
config_job	callbacksc                 C   s  || _ || _|jj| _|jj| _tj|jd| _	tj|jd| _
|j| _|jp*d| _|j| _|j| _d| _|j| _|j| _|j| _|jjr`d| _tjdddd|jj di|jjdd	 nd| _|jjrd
| _tjd
ddd|jj di|jjdd	 dS d| _dS )zConstructor of the checkpointer.

        Args:
            config_checkpoint (CheckpointConfig): The config object for the checkpointer.
        checkpointsNZ_ckpt_s3_loaders3z
s3://ckpt/s3:///)backendpath_mappings3_credential_path)keybackend_argsZ_ckpt_s3_saver)r   r   save_to_object_storeenabledload_from_object_storeospathjoin
path_local_local_dirname_object_store_dirnamestrict_resume	load_pathload_training_stateonly_load_scheduler_statesave_threadverbosekeys_not_to_resumebroadcast_via_filesystemload_s3_backend_keyr	   set_s3_backendbucketcredentialssave_s3_backend_key)selfr   r   r    r.   U/data/cameron/vidgen/cosmos-policy/cosmos_policy/_src/imaginaire/checkpointer/base.py__init__   sH   



zAbstractCheckpointer.__init__model	optimizer	schedulergrad_scaler	iterationreturnc                 C      d S Nr.   )r-   r1   r2   r3   r4   r5   r.   r.   r/   save[   s   	zAbstractCheckpointer.savec                 C   r7   r8   r.   )r-   r1   r2   r3   r4   r.   r.   r/   loadf   s   zAbstractCheckpointer.loadc                 C      | j r| jj jS dS )z+Get the bucket name for saving checkpoints.N)r   r   r*   r-   r.   r.   r/   save_bucketp      z AbstractCheckpointer.save_bucketc                 C   r;   )z,Get the bucket name for loading checkpoints.N)r   r   r*   r<   r.   r.   r/   load_bucketu   r>   z AbstractCheckpointer.load_bucketc                 C       | j rd| j d| j S | jS Nr   r   )r   r=   r   r   r<   r.   r.   r/   save_dirnamez   
   z!AbstractCheckpointer.save_dirnamec                 C   r@   rA   )r   r?   r   r   r<   r.   r.   r/   load_dirname   rC   z!AbstractCheckpointer.load_dirnamec                 C   s   | j r
| j   dS dS )zFinalize the checkpointer.N)r$   r   r<   r.   r.   r/   finalize   s   zAbstractCheckpointer.finalizec                 C   s@   d}t j| jd}tj| | jdrtj| | jd }|S )zGet the file name of the latest saved checkpoint. If it doesn't exist, return None.

        Returns:
            checkpoint_file (str | None): file name of the latest saved checkpoint.
        Nlatest_checkpoint.txtbackend_key)	r   r   r   rD   r	   existsr(   r:   strip)r-   checkpoint_filecheckpoint_pathr.   r.   r/   _read_latest_checkpoint_file   s
   z1AbstractCheckpointer._read_latest_checkpoint_filerK   c                 C   s0   | d}t j| jd}tj||| jd dS )zTrack the file name of the latest saved checkpoint.

        Args:
            checkpoint_file (str): file name of the latest saved checkpoint.
        
rF   rG   N)r   r   r   rB   r	   dumpr,   )r-   rK   contentrL   r.   r.   r/   _write_latest_checkpoint_file   s   

z2AbstractCheckpointer._write_latest_checkpoint_filerL   c                 C   s$   t j| | jdstd| dS )zIf the file checkpoint_path does not exist, raise an error.

        Args:
            checkpoint_path (str): full path to the checkpoint.
        rG   zFile not found (object store): N)r	   rI   r(   FileNotFoundError)r-   rL   r.   r.   r/   _check_checkpoint_exists   s   z-AbstractCheckpointer._check_checkpoint_existsr8   )NNN)r6   N) __name__
__module____qualname____doc__r   r   r   r   CallBackGroupr0   r   r   torchoptim	Optimizerlr_schedulerLRScheduleramp
GradScalerintr9   r:   propertyr=   r?   rB   rD   rE   strrM   rQ   rS   r.   r.   r.   r/   r
      sf    
<


	




r
   )r   abcr   r   typingr   rY   $cosmos_policy._src.imaginaire.configr   r   #cosmos_policy._src.imaginaire.modelr   #cosmos_policy._src.imaginaire.utilsr   +cosmos_policy._src.imaginaire.utils.easy_ior	   r
   r.   r.   r.   r/   <module>   s   