o Si,@s2dZddlZddlZddlZddlZddlZddlZddlZddl m Z m Z m Z ddlm Z ddlmZddlmZddlmZejdd Zeejd d Zeejd d ZddZdefddZdefddZdefddZeddZ d%ddZ!GdddZ"d&dd Z#Gd!d"d"Z$Gd#d$d$Z%dS)'z`Code related to caching files downloaded from storage servers, object servers, and web servers. N)CallableIterableOptional)urlparse)reraise_exception)obsolete WDS_CACHEz./_cacheWDS_CACHE_SIZE1e18WDS_VERBOSE_CACHE0cCst|}|jdvS)z$Check whether a URL is a local file.file)rscheme)urlparsedrD/home/ubuntu/.local/lib/python3.10/site-packages/webdataset/cache.pyislocals rfnamec Csjtj|s J|z tjddgtjddWntjtfy%tdwtjd|gtj ddd}|j S)zGet the file type of a file.r.T)stdoutcheckz%UNIX/Linux file command not available)rtextr) ospathexists subprocessrunDEVNULLSubprocessErrorFileNotFoundErrorAssertionErrorPIPEr)rresultrrr get_filetype sr'cCst|dF}|d}t|dkr%|dddkr%|dWdSt|dkr?|d d d kr?|d WdS|d WdS1sNwYdS)zDetermine file type by checking magic numbers. It checks common formats used with WebDataset: tar archives and gzip files. Args: fname (str): Path to the file to check Returns: str: Description of the file type rbirsz: gzip compressed dataNiiisustarz: POSIX tar archivez: data)openreadlen)rfheaderrrrmagic_filetype-s  $r/cCs4tj|s J|t|}d|vpd|vS)z&Check whether a file is a tar archive.z tar archivezgzip compressed)rrrr/lower)rftyperrrcheck_tar_formatIsr2cCsB|dr|dd}|d}|D] }td|r|Sq|S)z2Guess the actual URL from a "pipe:" specification.zpipe:N z^(https?|hdfs|gs|ais|s3):) startswithsplitrematch)specwordswordrrr pipe_cleanerRs    r<cCspt|tsJt|}|jdvr(|j}|d}|d}d|d|dStj j |dd}|dd}|S)z Guess the cache name from a URL.) Nrrhttphttpsftpftpsgss3ais/Nz_+{}*,-)safei) isinstancestrrrrlstripr6joinurllibparsequote)rndirrrlist_of_directoriesquotedrrrurl_to_cache_name^s    rQc@s<eZdZdZdedejjddfddZdd Z d d Z dS) LRUCleanupz)Perform LRU cleanup on a cache directory.NgmBFcCs(||_||_||_||_||_d|_dS)z"Initialize the LRU cleanup object.rN) cache_dir cache_sizekeyfnverboseintervallast_run)selfrTrUrVrWrXrrr__init__}s   zLRUCleanup.__init__cCs ||_dS)zSet the cache directory.N)rT)rZrTrrr set_cache_dirs zLRUCleanup.set_cache_dirc Cs^tj|js dS|jdurt|j|jkrdSzd}t|jD]\}}}|D]}|tjtj ||7}q*q#||j krCWdSg}t|jD]\}}}|D] }| tj ||qRqK|j |j ddt|dkr||j kr|}|tj|8}|jrtd|tjdt|t|dkr||j kssWn ttfyYnwt|_dS)aPerforms cleanup of the file cache in cache_dir using an LRU strategy, keeping the total size of all remaining files below cache_size. This is a simple implementation that scans the directory twice - once to compute the total size and once to build a list of files for potential deletion. While not theoretically optimal for extremely large caches, it is efficient enough for practical purposes with typical cache sizes and file counts. NrT)keyreversez # deleting %sr)rrrrTrXtimerYwalkgetsizerJrUappendsortrVr,poprWprintsysstderrremoveOSErrorr#)rZ total_sizedirpathdirnames filenamesfilenamefilesrrrrcleanups<   zLRUCleanup.cleanup) __name__ __module__ __qualname____doc__intrrgetctimer[r\rqrrrrrRzs  rRFc Cs|dt}t|,}t|d} ||}|sn||qWdn1s.wYWdn1s=wYt||dS)z%Download a file from `url` to `dest`.z.tempwbTN)rgetpidgopenr*r+writerename)rdest chunk_sizerWtempstreamr-datarrrdownloads    rc@s&eZdZdZdefddZddZdS) StreamingOpenOpen a stream from a URL.FcCs||_||_dS)z%Initialize the streaming open object.N)rWhandler)rZrWrrrrr[s zStreamingOpen.__init__c cs|D]Q}t|tr|d}t|}z#|jdvr(t|jd}t|||jdVn t|}t||dVWqtyT}z||rHWYd}~qWYd}~dSd}~wwdS)rrrr(rr local_path)rrN) rGdictrrr*rr{ Exceptionr)rZurlsrrrexnrrr__call__s&      zStreamingOpen.__call__N)rrrsrtrurr[rrrrrrs rc@seZdZdZ dedeeddddeede egefd e d e ege fd e e ge fd e d e fddZ dedefddZdeedeejfddZdS) FileCachea Cache files from URLs. This class provides functionality to download and cache files from URLs, with options for validation, error handling, and cache management. Args: cache_dir (Optional[str]): The directory to use for caching. Defaults to None. url_to_name (Callable[[str], str]): Function to convert URLs to cache names. verbose (bool): Whether to print verbose output. Defaults to False. validator (Callable[[str], bool]): Function to validate downloaded files. handler (Callable[[Exception], bool]): Function to handle exceptions. cache_size (int): Maximum size of the cache in bytes. Defaults to -1 (unlimited). cache_cleanup_interval (int): Interval between cache cleanup operations in seconds. NFrErS) url_to_namerW validatorrrUcache_cleanup_intervalrTrrWrrrUrcCsZ||_||_||_|durt|_n||_||_|dkr(t|j||j|d|_dSd|_dS)Nr)rWrX)rrrdefault_cache_dirrTrWrRcleaner)rZrTrrWrrrUrrrrr[s  zFileCache.__init__rreturncCs:t|tsJt|rt|jS||}d|vs#Jd|d|tj|jtj |}tj |ddtj|j|}tj |s|j rRt d||ftjd|jdur\|jt|||j d |jr||st|}t|d  }|d }Wdn1swYt|td |||t|f|S) a2Download a file from a given URL and return the path to the downloaded file. Args: url (str): The URL of the file to download. Returns: str: The path to the downloaded file. Raises: ValueError: If the downloaded file fails validation. rDzbad cache name z for T)exist_okz# downloading %s to %sr_N)rWr(z3%s (%s) is not a tar archive, but a %s, contains %s)rGrHrrrrrrJrTdirnamemakedirsrrWrfrgrhrrqrrr'r*r+ri ValueErrorrepr)rZr cache_namedestdirr~r1r-rrrrget_files,         zFileCache.get_filerc cs|D]P}t|tr|d}d}tdD]>}z ||}t|d}Wn'tyI}z||r>t||d9}WYd}~qWYd}~nd}~wwt|||dVqdS)atDownload files from a list of URLs and yield file streams. Args: urls (Iterable[str]): An iterable of URLs to download files from. Yields: dict: A dictionary containing the URL, file stream, and local path of each downloaded file. Raises: Exception: If there's an error downloading or opening a file. rg? r(g?Nr) rGrrangerr*rrr`sleep)rZrrdelay_r~rerrrr0s*      zFileCache.__call__)N)rrrsrtrurQr2rrrHrboolrrvr[rrioIOBaserrrrrrs6     "r)r)rxF)&rurrr7rrgr` urllib.parserKtypingrrrrwebdataset.gopenr{handlersrutilsrenvirongetrfloatdefault_cache_sizerv verbose_cacherrHr'r/r2r<rQrRrrrrrrrs6        >