FM / iter

/usr/local/lib64/python3.6/site-packages/torch/utils/data/datapipes/iter/

/usr/local/lib64/python3.6/site-packages/torch/utils/data/datapipes/iter

Name	Size	Mode	Actions
__pycache__/	-	0755	rm
callable.py	8953	0644	edit dl rm
combinatorics.py	4337	0644	edit dl rm
combining.py	16028	0644	edit dl rm
filelister.py	1385	0644	edit dl rm
fileloader.py	1911	0644	edit dl rm
grouping.py	12920	0644	edit dl rm
httpreader.py	1764	0644	edit dl rm
linereader.py	732	0644	edit dl rm
routeddecoder.py	2354	0644	edit dl rm
selecting.py	6018	0644	edit dl rm
streamreader.py	822	0644	edit dl rm
tararchivereader.py	2921	0644	edit dl rm
utils.py	1501	0644	edit dl rm
ziparchivereader.py	2921	0644	edit dl rm
__init__.py	2463	0644	edit dl rm

Edit: /usr/local/lib64/python3.6/site-packages/torch/utils/data/datapipes/iter/tararchivereader.py (2921B)

import os
import tarfile
import warnings

class TarArchiveReaderIterDataPipe(IterDataPipe[Tuple[str, BufferedIOBase]]):
    r""" :class:`TarArchiveReaderIterDataPipe`.

Iterable datapipe to extract tar binary streams from input iterable which contains tuples of pathnames and
    tar binary stream. This yields a tuple of pathname and extracted binary stream.

Args:
        datapipe: Iterable datapipe that provides tuples of pathname and tar binary stream
        mode: File mode used by `tarfile.open` to read file object.
            Mode has to be a string of the form 'filemode[:compression]'
        length: a nominal length of the datapipe

Note:
        The opened file handles will be closed automatically if the default DecoderDataPipe
        is attached. Otherwise, user should be responsible to close file handles explicitly
        or let Python's GC close them periodically.
    """
    def __init__(
        self,
        datapipe: Iterable[Tuple[str, BufferedIOBase]],
        mode: str = "r:*",
        length: int = -1
    ):
        super().__init__()
        self.datapipe: Iterable[Tuple[str, BufferedIOBase]] = datapipe
        self.mode: str = mode
        self.length: int = length
        deprecation_warning_torchdata(type(self).__name__)

def __iter__(self) -> Iterator[Tuple[str, BufferedIOBase]]:
        for data in self.datapipe:
            validate_pathname_binary_tuple(data)
            pathname, data_stream = data
            folder_name = os.path.dirname(pathname)
            try:
                # typing.cast is used here to silence mypy's type checker
                tar = tarfile.open(fileobj=cast(Optional[IO[bytes]], data_stream), mode=self.mode)
                for tarinfo in tar:
                    if not tarinfo.isfile():
                        continue
                    extracted_fobj = tar.extractfile(tarinfo)
                    if extracted_fobj is None:
                        warnings.warn("failed to extract file {} from source tarfile {}".format(tarinfo.name, pathname))
                        raise tarfile.ExtractError
                    inner_pathname = os.path.normpath(os.path.join(folder_name, tarinfo.name))
                    yield (inner_pathname, extracted_fobj)  # type: ignore[misc]
            except Exception as e:
                warnings.warn(
                    "Unable to extract files from corrupted tarfile stream {} due to: {}, abort!".format(pathname, e))
                raise e

def __len__(self):
        if self.length == -1:
            raise TypeError("{} instance doesn't have valid length".format(type(self).__name__))
        return self.length

from torch.utils.data import IterDataPipe from torch.utils.data.datapipes.utils.common import validate_pathname_binary_tuple, deprecation_warning_torchdata from typing import Iterable, Iterator, Tuple, Optional, IO, cast from io import BufferedIOBase import os import tarfile import warnings class TarArchiveReaderIterDataPipe(IterDataPipe[Tuple[str, BufferedIOBase]]): r""" :class:`TarArchiveReaderIterDataPipe`. Iterable datapipe to extract tar binary streams from input iterable which contains tuples of pathnames and tar binary stream. This yields a tuple of pathname and extracted binary stream. Args: datapipe: Iterable datapipe that provides tuples of pathname and tar binary stream mode: File mode used by `tarfile.open` to read file object. Mode has to be a string of the form 'filemode[:compression]' length: a nominal length of the datapipe Note: The opened file handles will be closed automatically if the default DecoderDataPipe is attached. Otherwise, user should be responsible to close file handles explicitly or let Python's GC close them periodically. """ def __init__( self, datapipe: Iterable[Tuple[str, BufferedIOBase]], mode: str = "r:*", length: int = -1 ): super().__init__() self.datapipe: Iterable[Tuple[str, BufferedIOBase]] = datapipe self.mode: str = mode self.length: int = length deprecation_warning_torchdata(type(self).__name__) def __iter__(self) -> Iterator[Tuple[str, BufferedIOBase]]: for data in self.datapipe: validate_pathname_binary_tuple(data) pathname, data_stream = data folder_name = os.path.dirname(pathname) try: # typing.cast is used here to silence mypy's type checker tar = tarfile.open(fileobj=cast(Optional[IO[bytes]], data_stream), mode=self.mode) for tarinfo in tar: if not tarinfo.isfile(): continue extracted_fobj = tar.extractfile(tarinfo) if extracted_fobj is None: warnings.warn("failed to extract file {} from source tarfile {}".format(tarinfo.name, pathname)) raise tarfile.ExtractError inner_pathname = os.path.normpath(os.path.join(folder_name, tarinfo.name)) yield (inner_pathname, extracted_fobj) # type: ignore[misc] except Exception as e: warnings.warn( "Unable to extract files from corrupted tarfile stream {} due to: {}, abort!".format(pathname, e)) raise e def __len__(self): if self.length == -1: raise TypeError("{} instance doesn't have valid length".format(type(self).__name__)) return self.length