&`i1 ddlZddlZddlZddlZddlZddlmZddlmZm Z m Z m Z m Z m Z mZddlmZddlmZddlmZerddlZdefdZd e ee ffd Z d%d ee e e fd e ee fd e fd ZdedeeeffdZ d&de de eeeefde eeeefdedef dZeddfde e ee fdede eeefdefdZ d'd e ee fde eeeffdZ!ddiZ"d'd e ee fde eeeffd Z#d!efd"Z$Gd#d$eZ%dS)(N)partial) TYPE_CHECKINGAnyCallableDictListOptionalUnion)iterate_with_retry) BlockAccessor)FileBasedDatasourcepathctjd|}|sdS|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)rrs /home/jaya/work/projects/VOICE-AGENT/VIET/agent-env/lib/python3.11/site-packages/ray/data/_internal/datasource/webdataset_datasource.py_base_plus_extrsB H3T : :E z ;;q>>5;;q>> ))samplec|duo]t|toHtt|dko|dd S)zUCheck whether a sample is valid. Args: sample: sample to be checked Nr__bad__F) isinstancedictlenlistkeysget)rs r _valid_sampler!%sc d - vt $ $ - V[[]]## $ $q ( - 9e,, , rfdefaultc||St|ts|g}|D]/}| t|st||}||}0|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)r"rr#gs r _apply_listr)3sm y a   C   x{{ ***A6 Mrsuffixsuffixesc|dSt|r ||S|D]8}d|vsd|vrtjd|z|rdS&||ks d|z|krdS9dS)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+patterns r _check_suffixr2Lst x '>>SG^^sV|W55 tt  w  #,'"9"944#: 5rFfileobj fileselect filerename verbose_openmetac#K|pi}tj|d}|rtd||D]}|j}|r| ||}t||}t|tsJt||st||} | V|rtd|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|*)r3modezstart N)fnamedatazdone ) tarfileopenprintnameisreg extractfilereadr)rstrr2r) r3r4r5r6r7streamtarinfor:r;results r_tar_file_iteratorrGds :2D \' 6 6 6F otoo   }} %- !!'**//11J..%%%%%%UJ//  E---  ndnnrr;rc #.K|pi}d}|D]}t|tsJ|d|d}}||\}} |:| ||dkrGt|r|||Vt|}d|vr |d|d<| |vr5t |d| d|d |dz|t | |r||| <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 Nr:r;__key__)rI__url__z": duplicate file name in tar file  z , tar is )rrr!update ValueErrorrr2) r;rr+r7current_sample filesampler:valueprefixr*s r_group_by_keysrRs :2DN++ *d+++++!'*Jv,>ue >   !V~i/H%H%H^,, %%%d+++$$$$!&111NJ&&,6y,Ay) ^ # #<<<OOn1133OOd9oOOP   }VX>> %*N6 "^$$d###rTr&ct|}|D]\}}|dd}|dr7|dvr|d||<T|dvr&t |d||<~|dvrdd l}dd l}|d kr0|j tj |||<| |j tj |||<|d krtj|||<#|d kr0dd l}|tj |||<Y|d kr dd l}||d||<|dvr0dd l}|tj |||<|dvrdd l} | |||<|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.ImageImager=ioBytesIOasarrayrcloadsloadmsgpackunpackbtorchrk) rr&keyrP extensionnprbrzr|rks r_default_decoderrs &\\Fllnn!.!. UIIcNN2& >>$   .  / ) ),,w//F3KK / ) )ell73344F3KK D D D        !innRZ->->??s  jj 58I8I)J)JKKs & *U++F3KK %      ''"*U"3"344F3KK $   NNN!//%U/;;F3KK - ' ' LLL**RZ%6%677F3KK + + + MMM ,,u--F3K Mrr\jpegc>t|}|D]w\}}|dd}|dr7|dvr|d||<T|dvr&t |d||<~|dvrdd l}dd l}t||j r|j |}t||j j sJtj }||t|| |||<C|d kr,t'j|d||<u|d krFdd l}tj }||||||<|d krdd l}||||<|dvrFdd l} tj }| |||||</|dvrDdd l} tj }| |||||<y|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/rTrU)rVrXrY)r\rr]r^r_r`rarNr%rcrdrergrj)rrmrnroencoderCrrrsrndarrayrt fromarrayrurvsaveextension_to_formatr lowergetvaluercdumpsrzr|rkdump) rr&r}rPr~rrbrDrzr|rks r_default_encoderrs&\\Fllnn+,+, UIIcNN2& >>$  ) ,  ' ! !,,w//F3KK / ) )e**++G44F3KK L L L        %,, 3 ++E22eSY_55 5 55Z\\F JJ266y7H7H)TT    !//++F3KK & *U++227;;F3KK %      Z\\F GGFE " " " //++F3KK $   NNN!--..F3KK - ' ' LLLZ\\F JJuf % % % //++F3KK + + + MMMZ\\F KKv & & & //++F3K Mrblockc.|dS)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)rs r_make_iterablers ??U? 3 33rceZdZdZdgZ ddeeeefdeee ee e fdeee e e fd eee e e fd eee e e fd e d e ffd Z dddefdZ xZS)WebDatasetDatasourcezJA Datasource for WebDataset datasets (tar format with naming conventions).tarTNFpathsdecoderr4r5r+r6 expand_jsonc tj|fi|||_||_||_||_||_||_dSN)super__init__rr4r5r+r6r) selfrrr4r5r+r6rfile_based_datasource_kwargs __class__s rrzWebDatasetDatasource.__init__1sW ??">??? $$  (&rrDzpyarrow.NativeFilerc#|Kddl}fd}t|djj}t |t |j}|D]h}jtj|t}j rt|d tr.tj|d d }nt|d t rtj|d }nJt|d t r |d }n&t#d t%|d d |D])\} } | |vrg|| <|| | *|d |DVjdS)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). rNcHtjjjS)N)r4r5r6)rGr4r5r6)rrDsrget_tar_file_iteratorz@WebDatasetDatasource._read_stream..get_tar_file_iteratorZs-%??!.  rziterate tar file)r)rJ)r7r+)r#rcrXzUnsupported data type z for samplecpi|]3\}}|t|trt|dkr|n|g4S)r)rrr).0kvs r z5WebDatasetDatasource._read_stream..}sQ1Jq$//HCFFaKKqqaSr)pandasr _data_contextretried_io_errorsrRrr+rr)rrrbytesrcrxrprC TypeErrortypermappend DataFrame) rrDrpdrfilessamplesr parsed_jsonrrs `` r _read_streamz!WebDatasetDatasource._read_streamEs&       # ! $6    !T$-?-?-?$-XXX  F|'$T\6CSTTT (fVne44 "&*VF^-B-B7-K-K"L"LKKv44"&*VF^"<""!''S$s)^$'%c8T 9:;'U44#789 ' U44#789 ' 5x!567 ''''''''(<#7<s<<<<<<<<rrr)NNFN)T)&r0rurcrr< functoolsrtypingrrrrrr r ray.data._internal.utilr ray.data.blockr )ray.data.datasource.file_based_datasourcer pyarrowrCrr!r)rr'r2rrrGrRrrrrrrrrs8 LLLLLLLLLLLLLLLLLL666666((((((IIIIIINNN*****" $sCx.    UY XtH~ %&04S#XIQ2#tX~)>49=8<  tXt345tXt345   H$04 && tCH~ & &uT8^,-&  &&&&R..T#s(^.XeD#I>N5O....bfo88T#s(^8XeCI>N5O8888v 4- 4 4 4 4UUUUU.UUUUUr