&`iBddlZddlZddlZddlZddlZddlZddlZddlZddlZddl Z ddl m Z ddl m Z mZmZmZmZmZddlZddlmZddlmZmZddlmZejeZdZdZ Gd d e!Z"ed Gd de"Z#dde$de$de$fdZ%de$dee$fdZ&de$fdZ'de$fdZ(de$fdZ)dS)N)Path)AnyDictIteratorListOptionalUnion) TempFileLock)_download_from_fs_path_exists_at_fs_path) PublicAPIz.metadata.jsoncheckpoint_tmp_ceZdZfdZxZS)_CheckpointMetaClassc t|S#t$rC}|dvrt|||dvrt |||dvrt |||d}~wwxYw)N>to_dictto_bytes from_dict from_bytesget_internal_representation>urito_urifrom_uri>get_preprocessorset_preprocessor)super__getattribute__AttributeError_get_migration_error_get_uri_error_get_preprocessor_error)selfitemexc __class__s i/home/jaya/work/projects/VOICE-AGENT/VIET/agent-env/lib/python3.11/site-packages/ray/train/_checkpoint.py __getattr__z _CheckpointMetaClass.__getattr__s 77++D11 1   +400c9 %T**3AAA-d33<I% s $ A1>A,,A1)__name__ __module__ __qualname__r' __classcell__)r%s@r&rrs8rbeta) stabilitycdeZdZdZ ddeeejfdedfdZ dZ de ee ffd Z d e ee fddfd Zd e ee fddfd Zedeeejfddfd ZddeeeejfdefdZejdeefdZdefdZdZdS) CheckpointaA reference to data persisted as a directory in local or remote storage. Access the checkpoint contents locally using ``checkpoint.to_directory()`` or ``checkpoint.as_directory``. Attributes ---------- path: A path on the filesystem containing the checkpoint contents. filesystem: PyArrow FileSystem that can be used to access data at the `path`. See Also -------- ray.train.report : Report a checkpoint during training (with Ray Train/Tune). ray.train.get_checkpoint : Get the latest checkpoint during training (for restoration). :ref:`train-checkpointing` :ref:`persistent-storage-guide` Examples -------- Creating a checkpoint using ``Checkpoint.from_directory``: >>> from ray.train import Checkpoint >>> checkpoint = Checkpoint.from_directory("/tmp/example_checkpoint_dir") >>> checkpoint.filesystem # doctest: +ELLIPSIS >> checkpoint.path '/tmp/example_checkpoint_dir' Creating a checkpoint from a remote URI: >>> checkpoint = Checkpoint("s3://bucket/path/to/checkpoint") >>> checkpoint.filesystem # doctest: +ELLIPSIS >> checkpoint.path 'bucket/path/to/checkpoint' Creating a checkpoint with a custom filesystem: >>> checkpoint = Checkpoint( ... path="bucket/path/to/checkpoint", ... filesystem=pyarrow.fs.S3FileSystem(), ... ) >>> checkpoint.filesystem # doctest: +ELLIPSIS >> checkpoint.path 'bucket/path/to/checkpoint' Accessing a checkpoint's contents: >>> import os # doctest: +SKIP >>> with checkpoint.as_directory() as local_checkpoint_dir: # doctest: +SKIP ... print(os.listdir(local_checkpoint_dir)) # doctest: +SKIP ['model.pt', 'optimizer.pt', 'misc.pt'] Npath filesystemzpyarrow.fs.FileSystemct||_||_|r3|s1tjj|\|_|_tj|_ dS)aConstruct a Checkpoint. Args: path: A local path or remote URI containing the checkpoint data. If a filesystem is provided, then this path must NOT be a URI. It should be a path on the filesystem with the prefix already stripped. filesystem: PyArrow FileSystem to use to access data at the path. If not specified, this is inferred from the URI scheme. N) strr1r2pyarrowfs FileSystemruuiduuid4_uuid)r"r1r2s r&__init__zCheckpoint.__init__ss]II $  N  N)0)>)G)G)M)M &DOTYZ\\ r,c2d|jjd|jdS)NzCheckpoint(filesystem=z, path=))r2 type_namer1r"s r&__repr__zCheckpoint.__repr__s"V(AVV$)VVVVr,returncbt|jt}t |j|siS|j|5}tj| dcdddS#1swxYwYdS)z{Return the metadata dict stored with the checkpoint. If no metadata is stored, an empty dict is returned. utf-8N) rr1_METADATA_FILE_NAMEas_posixr r2open_input_filejsonloadsreadalldecode)r" metadata_pathfs r& get_metadatazCheckpoint.get_metadatas TY(;<<EEGG !$/=AA I _ , ,] ; ; ;q:aiikk0099:: ; ; ; ; ; ; ; ; ; ; ; ; ; ; ; ; ; ;s9B$$B(+B(metadatac8t|jt}|j|5}|tj| dddddS#1swxYwYdS)zSet the metadata stored with this checkpoint. This will overwrite any existing metadata stored with this checkpoint. rCN) rr1rDrEr2open_output_streamwriterGdumpsencode)r"rNrKrLs r& set_metadatazCheckpoint.set_metadatas TY(;<<EEGG _ / / > > :! GGDJx((//88 9 9 9 : : : : : : : : : : : : : : : : : :s;BBBc|}||||dS)zUpdate the metadata stored with this checkpoint. This will update any existing metadata stored with this checkpoint. N)rMupdaterT)r"rNexisting_metadatas r&update_metadatazCheckpoint.update_metadatasE !--//  *** +,,,,,r,cT||tjS)zCreate checkpoint object from a local directory. Args: path: Local directory containing checkpoint data. Returns: A ray.train.Checkpoint object. )r2)r5r6LocalFileSystem)clsr1s r&from_directoryzCheckpoint.from_directorys's4GJ$>$>$@$@AAAAr,cb|du}|r|n|}tjtjt |}tj|d t|d5t|j |j|dddn #1swxYwYnl#t$r_t|d5 dddn #1swxYwYtj |std|d YnwxYw|S) a9Write checkpoint data to a local directory. *If multiple processes on the same node call this method simultaneously,* only a single process will perform the download, while the others wait for the download to finish. Once the download finishes, all processes receive the same local directory to read from. Args: path: Target directory to download data to. If not specified, this method will use a temporary directory. Returns: str: Directory containing checkpoint data. NT)exist_okrtimeout)r6fs_path local_pathzCheckpoint directory z does not exist, even though it should have been created by another process. Please raise an issue on GitHub: https://github.com/ray-project/ray/issues) _get_temporary_checkpoint_dirosr1normpath expanduserr4makedirsr r r2 TimeoutErrorexists RuntimeError)r"r1user_provided_pathrbs r& to_directoryzCheckpoint.to_directorys"-& PDDD,N,N,P,P W%%bg&8&8Z&I&IJJ  J.... j!444  & j                  j"555                 7>>*-- "@J@@@   sZ=CB7+ C7B;;C>B;?CD,C, D,,C0 0D,3C0 45D,+D,c #Kt|jtjjr |jVdSt |}t|d | } |V tj |nA#t$r4td|dt!jYnwxYwt%|}|sT t'|d5t)j|dddddS#1swxYwYdS#t,$rYdSwxYwdS# tj |nA#t$r4td|dt!jYnwxYwt%|}|sQ t'|d5t)j|ddddw#1swxYwYw#t,$rYwwxYwwxYw) aEReturns checkpoint contents in a local directory as a context. This function makes checkpoint data available as a directory while avoiding unnecessary copies and left-over temporary data. *If the checkpoint points to a local directory*, this method just returns the local directory path without making a copy, and nothing will be cleaned up after exiting the context. *If the checkpoint points to a remote directory*, this method will download the checkpoint to a local temporary directory and return the path to the temporary directory. *If multiple processes on the same node call this method simultaneously,* only a single process will perform the download, while the others wait for the download to finish. Once the download finishes, all processes receive the same local (temporary) directory to read from. Once all processes have finished working with the checkpoint, the temporary directory is cleaned up. Users should treat the returned checkpoint directory as read-only and avoid changing any data within it, as it may be deleted when exiting the context. Example: .. testcode:: :hide: from pathlib import Path import tempfile from ray.train import Checkpoint temp_dir = tempfile.mkdtemp() (Path(temp_dir) / "example.txt").write_text("example checkpoint data") checkpoint = Checkpoint.from_directory(temp_dir) .. testcode:: with checkpoint.as_directory() as checkpoint_dir: # Do some read-only processing of files within checkpoint_dir pass # At this point, if a temporary directory was created, it will have # been deleted. azCould not remove z deletion file lock. Traceback: rr_T) ignore_errorsN) isinstancer2r5r6rZr1_get_del_lock_pathrdopenclosermreremove Exceptionloggerwarning traceback format_exc_list_existing_del_locksr shutilrmtreeri)r" del_lock_pathtemp_dirremaining_lockss r& as_directoryzCheckpoint.as_directorysHd dowz'A B B$ )OOOOO.t/Q/Q/S/STTM  $ $ * * , , ,((**H  Im,,,, NN@M@@'0';'='=@@#;8"D"D&*(A>>>HH"M($GGGGHHHHHHHHHHHHHHHHHH'Im,,,, NN@M@@'0';'='=@@#;8"D"D&*(A>>>HH"M($GGGGHHHHHHHHHHHHHHHH's E B$$;C"!C"7D9D, D9,D00D93D04D99 EE HE#"H#;F!H F!!H6G6G* G6)H*G. .G61G. 2G65H6 HHHHctj}t|jjz}t jdkrtd}t|jjdttzt|zt|zdz}| tstdt|| S)zReturn the name for the temporary checkpoint dir that this checkpoint will get downloaded to, if accessing via `to_directory` or `as_directory`. WindowsiNziCouldn't create checkpoint directory due to length constraints. Try specifying a shorter checkpoint path.) tempfile gettempdir_CHECKPOINT_TEMP_DIR_PREFIXr:hexplatformsystemrrlen startswithrkrrE)r" tmp_dir_pathcheckpoint_dir_name del_lock_names r&rdz(Checkpoint._get_temporary_checkpoint_dirAs *,, 9DJNJ ?   ) )/r22M+*.5667,''(-(()++ '112MNN "ML"566??AAAr,c td)Nz~You cannot use `Checkpoint` objects directly as paths. Use `Checkpoint.to_directory()` or `Checkpoint.as_directory()` instead.) TypeErrorr?s r& __fspath__zCheckpoint.__fspath__[s V   r,N)r(r)r*__doc__r r4rePathLikerr;r@rrrMrTrX classmethodr\rm contextlibcontextmanagerrrrdrr,r&r0r07s88z9=""C$%"45""""<WWW ;d38n ; ; ; ;:T#s(^:::::-S#X-4---- B%R[(8"9 Bl B B B[ B))%R[0@*A!B)c))))VUhsmUUUUnBsBBBB4     r,r0) metaclassr1suffixrAc~||nttj}|dd|S)aGet the path to the deletion lock file for a file/directory at `path`. Example: >>> _get_del_lock_path("/tmp/checkpoint_tmp") # doctest: +ELLIPSIS '/tmp/checkpoint_tmp.del_lock_... >>> _get_del_lock_path("/tmp/checkpoint_tmp/") # doctest: +ELLIPSIS '/tmp/checkpoint_tmp.del_lock_... >>> _get_del_lock_path("/tmp/checkpoint_tmp.txt") # doctest: +ELLIPSIS '/tmp/checkpoint_tmp.txt.del_lock_... N/z .del_lock_)r4regetpidrstrip)r1rs r&rrrrbs@)VVs29;;/?/?Fkk# 2 2& 2 22r,cdttjt|dS)zList all the deletion lock files for a file/directory at `path`. For example, if 2 checkpoints are being read via `as_directory`, then this should return a list of 2 deletion lock files. *)r)listglobrr)r1s r&r{r{ss-  /SAAACDD E EEr,namec(td|dS)N7The new `ray.train.Checkpoint` class does not support `a()`. Instead, only directories are supported. Example to store a dictionary in a checkpoint: import os, tempfile import ray.cloudpickle as pickle from ray import train from ray.train import Checkpoint with tempfile.TemporaryDirectory() as checkpoint_dir: with open(os.path.join(checkpoint_dir, 'data.pkl'), 'wb') as fp: pickle.dump({'data': 'value'}, fp) checkpoint = Checkpoint.from_directory(checkpoint_dir) train.report(..., checkpoint=checkpoint) Example to load a dictionary from a checkpoint: if train.get_checkpoint(): with train.get_checkpoint().as_directory() as checkpoint_dir: with open(os.path.join(checkpoint_dir, 'data.pkl'), 'rb') as fp: data = pickle.load(fp)rrs r&rr|s*  ($ ( ( (  r,c(td|dS)Nral()`. To create a checkpoint from remote storage, create a `Checkpoint` using its constructor instead of `from_directory`. Example: `Checkpoint(path="s3://a/b/c")`. Then, access the contents of the checkpoint with `checkpoint.as_directory()` / `checkpoint.to_directory()`. To upload data to remote storage, use e.g. `pyarrow.fs.FileSystem` or your client of choice.rrs r&r r s*  %$ % % %   r,c(td|dS)Nraf()`. To include preprocessor information in checkpoints, pass it as metadata in the Trainer constructor. Example: `TorchTrainer(..., metadata={...})`. After training, access it in the checkpoint via `checkpoint.get_metadata()`. See here: https://docs.ray.io/en/master/train/user-guides/data-loading-preprocessing.html#preprocessing-structured-datarrs r&r!r!s.  I$ I I I  r,r)*rrrGloggingrerr|rryr8pathlibrtypingrrrrrr pyarrow.fsr5ray.air._internal.filelockr ray.train._internal.storager r ray.util.annotationsr getLoggerr(rwrDrtyperr0r4rrr{rr r!rr,r&rsD    ================333333RRRRRRRR******  8 $ $'042 Vg g g g g /g g g g T 33S3#33333"F3F49FFFFs,      #      r,