o SiD@sdZddlZddlZddlZddlZddlZddlZddlZddlm Z m Z ddl m Z ddl mZddlZddlZddlmZddlmZdd lmZdd lmZmZd d Zd dZd2ddZd2ddZddZddZedfddZ GdddeZ!ej"fddZ#ee#Z$ddZ%e Gd d!d!Z&e'Z(d"d#Z)Gd$d%d%eZ*e*Z+d&d'Z,d(d)Z-d*d+Z.Gd,d-d-eZ/Gd.d/d/eZ0d0d1Z1dS)3zdTrain PyTorch models directly from POSIX tar archive. Code works locally or over HTTP connections. N) dataclassfield)islice)List)utils)pipelinefilter)IterableDataset) is_iterableobsoletecCs4|d}d|}|tjvsJd|tj|S)zLook up match in the environment with prefix WDS_. Args: m: A match object. Returns: str: The value of the environment variable WDS_. Raises: AssertionError: If the environment variable is not found. rWDS_z!missing environment variable wds_)grouposenviron)mkeyrI/home/ubuntu/.local/lib/python3.10/site-packages/webdataset/shardlists.py envlookup!s  rcCstdt|S)zSubstitute ${var} with the value of the environment variable WDS_var. Args: s (str): String to be substituted. Returns: str: The substituted string. z \$\{(\w+)\})resubrsrrrenvsubst3 rccsDtj|d\}}}}|dkrt||d|EdHdS|EdHdS)zSplit the input sequence by PyTorch distributed rank. Args: src: The input sequence to be split. group: The process group for distributed training. Yields: Elements from the input sequence based on the node's rank. r rNrpytorch_worker_inforsrcr rank world_sizeworker num_workersrrr split_by_node?s  r$ccs4tj|d\}}}}|dkrtd|EdHdS)a'Ensure the input sequence is not split for multi-node training. Args: src: The input sequence. group: The process group for distributed training. Yields: Elements from the input sequence. Raises: ValueError: If multi-node training is detected. rrzWyou need to add an explicit nodesplitter to your input pipeline for multi-node trainingN)rr ValueErrorrrrrsingle_node_onlyPs  r&ccs@t\}}}}|dkrt||d|EdHdS|EdHdS)zSplit the input sequence by PyTorch DataLoader worker. Args: src: The input sequence to be split. Yields: Elements from the input sequence based on the worker's ID. rNr)rr r!r"r#rrrsplit_by_workercs  r'cCsR|d}g}|D]}tdD]}|}t|}||krnq|t|q |S)aQExpand the urls if they are a string. If input is a string: - split on '::' - expand environment variables (using WDS_ prefix) - expand braces Otherwise: - return the input as a list Args: urls (str or List[str]): URL list or URL string. Returns: List[str]: List of expanded URLs. z:: )splitrangerextend braceexpand)urlsurllistresulturlilastrrr expand_urlsss  r3geAcCsHt|tr t|St|tr|St|rtt||Stdt|)a>Expand the given source into a list of URLs. Args: source (str or List[str] or Iterable): The source to be expanded. max_urls (int): Maximum number of URLs to return. Returns: List[str]: List of expanded URLs. Raises: ValueError: If the source type is not supported. zcannot handle ) isinstancestrr3listr rr%type)sourcemax_urlsrrr expand_sources  r:cs2eZdZdZd fdd ZddZddZZS) SimpleShardListz,An iterable dataset yielding a list of URLs.NcsZtt|trt|}nt|}||_t|jdts J|dur(t}||_dS)a-Initialize the SimpleShardList. Args: urls (str or List[str]): A list of URLs as a Python list or brace notation string. seed (int or bool or None): Random seed for shuffling; if None, no shuffling is done, if True, a random seed is generated. rTN) super__init__r4r5r3r6r-timeseed)selfr-r? __class__rrr=s    zSimpleShardList.__init__cCs t|jS)zfReturn the number of URLs in the list. Returns: int: The number of URLs. )lenr-)r@rrr__len__s zSimpleShardList.__len__ccsB|j}|jdurt|j||D]}t|dVqdS)~Return an iterator over the shards. Yields: dict: A dictionary containing the URL of each shard. Nr0)r-copyr?randomRandomshuffledict)r@r-r0rrr__iter__s  zSimpleShardList.__iter__N)__name__ __module__ __qualname____doc__r=rDrL __classcell__rrrArr;s r;c csddl}t}z tddd}Wnty2}ztt|ddtjdWYd}~nd}~ww| |}tdtjdt |}td t |d |tjdt |D]}| |VqVdS) zResample items from the source with replacement. Args: src: The source iterable. n (int): The number of items to yield. Defaults to sys.maxsize. Yields: Randomly chosen items from the source. rNz /dev/randomrb2filez# resampled loadingz# resampled got z samples, yielding )rHr>openread ExceptionprintreprsysstderrrIr6rCr*choice)rnrHr?exnrngitems_rrr resampled_s  &  reccs2d}|D] }|V|d7}q|dkrtddS)zEnsure the source yields at least one item. Args: src: The source iterable. Yields: Items from the source. Raises: ValueError: If the source yields no items. rrzHpipeline stage received no data at all and this was declared as an errorN)r%)rcountrrrr non_emptys  rgc@sLeZdZUdZdZeed<dZeed<dZ e ed<e e dZ eeed <d S) MSSourcez!Class representing a data source.nameperepochFresample)default_factoryr-N)rNrOrPrQrjr5__annotations__rlintrmboolrr6r-rrrrrrhs    rhcCstjtj|S)zExpand user and environment variables in a string. Args: s (str): The string to expand. Returns: str: The expanded string. )rpath expanduser expandvarsrrrrexpands rucs<eZdZdZejdddeddffdd Zd d ZZ S) ResampledShardsztime_nsurandomrgetr[rHrIrbr*rwrandintrCr-rK)r@r?rdindexrrrrLBs& zResampledShards.__iter__) rNrOrPrQr]maxsizerpr=rLrRrrrArrvs$rvcCs(z t|dWdStyYdSw)zCheck for the existence of a Unix PID. Args: pid (int): The process ID to check. Returns: bool: True if the process is running, False otherwise. rFT)rkillOSError)pidrrrcheck_pid_is_running_s   rcCstdd|S)zRemove the last extension from a filename. Args: fname (str): The filename to process. Returns: str: The filename without the last extension. z\.[^.]*$ri)rr)fnamerrrwithout_last_extensionprrcCs"td|}|s dSt|dS)zGet the PID from a filename. Args: fname (str): The filename to process. Returns: int or None: The PID if found in the filename, None otherwise. z^(.*)\._(\d+)_$N)rmatchrpr )rrrrrget_pid_from_filename|s rc@s>eZdZdZ      ddd Zd d Zd d ZddZdS)DirectoryShardListz8An iterable dataset that yields shards from a directory.*.{tar,tgz,tar.tgz}rmBrmrHNcCsL|dsJtj|sJ||_||_||_||_||_||_||_ dS)aInitialize the DirectoryShardList. Args: path (str): The directory path to monitor for shards. pattern (str): The glob pattern to match shard files. poll (int): The polling interval in seconds. timeout (float): The maximum time to wait for new shards. mode (str): The mode for handling processed shards. select (str): The strategy for selecting shards. fate (Any): Currently unused parameter. /N) endswithrrrisdirpollpatternmodeselectfatetimeout)r@rrrrrrrrrrrr=s zDirectoryShardList.__init__cCs\|jdkr t|dS|jdkrt|t|ddS|jdkr,t|t|dSdS)zRecycle a processed shard based on the current mode. Args: activename (str): The name of the active shard file. unlinkkeepz._done_rmN)rrrrenamer)r@ activenamerrrrecycles   zDirectoryShardList.recyclecCsDttj|jdD]}t|}|durq t|s||q dS)zs z-DirectoryShardList.__iter__..)rrHzunknown selection strategy z._rdrF)r>rsortedrrrrrCrsleeprminrHr_r%rrrFileNotFoundErrorrKrr)r@r2 candidates candidaterrrrrLs2          zDirectoryShardList.__iter__)rrrrmrHN)rNrOrPrQr=rrrLrrrrrs  rc@sBeZdZdZeddddZddZdd Zd d Zd d Z dS)MultiShardSamplez=An iterable dataset that samples from multiple shard sources.z6this is going to be replaced with the WIDS JSON format)reasoncCsd|_||dS)zInitialize the MultiShardSample. Args: fname (str or dict): The filename of the YAML spec or a dictionary containing the spec. rkN)r{ parse_spec)r@rrrrr=szMultiShardSample.__init__c st|_t|tr |}d}nt| }t|}Wdn1s!wYt| td s;Jt |t | ddg|_|dD]}t| td saJt || d| dg}t|trs|g}d d |D}|gkrdg}t|d ksJ|d |d | dd}|d}t|tr|g}fdd |D}| dd}| dd} | t|krtd| d| d kr|d krtdt||| |d} |j| td|dt|d| tjdqJdS)zParse the specification for multiple shard sources. Args: fname (str or dict): The filename of the YAML spec or a dictionary containing the spec. z{dict}Nzprefix datasets bucketsprefixridatasetsz#buckets name shards resample choosebucketscSsg|]}t|qSr)ru).0rrrr sz/MultiShardSample.parse_spec..rz2: FIXME support for multiple buckets unimplementedrrj@shardscs2g|]}tt|D] }tj|q qSr)r,rurrrr)rr0ubucketrrrrs2rmrkchoosez perepoch z- must be no greater than the number of shardsz&specify only one of perepoch or choose)rjr-rlrmz#  rV) default_rngrbr4rKrXyaml safe_loadsetkeysissubsetr)r6rursourcesr5rCr%rhappendr[r]r^) r@rspecstreamdsrrjr-rmnsampleentryrrrrsF   * *      &zMultiShardSample.parse_speccCst||_dS)zSet the current epoch for consistent shard selection among nodes. Args: seed (int): The seed for the random number generator. N)rHrIrb)r@r?rrr set_epochszMultiShardSample.set_epochcCsg}|jD]3}|jdkr|jj|j|jd}n|jdkr/t|j}|j||d|j}nt|j}||7}q|j||S)zGet the list of shards for the current epoch. Returns: list: A list of shard URLs for the current epoch. r)kN)rrmrbchoicesr-rlr6rJ)r@r/r8lrrrget_shards_for_epoch%s        z%MultiShardSample.get_shards_for_epochccs$|}|D]}t|dVqdS)zIterate over the shards for the current epoch. Yields: dict: A dictionary containing the URL of each shard. rFN)rrK)r@rshardrrrrL;s zMultiShardSample.__iter__N) rNrOrPrQr r=rrrrLrrrrrs  * rcCs|dr t|St|S)zCreate a shard list based on the given specification. Args: spec (str): The specification for creating the shard list. Returns: IterableDataset: Either a MultiShardSample or a SimpleShardList based on the spec. z.yaml)rrr;)rrrr shardspecFs rrM)2rQrros.pathrHrr]r> dataclassesrr itertoolsrtypingrr,rrirfiltersrpytorchr r r rrr$r&r'r3rpr:r;rre resampledrgrhrIrrurvResampledShardListrrrrrrrrrrsP      +  A [ `