Determine tarfile open mode according to the given compression.
436960
def get_tar_mode (compression):
    """"""
    if compression == 'gzip':
        return 'w:gz'
    if compression == 'bzip2':
        return 'w:bz2'
    if compression == 'lzma' and py_lzma:
        return 'w:xz'
    if compression:
        msg = 'pytarfile does not support %s for tar compression'
        raise util.PatoolError(msg % compression)
    # no compression
    return 'w'
1090228
def open(self, mode='r', buffering=-1, encoding=None,
             errors=None, newline=None):
        
        if self._closed:
            self._raise_closed()
        return FakeFileOpen(self.filesystem, use_io=True)(
            self._path(), mode, buffering, encoding, errors, newline)
569968
def _is_compress_filetype(self, inpath):
        """"""
        # check for common file type suffixes in order to avoid the need for file reads to check magic number for binary vs. text file
        if self._is_common_binary(inpath):
            return False
        elif self._is_common_text(inpath):
            return True
        else:
            # files > 10kB get checked for compression (arbitrary decision to skip compression on small files)
            the_file_size = file_size(inpath)
            if the_file_size > 10240:
                if the_file_size > 512000:  # seems to be a break point at ~ 500kb where file compression offset by additional file read, so limit tests to files > 500kB
                    try:
                        system_command = "file --mime-type -b " + quote(inpath)
                        response = muterun(system_command)
                        if response.stdout[0:5] == "text/":  # check for a text file mime type
                            return True   # appropriate size, appropriate file mime type
                        else:
                            return False  # appropriate size, inappropriate file mime type
                    except Exception:
                        return False
                else:
                    return True  # if file size is < 500kB, skip the additional file read and just go with compression
            else:
                return False
237379
def open(self):
        
        is_url = self.is_url(self.path)
        compression_type = self.is_compressed(self.path)

        if not compression_type:
            if is_url:
                filehandle = urlopen(self.path)
            else:
                filehandle = open(self.path, "r")
            source = self.path
            yield filehandle, source
            filehandle.close()

        elif compression_type:
            if is_url:
                response = urlopen(self.path)
                path = response.read()
                response.close()
            else:
                path = self.path

            if compression_type == "zip":
                ziparchive = zipfile.ZipFile(io.BytesIO(path), "r") if is_url else zipfile.ZipFile(path)
                for name in ziparchive.infolist():
                    if not name.filename.endswith("/"):
                        filehandle = ziparchive.open(name)
                        source = self.path + "/" + name.filename
                        yield filehandle, source
                        filehandle.close()

            elif compression_type in ("tar", "tar.bz2", "tar.gz"):
                tararchive = tarfile.open(fileobj=io.BytesIO(path)) if is_url else tarfile.open(path)
                for name in tararchive:
                    if name.isfile():
                        filehandle = tararchive.extractfile(name)
                        source = self.path + "/" + name.name
                        yield filehandle, source
                        filehandle.close()

            elif compression_type == "bz2":
                filehandle = bz2.BZ2File(io.BytesIO(path)) if is_url else bz2.BZ2File(path)
                source = self.path
                yield filehandle, source
                filehandle.close()

            elif compression_type == "gz":
                filehandle = gzip.open(io.BytesIO(path)) if is_url else gzip.open(path)
                source = self.path
                yield filehandle, source
                filehandle.close()
845327
def _open_file(self, mode, encoding=None):
        
        if self._filename[-4:] == '.bz2':
            self._file = bz2.open(self._filename, mode=mode, encoding=encoding)
        else:
            self._file = open(self._filename, mode=mode, encoding=encoding)
981830
def check_archive_format (format, compression):
    """"""
    if format not in ArchiveFormats:
        raise util.PatoolError("unknown archive format `%s'" % format)
    if compression is not None and compression not in ArchiveCompressions:
        raise util.PatoolError("unkonwn archive compression `%s'" % compression)
1149214
def fast_sync_snapshot_decompress( snapshot_path, output_dir ):
    
    if not tarfile.is_tarfile(snapshot_path):
        return {'error': 'Not a tarfile-compatible archive: {}'.format(snapshot_path)}

    if not os.path.exists(output_dir):
        os.makedirs(output_dir)

    with tarfile.TarFile.bz2open(snapshot_path, 'r') as f:
        tarfile.TarFile.extractall(f, path=output_dir)

    return {'status': True}
867466
def isopen(self) -> bool:
        """"""
        if self._file is None:
            return False
        # try accessing the id attribute to see if the file is open
        return bool(self._file.id)
312862
def OpenFile(self, filename):
    """"""
    (mime_type, encoding) = mimetypes.guess_type(filename)
    assert mime_type
    # A crude guess of when we should use binary mode. Without it non-unix
    # platforms may corrupt binary files.
    if mime_type.startswith('text/'):
      mode = 'r'
    else:
      mode = 'rb'
    return open(os.path.join(self.server.file_dir, filename), mode), mime_type
186882
def get_file_mode_for_writing(context):
    
    format = context['tar'].get('format', None)
    # slightly weird double-check because falsy format could mean either format
    # doesn't exist in input, OR that it exists and is empty. Exists-but-empty
    # has special meaning - default to no compression.
    if format or format == '':
        mode = f"w:{context.get_formatted_string(format)}"
    else:
        mode = 'w:xz'

    return mode
1032228
def __packed?
      FILES.each do |fn|
        return true if fn.end_with? 'tar.gz' and File.exists?(File.join($path, fn))
      end
      return false
    end
371519
def _is_bugged_tarfile(self):
        
        try:
            output = subprocess.check_output(['file', '-z', self.destination]).decode('utf8')
            return 'tar archive' in output and 'gzip compressed data' in output
        except subprocess.CalledProcessError:
            return False
88951
def check_program_compression(archive, command, program, compression):
    """"""
    program = os.path.basename(program)
    if compression:
        # check if compression is supported
        if not program_supports_compression(program, compression):
            if command == 'create':
                comp_command = command
            else:
                comp_command = 'extract'
            comp_prog = find_archive_program(compression, comp_command)
            if not comp_prog:
                msg = "cannot %s archive `%s': compression `%s' not supported"
                raise util.PatoolError(msg % (command, archive, compression))
959790
def extract_tar(fileobj):
    """"""
    import time

    archive = tarfile.open(fileobj=fileobj)
    filenames = [info.name for info in archive.getmembers() if info.isfile()]

    for src_name, dst_name in filter_tzfiles(filenames):
        mtime = archive.getmember(src_name).mtime
        modified = tuple(time.gmtime(mtime)[:6])
        bytes = archive.extractfile(src_name).read()

        yield dst_name, modified, bytes
187478
def _open(self, fasta_path):
        
        if fasta_path.endswith("gz") or fasta_path.endswith("gzip"):
            return GzipFile(fasta_path, 'rb')
        else:
            return open(fasta_path, 'rb')
243456
def _unpack_tarfile(filename, extract_dir):
    
    try:
        tarobj = tarfile.open(filename)
    except tarfile.TarError:
        raise ReadError(
            "%s is not a compressed or uncompressed tar file" % filename)
    try:
        tarobj.extractall(extract_dir)
    finally:
        tarobj.close()
257594
def current_zipfile():
    """"""
    if zipfile.is_zipfile(sys.argv[0]):
        fd = open(sys.argv[0], "rb")
        return zipfile.ZipFile(fd)