o }oi~/ @sddlZddlZddlZddlZddlmZddlmZmZm Z m Z m Z ddl m Z zddlmZWn ey=dZYnwddlmZddlmZdd lmZz dd lmZd ZWn eyed ZYnwd ejfddZd efddZd efddZd efddZ d e!fddZ"d e!fddZ#de!d e e!e!ffddZ$de!d e!fddZ%ed d!d e!fd"d#Z&d$e!d e!fd%d&Z'd=d(e!d)e(fd*d+Z)d>d(e!d-e!fd.d/Zd?d(e!d0ed)e(d e!fd1d2Z*Gd3d4d4Z+d5e+d efd6d7Z,d8e e e!efd9ee-gefd:e e!effd;d<Z.dS)@N) lru_cache)AnyCallableDictIterableTuple)urlparse) __version__git) constants)logging)LogMode) open_bestTFreturncCsHtjtjd}|dkrtjtjdt }|St| }|S)a Utility method to resolve a cache directory for NeMo that can be overriden by an environment variable. Example: NEMO_CACHE_DIR="~/nemo_cache_dir/" python nemo_example_script.py Returns: A Path object, resolved to the absolute path of the cache directory. If no override is provided, uses an inbuilt default which adapts to nemo versions strings. z.cache/torch/NeMo/NeMo_) osenvirongetr NEMO_ENV_CACHE_DIRpathlibPathjoinpathhome NEMO_VERSIONresolve) override_dirpathrI/home/ubuntu/.local/lib/python3.10/site-packages/nemo/utils/data_utils.pyresolve_cache_dir(s  rcCs4zt|}t|jot|jWStyYdSw)z,Check if a path is from a data object store.F)rboolschemenetlocAttributeError)rresultrrris_datastore_path;s  r%cCs |dS)z%Check if a path is for a tarred file.z.tar)endswith)rrrris_tarred_pathD r'cCs<ttjtjd}|dkrdS|dkrdStdtj)zCheck if store cache is shared.rFTzUnexpected value of env )intrrrr NEMO_ENV_DATA_STORE_CACHE_SHARED ValueError) cache_sharedrrris_datastore_cache_sharedIs r.cCsZtjtjd}|dkrt}n t| }| t r&tj |}tj |dS)z#Return path to local cache for AIS.rais)rrrr NEMO_ENV_DATA_STORE_CACHE_DIRras_posixrrrr&rrdirnamejoin)r cache_dirrrrais_cache_baseVs   r5cCs tdS)zGet configured AIS endpoint. AIS_ENDPOINT)rgetenvrrrr ais_endpointdr(r8uricCsLt|s td|t|j}|d}tj|dd}t|t|fS)zParse a path to determine bucket and object path. Args: uri: Full path to an object on an object store Returns: Tuple of strings (bucket_name, object_path) z(Provided URI is not a valid store path: r)N)r%r,rPurePathpartsstr)r9 uri_partsbucket object_pathrrrbucket_and_object_from_uriis   rAendpointcCs8t|}|jr |jstd|tj|jt|jS)zConvert AIS endpoint to a valid dir name. Used to build cache location. Args: endpoint: AIStore endpoint in format https://host:port Returns: Directory formed as `host/port`. z$Unexpected format for ais endpoint: )rhostnameportr,rrr3r=)rBr$rrrais_endpoint_to_dir{s rEr))maxsizecCsdtd}|durtd||Sd}tj|r$tjd|tj d|Stj d|dtj ddS) z-Return location of `ais` binary if available.r/NzFound AIS binary at %sz/usr/local/bin/aisz%ais available at the default path: %smodez>AIS binary not found with `which ais` and at the default path .) shutilwhichr debugrrisfileinfor ONCEwarning)r default_pathrrr ais_binarys   rR store_pathcCsbt|r*t}|std|tjtt|}t|\}}tj|||}|St d|)zConvert a data store path to a path in a local cache. Args: store_path: a path to an object on an object store Returns: Path to the same object in local cache. %AIS endpoint not set, cannot resolve zUnexpected store path format: ) r%r8 RuntimeErrorrrr3r5rErAr,)rSrBlocal_ais_cache store_bucket store_object local_pathrrrdatastore_path_to_local_paths  rZr num_retriesc Cst|rdt}|durtd|t}|std|d|d|d}d}t|D]}tj|dtjtjdd }|j}| d rGd}nq-|sb|j j d d d  } t|d|d| |SdS)a!Open a datastore object and return a file-like object. Args: path: path to an object num_retries: number of retries if the get command fails with ais binary, as AIS Python SDK has its own retry mechanism Returns: File-like object that supports read() NrTz(AIS binary is not found, cannot resolve a?. Please either install it or install Lhotse with `pip install lhotse`. Lhotse's native open_best supports AIS Python SDK, which is the recommended way to operate with the data from AIStore. See AIS binary installation instructions at https://github.com/NVIDIA/aistore?tab=readme-ov-file#install-from-release-binaries. z get z -FT)shellstdoutstderrtextr)zutf-8ignore)errorsz* couldn't be opened with AIS binary after z. attempts because of the following exception: )r%r8rUrRrange subprocessPopenPIPEr^peekr_readdecodestripr,) rr\rBbinarycmddone_procstreamerrorrrr!open_datastore_object_with_binarys6    rrrbrHcCs,trt||dSt|rt|St||dS)NrG)LHOTSE_AVAILABLElhotse_open_bestr%rropen)rrHrrrrs   rforcecCst|rGt|d}tj|r|rEtj|}tj|s$tj|ddt|d}|j t | |dWd|S1s@wY|S|S)aDownload an object from a store path and return the local path. If the input `path` is a local path, then nothing will be done, and the original path will be returned. Args: path: path to an object force: force download, even if a local file exists num_retries: number of retries if the get command fails with ais binary, as AIS Python SDK has its own retry mechanism Returns: Local path of the object. )rST)exist_okwb)r\N) r%rZrrrMr2isdirmakedirsrvwriterrh)rrwr\rY local_dirfrrrget_datastore_objects     rc@s|eZdZdZddededefddZed efd d Zed efd d Z dded efddZ dded efddZ ddZ dS)DataStoreObjecta'A simple class for handling objects in a data store. Currently, this class supports objects on AIStore. Args: store_path: path to a store object local_path: path to a local object, may be used to upload local object to store get: get the object from a store NFrSrYrcCs0|durtd||_||_|r|dSdS)Nz3Specifying a local path is currently not supported.)NotImplementedError _store_path _local_pathr)selfrSrYrrrr__init__s zDataStoreObject.__init__rcC|jS)z Return store path of the object.)rrrrrrS$zDataStoreObject.store_pathcCr)z Return local path of the object.)rrrrrrY)rzDataStoreObject.local_pathrwcCs|js t|j|d|_|jS)zGet an object from the store to local cache and return the local path. Args: force: force download, even if a local file exists Returns: Path to a local object. )rw)rYrrSrrrwrrrr.s zDataStoreObject.getcCst)zPush to remote and return the store path Args: force: force download, even if a local file exists Returns: Path to a (remote) object object on the object store. )rrrrrput<s zDataStoreObject.putcCs t|d|jd|j}|S)z2Return a human-readable description of the object.z : store_path=z , local_path=)typerSrY)r descriptionrrr__str__GszDataStoreObject.__str__)NF)F) __name__ __module__ __qualname____doc__r=r rpropertyrSrYrrrrrrrrs   rrXcCs |duS)zA convenience wrapper for multiprocessing.imap. Args: store_object: An instance of DataStoreObject Returns: True if get() returned a path. N)r)rXrrrdatastore_object_getMs rdatahandlerkwc ks|D]D}t|tsJ|d|vsJ|d}zt|dd}|j|d|VWqtyG}z||r;WYd}~qWYd}~dSd}~wwdS)a Open URLs and yield a stream of url+stream pairs. This is a workaround to use lhotse's open_best instead of webdataset's default url_opener. webdataset's default url_opener uses gopen, which does not support opening datastore paths. Args: data: Iterator over dict(url=...). handler: Exception handler. **kw: Keyword arguments for gopen.gopen. Yields: A stream of url+stream pairs. urlrsrG)rpN) isinstancedictrupdate Exception)rrrsamplerrpexnrrrwds_url_openerYs      r)r[)rs)Fr[)/rrrJrd functoolsrtypingrrrrr urllib.parsernemor r ImportErrorr nemo.utilsr nemo.utils.nemo_loggingr lhotse.serializationrrurtrrr r%r'r.r=r5r8rArErRrZr*rrrrrrrrrrrsX          /"=