o $i @s*ddlZddlZddlZddlZddlZddlZddlZddlZddlm Z ddl m Z m Z m Z mZmZmZmZddlZddlmZddlZddlmZddlmZmZmZddlmZddl m!Z!ddl"m#Z#m$Z$dd l%m&Z&m'Z'dd l(m)Z)dd l*m+Z,dd l-m+Z.dd l/m0Z0dd l1m2Z2ddl3m4Z4e5e6Z7e8dZ9e8dZ:e4ddGdddej;Zde?de@fddZAdej=jBde@fddZCe# d,dee?e,e.fdedde e?e ffd d!ZDe#dee?e,e.fd"e?de?fd#d$ZEe#d%ee?e,e.fd"e?de?fd&d'ZFe4d-d)e@fd*d+ZGdS).N)MappingProxyType)Any CollectionDictListOptionalTupleUnion)version)COMPONENT_LEARNERCOMPONENT_LEARNER_GROUPCOMPONENT_RL_MODULE) force_list)FaultTolerantActorManager) OldAPIStack5OverrideToImplementCustomLogic_CallToSuperRecommended)NOT_SERIALIZABLEserialize_type) StateDict) Checkpoint)sync_dir_between_nodes)log_once) PublicAPIz1.1z2.1alpha) stabilityc @seZdZdZdZdZdZ d(dddddeee e j fdee d ed d e d e f d dZddddee e j fdee d ed d dfddZe d(dee e j fd ed d dfddZej d(dddeee ee fdeee ee fd e fddZejde d dfddZejd eeee efffddZed efddZd eee dffd d!Zd e fd"d#Zd$d%Z d)d&d'Z!dS)*CheckpointableaMAbstract base class for a component of RLlib that can be checkpointed to disk. Subclasses must implement the following APIs: - save_to_path() - restore_from_path() - from_checkpoint() - get_state() - set_state() - get_ctor_args_and_kwargs() - get_metadata() - get_checkpointable_components() statezclass_and_ctor_args.pklz metadata.jsonNF)r filesystem use_msgpackpathrpyarrow.fs.FileSystemrreturncCsL|durddl}tt}t|}|p||}t|tr"|n|}|r3|s3t j j |\}}|j |ddt|}|}d|vrMtt|d<|||j} | t|dWdn1smwY|||j} tt||d| Wdn1swY|du} |p|jdd |Dd }|D]\} } | r| |vrq|| } t| trA| d}dd d }tt | j!||gd }|j"s|#|#}|}d}| rt$%|&| }||kr | j!| ||fdd|gd q||fdd}tt | j!||gd }|j"s%|#|#}t'|||t| |fdd}| j!||gd q| rJ|&| }n|j| d| }| j(| |||dq||j)|rddnd}||*} |rt*dd}||| nt|| Wdt|SWdt|S1swYt|S)aSaves the state of the implementing class (or `state`) to `path`. The state of the implementing class is always saved in the following format: .. testcode:: :skipif: True path/ [component1]/ [component1 subcomponentA]/ ... [component1 subcomponentB]/ ... [component2]/ ... [cls.METADATA_FILE_NAME] (json) [cls.STATE_FILE_NAME] (pkl|msgpack) The main logic is to loop through all subcomponents of this Checkpointable and call their respective `save_to_path` methods. Then save the remaining (non subcomponent) state to this Checkpointable's STATE_FILE_NAME. In the exception that a component is a FaultTolerantActorManager instance, instead of calling `save_to_path` directly on that manager, the first healthy actor is interpreted as the component and its `save_to_path` method is called. Even if that actor is located on another node, the created file is automatically synced to the local node. Args: path: The path to the directory to save the state of the implementing class to. If `path` doesn't exist or is None, then a new directory will be created (and returned). state: An optional state dict to be used instead of getting a new state of the implementing class through `self.get_state()`. filesystem: PyArrow FileSystem to use to access data at the `path`. If not specified, this is inferred from the URI scheme of `path`. use_msgpack: Whether the state file should be written using msgpack and msgpack_numpy (file extension is `.msgpack`), rather than pickle (file extension is `.pkl`). Returns: The path (str) where the state has been saved. NrT recursivecheckpoint_versionzutf-8)classctor_args_and_kwargscSsg|]}|dqSr.0cr(r(X/home/ubuntu/veenaModal/venv/lib/python3.10/site-packages/ray/rllib/utils/checkpoints.py z/Checkpointable.save_to_path..not_componentscSsddl}|jSNr)rayutilget_node_ip_address)_r2r(r(r,_get_ips z,Checkpointable.save_to_path.._get_ipremote_actor_idscSs,|j||durt||dS||dS)Nrr) save_to_pathr2get get_state)w_path_state _use_msgpackr(r(r,sz-Checkpointable.save_to_path..cSs:ddl}|}|j||durt|n||d|S)Nrr9)tempfilemkdtempr:r2r;r<)r=r?r@rBtmpdirr(r(r,_savesz*Checkpointable.save_to_path.._savecSsddl}||dSr1)shutilrmtree)r5_dirrFr(r(r,_rmdir"sz+Checkpointable.save_to_path.._rmdir) components)rrr.msgpack.pklerrorN)+uuidpathlibPathrB gettempdirstruuid4 isinstanceas_posixpyarrowfs FileSystemfrom_uri create_dir get_metadata)CHECKPOINT_VERSION_LEARNER_AND_ENV_RUNNERopen_output_streamMETADATA_FILE_NAMEwritejsondumpsencodeCLASS_AND_CTOR_ARGS_FILE_NAMEpickledumptypeget_ctor_args_and_kwargsr<get_checkpointable_componentsrhealthy_actor_idsnextiter foreach_actorokr;r2putpoprr:STATE_FILE_NAMEtry_import_msgpack)selfrrrrrPtmp_dirrandom_dir_namemetadataf_state_provided comp_namecomp comp_path actor_to_user6_resultworker_ip_addr self_ip_addrcomp_state_refrEworker_temp_dirrI comp_statefilenamemsgpackr(r(r,r:^s4                 zCheckpointable.save_to_path) componentrrc Kszt|tr|n|}|r|stjj|\}}t|}t ||s,t d|ddd| D}|j ||fd|i||dur||j }|drxtdd }||d}|j|d d } Wdn1srwYn||d  }t|} Wdn1swY|| d d| D} | |} | r|j ||fd| i|dSdSdS)a.Restores the state of the implementing class from the given path. If the `component` arg is provided, `path` refers to a checkpoint of a subcomponent of `self`, thus allowing the user to load only the subcomponent's state into `self` without affecting any of the other state information (for example, loading only the NN state into a Checkpointable, which contains such an NN, but also has other state information that should NOT be changed by calling this method). The given `path` should have the following structure and contain the following files: .. testcode:: :skipif: True path/ [component1]/ [component1 subcomponentA]/ ... [component1 subcomponentB]/ ... [component2]/ ... [cls.METADATA_FILE_NAME] (json) [cls.STATE_FILE_NAME] (pkl|msgpack) Note that the self.METADATA_FILE_NAME file is not required to restore the state. Args: path: The path to load the implementing class' state from or to load the state of only one subcomponent's state of the implementing class (if `component` is provided). component: If provided, `path` is interpreted as the checkpoint path of only the subcomponent and thus, only that subcomponent's state is restored/loaded. All other state of `self` remains unchanged in this case. filesystem: PyArrow FileSystem to use to access data at the `path`. If not specified, this is inferred from the URI scheme of `path`. **kwargs: Forward compatibility kwargs. z`path` (z ) not found!cSh|]}|dqSr'r(r)r(r(r, r.z3Checkpointable.restore_from_path..rNrKTrMF)strict_map_keyrLcSrr'r(r)r(r(r,rr.only_comp_names)rVrTrWrXrYrZr[rQrR_exists_at_fs_pathFileNotFoundErrorrj$_restore_all_subcomponents_from_pathrr with_suffixis_filersopen_input_streamloadrf set_state) rtrrrkwargsorig_comp_namesrrrxrnew_comp_namesdiff_comp_namesr(r(r,restore_from_pathFsX0        z Checkpointable.restore_from_pathcKst|tr|n|}|r|stjj|\}}t|}zc| ||j  }t |}Wdn1s8wY|d}t |dd}|dd}tt|jD]%\} \} } | |vr~|| } | jtjjkrzt|| krz| || <qY| || <qYWnty|}g}|}Ynwt||std|d|d|d t|tstd|d|d ||i|} | j|fd |i|| S) aCreates a new Checkpointable instance from the given location and returns it. Args: path: The checkpoint path to load (a) the information on how to construct a new instance of the implementing class and (b) the state to restore the created instance to. filesystem: PyArrow FileSystem to use to access data at the `path`. If not specified, this is inferred from the URI scheme of `path`. kwargs: Forward compatibility kwargs. Note that these kwargs are sent to each subcomponent's `from_checkpoint()` call. Returns: A new instance of the implementing class, already set to the state stored under `path`. Nr%r&rz The class (z) stored in checkpoint (z+) does not seem to be a subclass of `cls` (z)!zA) does not seem to be an implementer of the `Checkpointable` API!r)rVrTrWrXrYrZr[rQrRrrerfrr enumerateinspect signature parametersitemsrqkind_ParameterKindPOSITIONAL_OR_KEYWORDlen Exception issubclass ValueErrorrr)clsrrrrx ctor_infoctor ctor_args ctor_kwargsi param_nameparamvalobjr(r(r,from_checkpointsV          zCheckpointable.from_checkpointr/rJr0cKdS)aReturns the implementing class's current state as a dict. The returned dict must only contain msgpack-serializable data if you want to use the `AlgorithmConfig._msgpack_checkpoints` option. Consider returning your non msgpack-serializable data from the `Checkpointable.get_ctor_args_and_kwargs` method, instead. Args: components: An optional collection of string keys to be included in the returned state. This might be useful, if getting certain components of the state is expensive (e.g. reading/compiling the weights of a large NN) and at the same time, these components are not required by the caller. not_components: An optional list of string keys to be excluded in the returned state, even if the same string is part of `components`. This is useful to get the complete state of the class, except one or a few components. kwargs: Forward-compatibility kwargs. Returns: The current state of the implementing class (or only the `components` specified, w/o those in `not_components`). Nr()rtrJr0rr(r(r,r<zCheckpointable.get_statecCr)abSets the implementing class' state to the given state dict. If component keys are missing in `state`, these components of the implementing class will not be updated/set. Args: state: The state dict to restore the state from. Maps component keys to the corresponding subcomponent's own state. Nr()rtrr(r(r,rrzCheckpointable.set_statecCr)zReturns the args/kwargs used to create `self` from its constructor. Returns: A tuple of the args (as a tuple) and kwargs (as a Dict[str, Any]) used to construct `self` from its class constructor. Nr(rtr(r(r,ri$rz'Checkpointable.get_ctor_args_and_kwargscCs|j|jtjtjdS)aReturns JSON writable metadata further describing the implementing class. Note that this metadata is NOT part of any state and is thus NOT needed to restore the state of a Checkpointable instance from a directory. Rather, the metadata will be written into `self.METADATA_FILE_NAME` when calling `self.save_to_path()` for the user's convenience. Returns: A JSON-encodable dict of metadata information. )class_and_ctor_args_file state_file ray_version ray_commit)rerrr2 __version__ __commit__rr(r(r,r]-s  zCheckpointable.get_metadatacCsgS)aReturns the implementing class's own Checkpointable subcomponents. Returns: A list of 2-tuples (name, subcomponent) describing the implementing class' subcomponents, all of which have to be `Checkpointable` themselves and whose state is therefore written into subdirectories (rather than the main state file (self.STATE_FILE_NAME) when calling `self.save_to_path()`). r(rr(r(r,rj@s z,Checkpointable.get_checkpointable_componentscsNt|}t|}|dustfdd|Ds|vr%|dus#|vr%dSdS)aoReturns True if a component should be checkpointed. Args: name: The checkpoint name. components: A list of components that should be checkpointed. non_components: A list of components that should not be checkpointed. Returns: True, if the component should be checkpointed and otherwise False. Nc3s|] }|dVqdS)/N) startswithr)namer(r, Zsz2Checkpointable._check_component..TF)rany)rtrrJr0 comp_list not_comp_listr(rr,_check_componentKs zCheckpointable._check_componentcCsV|durdSt|}g}|D]}||dr$||t|ddq|s)dS|S)Nrr)rrappendr)rtrrJ subcomponentsr{r(r(r,_get_subcomponents`s z!Checkpointable._get_subcomponentsc Ks|D]f\}}|dur||vrqd}|dur$||} t|| s#qn|} ||dr8|t|dd}n||kr=qt|tr^tj } | } t || | |fdd} |j | | dq|j | f||d|qdS)Nrrc Ssddl}ddl}|j}||kr|j|fd|i|dS|}t|||||j|fd|i|WddS1s@wYdS)Nrr)rBr2r3r4rTemporaryDirectoryr) r=_kwargsr>_head_ip _comp_argrBr2worker_node_iptemp_dirr(r(r,_restores"  "zECheckpointable._restore_all_subcomponents_from_path.._restorer7)rr)rjrrWrrrVrr2r3r4rkrrnr) rtrrrrrrzr{comp_argcomp_dir head_node_ipall_healthy_actorsrr(r(r,rls>   z3Checkpointable._restore_all_subcomponents_from_pathrO)NN)"__name__ __module__ __qualname____doc__rrrer`rr rTrQrRrboolr:r classmethodrabcabstractmethodrr<rrrrrirr]rrjrrrr(r(r(r,r?s m  Z  W    rrYrr!cCs||}|jtjjjkS)z:Returns `True` if the path can be found in the filesystem.) get_file_inforhrXrYFileTypeNotFound)rYrvalidr(r(r,rs r file_infocCs|jtjjjkS)z5Returns `True`, if the file info is from a directory.)rhrXrYr Directory)rr(r(r,_is_dirsr checkpointrr c Csddtddddd}t|ttfr|}|r"|s"tjj|\}}t |}t || rt || r|dt|i|tjj| dd}|D]}|jrotd|jro|td t|jd |SqQt ||d  r||d  }tj|d }Wdn1swYd |vrt|d |d <||n tdrtd|ddD](}t ||d| r|d|dkrdndtt|d|d|Sqd}dD]%}|d|} t || r || jr |dkrdnd}nq|durtd||t| d|d} t || rYt || rYt} |tjj| dd}|D] }| |jqH|d| i|t!t"t#} t || rt || rt} |tjj| dd}|D]}| |j}t || r| |jq|d| i|St || r|| jr|td t|j$t|d|Stdt|d )!aReturns a dict with information about an Algorithm/Policy checkpoint. If the given checkpoint is a >=v1.0 checkpoint directory, try reading all information from the contained `rllib_checkpoint.json` file. Args: checkpoint: The checkpoint directory (str) or a Checkpoint object. filesystem: PyArrow FileSystem to use to access data at the `checkpoint`. If not specified, this is inferred from the URI scheme provided by `checkpoint`. Returns: A dict containing the keys: "type": One of "Policy" or "Algorithm". "checkpoint_version": A version tuple, e.g. v1.0, indicating the checkpoint version. This will help RLlib to remain backward compatible wrt. future Ray and checkpoint versions. "checkpoint_dir": The directory with all the checkpoint files in it. This might be the same as the incoming `checkpoint` arg. "state_file": The main file with the Algorithm/Policy's state information in it. This is usually a pickle-encoded file. "policy_ids": An optional set of PolicyIDs in case we are dealing with an Algorithm checkpoint. None if `checkpoint` is a Policy checkpoint. Algorithm cloudpickleN)rhformatr$checkpoint_dirr policy_ids module_idsrFr"zcheckpoint-\d+z0.1)r$rrllib_checkpoint.json)fpr$no_rllib_checkpoint_json_filez>No `rllib_checkpoint.json` file found in checkpoint directory zG! Trying to extract checkpoint info from other files found in that dir.)pklmsgpckz policy_state.Policyrr)rhrr$r)rrrzalgorithm_state.zGiven checkpoint does not seem to be valid! No file with the name `algorithm_state.[pkl|msgpack|msgpck]` (or `checkpoint-[0-9]+`) found.)rrpoliciesrr)r$rrzGiven checkpoint (zV) not found! Must be a checkpoint directory (or a file for older checkpoint versions).)%CHECKPOINT_VERSIONrVCheckpoint_trainCheckpoint_tune to_directoryrXrYrZr[rQrRrrWrrupdaterT FileSelectorrrematch base_namer Versionrrbrrloggerwarningrsetaddr r r parent)rrinfofile_info_listrrxrllib_checkpoint_info extensionrr policies_dirr modules_dirr module_dirr(r(r,get_checkpoint_infos             rmsgpack_checkpoint_dirc Csddlm}ddlm}ddlm}tdd}||}|}t |d|d<t |dt s8|d |d<n | |d|d<i}d |vrUd |d vrU|d d i}t|d d <t|d d <tt|d <tj|d} t| d} ||| Wdn1swYttj|dd} td|d d| t|tjtjd| Wdn1swY|D]&\} } || ddtj|d| } tj| dd|| }|j | | ddq|!|S)a Converts an Algorithm checkpoint (pickle based) to a msgpack based one. Msgpack has the advantage of being python version independent. Args: checkpoint: The directory, in which to find the Algorithm checkpoint (pickle based). msgpack_checkpoint_dir: The directory, in which to create the new msgpack based checkpoint. Returns: The directory in which the msgpack checkpoint has been created. Note that this is the same as `msgpack_checkpoint_dir`. r)r)AlgorithmConfig)validate_module_idTrMalgorithm_classconfigworker policy_statespolicy_mapping_fnis_policy_to_trainr$zalgorithm_state.msgpckwbNrr=rr)rhr$rrrrrrexist_ok policy_statecheckpoint_format)"ray.rllib.algorithmsr%ray.rllib.algorithms.algorithm_configrray.rllib.core.rl_modulerrsr __getstate__rrVdict serialize_serialize_dictrqrrTrosrjoinopenrgrblistkeysr2rrrmakedirs get_policyexport_checkpointstop)rrrrrralgorr rrxpidr policy_dirpolicyr(r(r,convert_to_msgpack_checkpointsZ             r'policy_checkpointcCs>ddlm}||}tj|dd|j||dd~|S)a Converts a Policy checkpoint (pickle based) to a msgpack based one. Msgpack has the advantage of being python version independent. Args: policy_checkpoint: The directory, in which to find the Policy checkpoint (pickle based). msgpack_checkpoint_dir: The directory, in which to create the new msgpack based checkpoint. Returns: The directory in which the msgpack checkpoint has been created. Note that this is the same as `msgpack_checkpoint_dir`. r)rTrrr)ray.rllib.policy.policyrrrrr!r<)r(rrr&r(r(r,$convert_to_msgpack_policy_checkpoints  r*FrNcCs@zddl}ddl}||WSty|rtdYdSw)alTries importing msgpack and msgpack_numpy and returns the patched msgpack module. Returns None if error is False and msgpack or msgpack_numpy is not installed. Raises an error, if error is True and the modules could not be imported. Args: error: Whether to raise an error if msgpack/msgpack_numpy cannot be imported. Returns: The `msgpack` module, with the msgpack_numpy module already patched in. This means you can already encde and decode numpy arrays with the returned module. Raises: ImportError: If error=True and msgpack/msgpack_numpy is not installed. rNzkCould not import or setup msgpack and msgpack_numpy! Try running `pip install msgpack msgpack_numpy` first.)r msgpack_numpypatchr ImportError)rNrr+r(r(r,rss rsrO)F)HrrrbloggingrrQrrBtypesrtypingrrrrrrr pyarrow.fsrX packagingr r2ray.cloudpicklerrfray.rllib.corer r r ray.rllib.utilsrray.rllib.utils.actor_managerrray.rllib.utils.annotationsrrray.rllib.utils.serializationrrray.rllib.utils.typingr ray.trainrrray.tunerray.tune.utils.file_transferrray.utilrray.util.annotationsr getLoggerrrrrr^ABCrrYrZrTrrFileInforrr'r*rsr(r(r(r,s $             w   G  [  #