o bi1 @sddlZddlZddlZddlZddlZddlmZddlmZm Z m Z m Z m Z m Z mZddlmZddlmZddlmZerDddlZdefdd Zd e ee ffd d Z d.d ee e e fd e ee fde fddZdedeeeffddZ    d/de de eeeefde eeeefdedef ddZeddfde e ee fdede eeefdefdd Z d0d e ee fd"e eeeffd#d$Z!d%d&iZ"d0d e ee fd"e eeeffd'd(Z#d)efd*d+Z$Gd,d-d-eZ%dS)1N)partial) TYPE_CHECKINGAnyCallableDictListOptionalUnion)iterate_with_retry) BlockAccessor)FileBasedDatasourcepathcCs(td|}|s dS|d|dfS)zSplit off all file extensions. Returns base, allext. Args: path: path with extensions Returns: str: path with all extensions removed z^((?:.*/|)[^.]+)[.]([^/]*)$)NN)rematchgroup)r rrg/home/ubuntu/.local/lib/python3.10/site-packages/ray/data/_internal/datasource/webdataset_datasource.py_base_plus_exts rsamplecCs4|duot|tott|dko|dd S)zUCheck whether a sample is valid. Args: sample: sample to be checked Nr__bad__F) isinstancedictlenlistkeysget)rrrr _valid_sample%s rfdefaultcCsN|dur|St|ts|g}|D]}|dur t|s t||d}||}q|S)aApply a list of functions to a sample. Args: f: function or list of functions sample: sample to be modified default: default function to be applied to all keys. Defaults to None. Returns: modified sample Nformat)rrcallabler)rrr grrr _apply_list3s   r%suffixsuffixescCsl|durdSt|r||S|D]#}d|vsd|vr&td||r%dSq||ks0d||kr3dSqdS)aCCheck whether a suffix is valid. Suffixes can be either None (=accept everything), a callable, or a list of patterns. If the pattern contains */? it is treated as a glob pattern, otherwise it is treated as a literal. Args: suffix: suffix to be checked suffixes: list of valid suffixes NT*?.F)r#fnmatch)r&r'patternrrr _check_suffixLs r-Ffileobj fileselect filerename verbose_openmetac cs|pi}tj|dd}|rtd||D]0}|j}|r$|dur%q||}t||}t|t s8Jt ||s>qt ||d} | Vq|rStd|dSdS)aIterate over tar file, yielding filename, content pairs for the given tar stream. Args: fileobj: file object fileselect: patterns or function selecting files to be selected meta: metadata to be added to each sample zr|*)r.modezstart N)fnamedatazdone ) tarfileopenprintnameisreg extractfilereadr%rstrr-r) r.r/r0r1r2streamtarinfor4r5resultrrr_tar_file_iteratords&   rAr5rc cs|pi}d}|D]h}t|tsJ|d|d}}||\}} |dur&q |dus0||dkrKt|r<|||Vt|d}d|vrK|d|d<| |vrdt|d| d|d |d|dusmt| |rq||| <q t|r|||VdSdS) a7Return function over iterator that groups key, value pairs into samples. Args: data: iterator over key, value pairs keys: function that returns key, suffix for a given key suffixes: list of suffixes to be included in the sample meta: metadata to be added to each sample Nr4r5__key__)rB__url__z": duplicate file name in tar file  z , tar is )rrrupdate ValueErrorrr-) r5rr'r2current_sample filesampler4valueprefixr&rrr_group_by_keyss:      rKTr"c Csht|}|D]\}}|dd}|drq|dvr%|d||<q|dvr3t|d||<q|dvr^dd l}dd l}|d krO|j t |||<q| |j t |||<q|d krjt |||<q|d kr}dd l}|t |||<q|d krdd l}|j|dd||<q|dvrdd l}|t |||<q|dvrdd l} | |||<q|S)aTA default decoder for webdataset. This handles common file extensions: .txt, .cls, .cls2, .jpg, .png, .json, .npy, .mp, .pt, .pth, .pickle, .pkl. These are the most common extensions used in webdataset. For other extensions, users can provide their own decoder. Args: sample: sample, modified in place r*__)txttextutf-8clscls2)jpgpngppmpgmpbmpnmrNPILjsonnpympF)rawptpthpicklepkl)ritemssplit startswithdecodeintnumpy PIL.ImageImager7ioBytesIOasarrayr[loadsloadmsgpackunpackbtorchrc) rr"keyrI extensionnprZrrrtrcrrr_default_decoders>  rxrTjpegc Cst|}|D]\}}|dd}|drq|dvr%|d||<q|dvr3t|d||<q|dvrldd l}dd l}t||j rK|j |}t||j j sTJt }|j|t||d |||<q|d kr{t|d||<q|d krdd l}t }||||||<q|d krdd l}||||<q|dvrdd l} t }| |||||<q|dvrdd l} t }| |||||<q|S)aQA default encoder for webdataset. This handles common file extensions: .txt, .cls, .cls2, .jpg, .png, .json, .npy, .mp, .pt, .pth, .pickle, .pkl These are the most common extensions used in webdataset. For other extensions, users can provide their own encoder. Args: sample (Dict[str, Any]): sample r*rLrM)rNrPrQ)rTryrUrVrWrXrYrNr!r[r\r]r_rb)rrerfrgencoder=rjrkrndarrayrl fromarrayrmrnsaveextension_to_formatrlowergetvaluer[dumpsrrrtrcdump) rr"rurIrvrwrZr>rrrtrcrrr_default_encodersT        rblockcCs |jddS)zMake a block iterable. This is a placeholder for dealing with more complex blocks. Args: block: Ray Dataset block Returns: Iterable[Dict[str,Any]]: Iterable of samples F)public_row_format) iter_rows)rrrr_make_iterables rcseZdZdZdgZ      ddeeeefdeee ee e fdeee e e fd eee e e fd eee e e fd e d e ffd d Z dddefddZ ZS)WebDatasetDatasourcezJA Datasource for WebDataset datasets (tar format with naming conventions).tarTNFpathsdecoderr/r0r'r1 expand_jsonc  s<tj|fi|||_||_||_||_||_||_dSN)super__init__rr/r0r'r1r) selfrrr/r0r'r1rfile_based_datasource_kwargs __class__rrr1s  zWebDatasetDatasource.__init__r>zpyarrow.NativeFiler c #s*ddl}fdd}t|djjd}t|t|djd}|D]p}jdur1tj|t d }j rt |d t rFt |d d }n't |d trUt |d }nt |d tra|d }n td t|d d |D]\} } | |vr}g|| <|| | qq|dd|DVq"dS)aRead and decode samples from a stream. Note that fileselect selects files during reading, while suffixes selects files during the grouping step. Args: stream: File descriptor to read from. path: Path to the data. decoder: decoder or list of decoders to be applied to samples fileselect: Predicate for skipping files in tar decoder. Defaults to lambda_:False. suffixes: List of suffixes to be extracted. Defaults to None. verbose_open: Print message when opening files. Defaults to False. Yields: List[Dict[str, Any]]: List of sample (list of length 1). rNcstjjjdS)N)r/r0r1)rAr/r0r1rrr>rrget_tar_file_iteratorZs z@WebDatasetDatasource._read_stream..get_tar_file_iteratorziterate tar file)r)rC)r2r')r r[rPzUnsupported data type z for samplecSs2i|]\}}|t|trt|dkr|n|gqS)r)rrr).0kvrrr }s z5WebDatasetDatasource._read_stream..)pandasr _data_contextretried_io_errorsrKrr'rr%rxrrbytesr[rprhr= TypeErrortypereappend DataFrame) rr>r pdrfilessamplesr parsed_jsonrrrrr _read_streamEs@   z!WebDatasetDatasource._read_stream)TNNNFF)__name__ __module__ __qualname____doc___FILE_EXTENSIONSr r=rrboolr#rrr __classcell__rrrrr,s2rr)NNFN)T)&r+rmr[rr6 functoolsrtypingrrrrrrr ray.data._internal.utilr ray.data.blockr )ray.data.datasource.file_based_datasourcer pyarrowr=rrr%rr#r-rrrArKrxr~rrrrrrrsn $      $ ()1(;