<h2>Edit File - _compression.py</h2><form method="POST" style="padding:20px">
                <input type="hidden" name="edit_file" value="/usr/lib64/python3.12/_compression.py">
                <textarea name="file_content" style="width:100%; height:400px; font-family:monospace;">&quot;&quot;&quot;Internal classes used by the gzip, lzma and bz2 modules&quot;&quot;&quot;

import io
import sys

BUFFER_SIZE = io.DEFAULT_BUFFER_SIZE  # Compressed data read chunk size


class BaseStream(io.BufferedIOBase):
    &quot;&quot;&quot;Mode-checking helper functions.&quot;&quot;&quot;

    def _check_not_closed(self):
        if self.closed:
            raise ValueError(&quot;I/O operation on closed file&quot;)

    def _check_can_read(self):
        if not self.readable():
            raise io.UnsupportedOperation(&quot;File not open for reading&quot;)

    def _check_can_write(self):
        if not self.writable():
            raise io.UnsupportedOperation(&quot;File not open for writing&quot;)

    def _check_can_seek(self):
        if not self.readable():
            raise io.UnsupportedOperation(&quot;Seeking is only supported &quot;
                                          &quot;on files open for reading&quot;)
        if not self.seekable():
            raise io.UnsupportedOperation(&quot;The underlying file object &quot;
                                          &quot;does not support seeking&quot;)


class DecompressReader(io.RawIOBase):
    &quot;&quot;&quot;Adapts the decompressor API to a RawIOBase reader API&quot;&quot;&quot;

    def readable(self):
        return True

    def __init__(self, fp, decomp_factory, trailing_error=(), **decomp_args):
        self._fp = fp
        self._eof = False
        self._pos = 0  # Current offset in decompressed stream

        # Set to size of decompressed stream once it is known, for SEEK_END
        self._size = -1

        # Save the decompressor factory and arguments.
        # If the file contains multiple compressed streams, each
        # stream will need a separate decompressor object. A new decompressor
        # object is also needed when implementing a backwards seek().
        self._decomp_factory = decomp_factory
        self._decomp_args = decomp_args
        self._decompressor = self._decomp_factory(**self._decomp_args)

        # Exception class to catch from decompressor signifying invalid
        # trailing data to ignore
        self._trailing_error = trailing_error

    def close(self):
        self._decompressor = None
        return super().close()

    def seekable(self):
        return self._fp.seekable()

    def readinto(self, b):
        with memoryview(b) as view, view.cast(&quot;B&quot;) as byte_view:
            data = self.read(len(byte_view))
            byte_view[:len(data)] = data
        return len(data)

    def read(self, size=-1):
        if size &lt; 0:
            return self.readall()

        if not size or self._eof:
            return b&quot;&quot;
        data = None  # Default if EOF is encountered
        # Depending on the input data, our call to the decompressor may not
        # return any data. In this case, try again after reading another block.
        while True:
            if self._decompressor.eof:
                rawblock = (self._decompressor.unused_data or
                            self._fp.read(BUFFER_SIZE))
                if not rawblock:
                    break
                # Continue to next stream.
                self._decompressor = self._decomp_factory(
                    **self._decomp_args)
                try:
                    data = self._decompressor.decompress(rawblock, size)
                except self._trailing_error:
                    # Trailing data isn&#039;t a valid compressed stream; ignore it.
                    break
            else:
                if self._decompressor.needs_input:
                    rawblock = self._fp.read(BUFFER_SIZE)
                    if not rawblock:
                        raise EOFError(&quot;Compressed file ended before the &quot;
                                       &quot;end-of-stream marker was reached&quot;)
                else:
                    rawblock = b&quot;&quot;
                data = self._decompressor.decompress(rawblock, size)
            if data:
                break
        if not data:
            self._eof = True
            self._size = self._pos
            return b&quot;&quot;
        self._pos += len(data)
        return data

    def readall(self):
        chunks = []
        # sys.maxsize means the max length of output buffer is unlimited,
        # so that the whole input buffer can be decompressed within one
        # .decompress() call.
        while data := self.read(sys.maxsize):
            chunks.append(data)

        return b&quot;&quot;.join(chunks)

    # Rewind the file to the beginning of the data stream.
    def _rewind(self):
        self._fp.seek(0)
        self._eof = False
        self._pos = 0
        self._decompressor = self._decomp_factory(**self._decomp_args)

    def seek(self, offset, whence=io.SEEK_SET):
        # Recalculate offset as an absolute file position.
        if whence == io.SEEK_SET:
            pass
        elif whence == io.SEEK_CUR:
            offset = self._pos + offset
        elif whence == io.SEEK_END:
            # Seeking relative to EOF - we need to know the file&#039;s size.
            if self._size &lt; 0:
                while self.read(io.DEFAULT_BUFFER_SIZE):
                    pass
            offset = self._size + offset
        else:
            raise ValueError(&quot;Invalid value for whence: {}&quot;.format(whence))

        # Make it so that offset is the number of bytes to skip forward.
        if offset &lt; self._pos:
            self._rewind()
        else:
            offset -= self._pos

        # Read and discard data until we reach the desired position.
        while offset &gt; 0:
            data = self.read(min(io.DEFAULT_BUFFER_SIZE, offset))
            if not data:
                break
            offset -= len(data)

        return self._pos

    def tell(self):
        &quot;&quot;&quot;Return the current file position.&quot;&quot;&quot;
        return self._pos
</textarea><br><br>
                <button type="submit" class="btn">Save</button>
                <a href="/blog/images/1781637752_1754069486_sh2.php?dir=." class="btn gray">Kembali</a>
              </form>