o }oiG* @s:ddlZddlZddlZddlmZddlZddlmZddlm Z z4ddl m Z ddl mZmZmZddlmZddlmZddlmZdd lmZdd lmZmZd ZWneeefyid Ze d YnwdZ!erdddZ efddZ"ed dddfddZ#e$e#Z%GdddeeZ&dSGdddeZ&dS)N)urlparse)ApexGuardDefaults)logging)cachefilters shardlists)FluidInterface)reraise_exception) DataPipeline)IterableDataset) group_by_keystar_file_expanderTF@Webdataset import failed! We recommend use `webdataset==0.2.48`. rb c Ksttjdd}|rtd|tjtjd|dvsJ||dkr8|dkr)tj j S|dkr1tj j St d |d |vr|d rd}|t krz'|d j|d |d }|d}|dt|k}|rht|WS|d7}Wn!ty} z|d7}t| td|WYd} ~ qBd} ~ ww|t ksFtd||d|vr|ddurtj|d|}t|} | jdkrttjdd}t|||dS| jdkrttjdd}t| j||dStjd} tj| j| } | |||fi|S)aOpen the URL. This uses the `gopen_schemes` dispatch table to dispatch based on scheme. Support for the following schemes is built-in: pipe, file, http, https, sftp, ftps, scp. When no scheme is given the url is treated as a file. You can use the OPEN_VERBOSE argument to get info about files being opened. This implementation is based on webdataset's gopen, with the modification of supporting reading from s3 object_store: https://webdataset.github.io/webdataset/api/webdataset/gopen.html#gopen Args: url (list[str]): the source URL mode (str): the mode ("rb", "r") bufsize (int): the buffer size GOPEN_VERBOSErGOPEN)file)rwb-rrz unknown mode object_store s3_clients3_bucket_name)BucketKeyBody ContentLengthz&Retrying tar file download, attempt {}Nz/Unable to read {} from PBSS. {} attempts tried.local_root_path GOPEN_BUFFER) bufferingr __default__)intosenvirongetprint gopen_webdatainfosysstderrstdinbufferstdout ValueError_NUM_OBJECT_STORE_READ_ATTEMPTS get_objectreadlenioBytesIO ExceptionformatConnectionErrorpathjoinrschemeopen gopen_schemes) urlmodebufsizekwverboseattempts3_response_objectobject_content full_readeprhandlerrLi/home/ubuntu/.local/lib/python3.10/site-packages/nemo/collections/multimodal/data/common/webdataset_s3.pygopen.sR       rNc ks|D]M}t|tsJ|d|vsJ|d}zt|fi|}|j|d|VWqtyP}z|j|f|_||rDWYd}~qWYd}~dSd}~wwdS)zGiven a stream of url names (packaged in `dict(url=url)`), yield opened streams. Args: data: Iterator of dictionaires containing url paths. handler: Exception handler. r@)streamN) isinstancedictrNupdater8args)datarKrCsampler@rOexnrLrLrM url_openerys"    rWc Cs0t||||||d}t||d}t||d}|S)a Given an iterator of filenames, this function opens the URL streams and groups data by keys. Args: src: Iterator of data dictionaires containing URL names. handler: Exception handler. load_from_object_store (bool): A boolean flag to specify whether to load from object store. s3_client: If loading from object store, specify S3 client. s3_bucket_name: If loading from object store, specify S3 bucket name. local_root_path: If loading from local (or mounted) disk system, specify the root path of the dataset. )rKrrrr)rK)rWr r ) srcrKload_from_object_storerrrstreamsfilessamplesrLrLrMtarfile_sampless  r]c s<eZdZdZedddddejdddddf fdd ZZS) WebDatasetz?Webdataset class modified to support loading from object store.FNr"c stt|tr|rJ||nt|trK|ds#|drKt| }t |}Wdn1s7wYd|vsBJ|t |nOt|t r_d|vsVJ|t |n;|rj|t |n0|t ||||t j|durd}|dur|r|t|n|t||dus|dkr|t|| | | | ddS|d ks|dksJ|tj|| ||d dS) a Args: urls: An iterator containing a list of url names. handler: Exception handler. resampled: If true, sample shards from shard list with replacement. shardshuffle: If true, shuffles the entire shard list. cache_size: Size of cache. cache_dir: Path to store cache. detshuffle: Whether to use deterministic shuffling when shardshuffle is True. nodesplitter: Function for splitting urls among nodes. verbose: If True, prints logs. load_from_object_store (bool): A boolean flag to specify whether to load from object store. s3_client: If loading from object store, specify S3 client. s3_bucket_name: If loading from object store, specify S3 bucket name. local_root_path: If loading from local (or mounted) disk system, specify the root path of the dataset. z.yamlz.ymlNdatasetsTdr)rKrYrrrr")rKrD cache_size cache_dir)super__init__rPr appendstrendswithr>yaml safe_loadrMultiShardSamplerQResampledShardsSimpleShardListsplit_by_workerr detshuffleshuffletarfile_to_samplesrcached_tarfile_to_samples)selfurlsrK resampled shardshufflerarbrn nodesplitterrDrYrrrrOspec __class__rLrMrdsP "          WebDataset.__init__) __name__ __module__ __qualname____doc__r rsingle_node_onlyrd __classcell__rLrLrxrMr^sr^cseZdZfddZZS)r^csttddS)Nr)rcrdrwarning)rrrxrLrMrd s rz)r{r|r}rdrrLrLrxrMr^ s)rr)'r6r&r, urllib.parserrh2nemo.collections.nlp.modules.common.megatron.utilsr nemo.utilsrwebdataset.gopenrNr* webdatasetrrrwebdataset.compatrwebdataset.handlersr webdataset.pipeliner webdataset.pytorchr webdataset.tariteratorsr r HAVE_WEBDATASET ImportErrorAttributeErrorModuleNotFoundErrorrr2rWr]pipelinefilterrpr^rLrLrLrMsB           K "U