o wiG@sddlZddlZddlZddlmZddlmZddlZddlm Z m Z m Z m Z m Z ddl mZmZddlmZddlmZdd lmZmZGd d d Zd d ZGdddeeZGdddeeZGdddeeZdS)N)SimpleNamespace)urlparse) autodecodecachefilters shardlistsutils)pipelinefilterreraise_exception) DataPipeline) DataLoader) group_by_keystar_file_expanderc@seZdZejdfddZddZd/ddZdd Zd0d d Z d dZ e fddZ d d d de dddZ e fddZddZddZe dddZddZd d!Zd1d#d$Zd%d&Zd'd(Zd)d*Zd+d,Zd-d.Zd S)2FluidInterfaceTcCs|tj|||dS)aCreate batches of the given size. This method forwards to the filters.batched function. Args: batchsize (int): Target batch size. collation_fn (callable, optional): Function to collate samples into a batch. Defaults to filters.default_collation_fn. partial (bool, optional): Whether to return partial batches. Defaults to True. Returns: FluidInterface: Updated pipeline with batched filter. ) collation_fnpartialcomposerbatched)self batchsizerrrN/home/ubuntu/sommelier/.venv/lib/python3.10/site-packages/webdataset/compat.pyrszFluidInterface.batchedcC|tS)zTurn batched data back into unbatched data. This method forwards to the filters.unbatched function. Returns: FluidInterface: Updated pipeline with unbatched filter. )rr unbatchedrrrrr!zFluidInterface.unbatchedcCs|tj|ddS)azCreate lists of samples without collation. This method forwards to the filters.batched function with collation_fn set to None. Args: batchsize (int): Target list size. partial (bool, optional): Whether to return partial lists. Defaults to True. Returns: FluidInterface: Updated pipeline with listed filter. N)rrr)rrrrrrlisted+ zFluidInterface.listedcCr)zTurn listed data back into individual samples. This method forwards to the filters.unlisted function. Returns: FluidInterface: Updated pipeline with unlisted filter. )rrunlistedrrrrr 9rzFluidInterface.unlistedNcC|t|S)a7Log keys of samples passing through the pipeline. This method forwards to the filters.log_keys function. Args: logfile (str, optional): Path to the log file. If None, logging is disabled. Returns: FluidInterface: Updated pipeline with log_keys filter. )rrlog_keys)rlogfilerrrr"C zFluidInterface.log_keyscKs$|dkr|S|tj|fi|S)abShuffle the data in the stream. This method forwards to the filters.shuffle function if size > 0. Args: size (int): Buffer size for shuffling. **kw: Additional keyword arguments for filters.shuffle. Returns: FluidInterface: Updated pipeline with shuffle filter, or self if size < 1. r)rrshuffle)rsizekwrrrr%Ps zFluidInterface.shufflecCs|tj||dS)aiApply a function to each sample in the stream. This method forwards to the filters.map function. Args: f (callable): Function to apply to each sample. handler (callable, optional): Exception handler. Defaults to reraise_exception. Returns: FluidInterface: Updated pipeline with map filter. handler)rrmap)rfr)rrrr*arzFluidInterface.mapF)prepostonlyrr)c Gs0dd|D}tj|||||d}|j||dS)aDecode data based on the decoding functions given as arguments. This method creates a decoder using autodecode.Decoder and applies it using filters.map. Args: *args: Decoding functions or strings representing image handlers. pre (callable, optional): Pre-processing function. post (callable, optional): Post-processing function. only (list, optional): List of keys to decode. partial (bool, optional): Whether to allow partial decoding. Defaults to False. handler (callable, optional): Exception handler. Defaults to reraise_exception. Returns: FluidInterface: Updated pipeline with decode filter. cSs$g|]}t|trt|n|qSr) isinstancestrr ImageHandler).0xrrr s$z)FluidInterface.decode..)r,r-r.rr()rDecoderr*) rr,r-r.rr)argshandlersdecoderrrrdecodeoszFluidInterface.decodecKs|tjdd|i|S)a}Map the entries in a dict sample with individual functions. This method forwards to the filters.map_dict function. Args: handler (callable, optional): Exception handler. Defaults to reraise_exception. **kw: Mapping of keys to functions to apply. Returns: FluidInterface: Updated pipeline with map_dict filter. r)Nr)rrmap_dict)rr)r'rrrr:s zFluidInterface.map_dictcKs|tj|fi|S)abSelect samples based on a predicate. This method forwards to the filters.select function. Args: predicate (callable): Function that returns True for samples to keep. **kw: Additional keyword arguments for filters.select. Returns: FluidInterface: Updated pipeline with select filter. )rrselect)r predicater'rrrr;s zFluidInterface.selectcO|tj|i|S)aCConvert dict samples to tuples. This method forwards to the filters.to_tuple function. Args: *args: Keys to extract from the dict. **kw: Additional keyword arguments for filters.to_tuple. Returns: FluidInterface: Updated pipeline with to_tuple filter. )rrto_tuplerr6r'rrrr> zFluidInterface.to_tupler(cGs|tj|d|iS)aMap the entries of a tuple with individual functions. This method forwards to the filters.map_tuple function. Args: *args: Functions to apply to each element of the tuple. handler (callable, optional): Exception handler. Defaults to reraise_exception. Returns: FluidInterface: Updated pipeline with map_tuple filter. r))rr map_tuple)rr)r6rrrrAr@zFluidInterface.map_tuplecGs|tj|S)zSlice the data stream. This method forwards to the filters.slice function. Args: *args: Arguments for slicing (start, stop, step). Returns: FluidInterface: Updated pipeline with slice filter. )rrslicerr6rrrrBr$zFluidInterface.slicecKs|tjdi|S)aRename samples based on keyword arguments. This method forwards to the filters.rename function. Args: **kw: Mapping of old names to new names. Returns: FluidInterface: Updated pipeline with rename filter. Nr)rrrename)rr'rrrrDs zFluidInterface.rename?cCr!)a%Randomly subsample a stream of data. This method forwards to the filters.rsample function. Args: p (float, optional): Probability of keeping each sample. Defaults to 0.5. Returns: FluidInterface: Updated pipeline with rsample filter. )rrrsample)rprrrrFr$zFluidInterface.rsamplecOr=)aZRename keys in samples based on patterns. This method forwards to the filters.rename_keys function. Args: *args: Positional arguments for filters.rename_keys. **kw: Keyword arguments for filters.rename_keys. Returns: FluidInterface: Updated pipeline with rename_keys filter. )rr rename_keysr?rrrrHr@zFluidInterface.rename_keyscOr=)aQExtract specific keys from samples. This method forwards to the filters.extract_keys function. Args: *args: Keys or patterns to extract. **kw: Additional keyword arguments for filters.extract_keys. Returns: FluidInterface: Updated pipeline with extract_keys filter. )rr extract_keysr?rrrrIr@zFluidInterface.extract_keyscOr=)aFDecode data based on file extensions. This method forwards to the filters.xdecode function. Args: *args: Positional arguments for filters.xdecode. **kw: Keyword arguments for filters.xdecode. Returns: FluidInterface: Updated pipeline with xdecode filter. )rrxdecoder?rrrrJr@zFluidInterface.xdecodecCr)zCache samples in memory. This method forwards to the filters.Cached class. Returns: FluidInterface: Updated pipeline with memory caching. )rrCachedrrrrmcachedrzFluidInterface.mcachedcOr=)a>Cache samples using LMDB. This method forwards to the filters.LMDBCached class. Args: *args: Positional arguments for filters.LMDBCached. **kw: Keyword arguments for filters.LMDBCached. Returns: FluidInterface: Updated pipeline with LMDB caching. )rr LMDBCachedr?rrr lmdb_cachedr@zFluidInterface.lmdb_cached)TN)rE)__name__ __module__ __qualname__rdefault_collation_fnrrrr r"r%r r*r9r:r;r>rArBrDrFrHrIrJrLrNrrrrrs4       rccs2d}|D] }|V|d7}q|dkrtddS)zCheck if the dataset is empty and yield samples. Args: source: An iterable source of samples. Yields: The samples from the source. Raises: ValueError: If no samples are found in the dataset. rrzNo samples found in dataset; perhaps you have fewer shards than workers. Turn off using empty_check=False in the WebDataset constructor.N) ValueError)sourcecountsamplerrr check_empty-s  rXcsheZdZdZeddddddejdejej dddddffdd Z dd Z d d Z d d Z ddZZS) WebDatasetaZCreate a WebDataset pipeline for efficient data loading. This class sets up a data pipeline for loading and processing WebDataset-format data. It handles URL generation, shard shuffling, caching, and sample grouping. Args: urls: The source URLs or specifications for the dataset. handler: Function to handle exceptions. Defaults to reraise_exception. mode: The mode of operation. Defaults to None. resampled: Whether to use resampled mode. Defaults to False. repeat: Whether to repeat the dataset. Defaults to False. shardshuffle: The number of shards to shuffle, or None. Defaults to None. cache_size: The size of the cache in bytes. Defaults to -1 (unlimited). cache_dir: The directory to use for caching. Defaults to None. url_to_name: Function to convert URLs to cache names. Defaults to pipe_cleaner. detshuffle: Whether to use deterministic shuffling. Defaults to False. nodesplitter: Function to split data by node. Defaults to single_node_only. workersplitter: Function to split data by worker. Defaults to split_by_worker. select_files: Function to select files from tar archives. Defaults to None. rename_files: Function to rename files from tar archives. Defaults to None. empty_check: Whether to check for empty datasets. Defaults to True. verbose: Whether to print verbose output. Defaults to False. seed: Random seed for shuffling. Defaults to None. Raises: ValueError: If the cache directory does not exist or if the URL type is not supported. NFTcst|r d}|dkr|dvrtdn |dur td|dur+tdd}tdit}|durBtjdt d d n||_ | || || durX|| | r_|| |jdur|jrt|tj|j|j d n |tj|j|j d |dus|d krtj|d }ntj|||d }||tt}|||| |dtt}|||d |r|tdSdS)N resampled)FNz>WebDataset(shardshuffle=...) is ignored for resampled datasetszIWebDataset(shardshuffle=...) is None; set explicitly to False or a numberTzDset WebDataset(shardshuffle=...) to a positive integer or 0 or FalsedWDS_SEEDri@B)seedr() cache_dir cache_sizer))r) select_files rename_filesr)super__init__warningswarnrlocalsosenvirongetrandomrandintr^update_cache_infocreate_url_iteratorappend shardshuffle detshufflerr%r StreamingOpen FileCacher rrrX)rurlsr)moder[repeatrpr`r_ url_to_namerq nodesplitterworkersplitterrarb empty_checkverboser^r6openerexpandergrouper __class__rrrdasB    $      zWebDataset.__init__cCsjttjd|j|_tjd|j|_|jdur1tj|j|_tj|js3t d|jddSdS)zUpdate cache information based on arguments and environment variables. Args: args: A SimpleNamespace object containing the arguments. Raises: ValueError: If the specified cache directory does not exist. WDS_CACHE_SIZE WDS_CACHENzcache directory z does not exist) intrhrirjr`r_path expanduserexistsrTrCrrrrms  zWebDataset.update_cache_infocCsZ|j}t|tr<|ds|drs&    9