Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
def decode(self, buf, out=None):
arr = ensure_contiguous_ndarray(buf).view('u1')
expect = arr[:4].view('
def decode(self, buf, out=None):
# normalise inputs
buf = ensure_contiguous_ndarray(buf)
if out is not None:
out = ensure_contiguous_ndarray(out)
# do decompression
dec = _zlib.decompress(buf)
# handle destination - Python standard library zlib module does not
# support direct decompression into buffer, so we have to copy into
# out if given
return ndarray_copy(dec, out)
def encode(self, buf):
# normalise input
buf = ensure_contiguous_ndarray(buf)
# do compression
return _bz2.compress(buf, self.level)
def decode(self, buf, out=None):
# normalise inputs
# BytesIO only copies if the data is not of `bytes` type.
# This allows `bytes` objects to pass through without copying.
buf = io.BytesIO(ensure_bytes(buf))
# do decompression
with _gzip.GzipFile(fileobj=buf, mode='rb') as decompressor:
if out is not None:
out_view = ensure_contiguous_ndarray(out)
decompressor.readinto(out_view)
if decompressor.read(1) != b'':
raise ValueError("Unable to fit data into `out`")
else:
out = decompressor.read()
return out
def decode(self, buf, out=None):
# normalise inputs
buf = ensure_contiguous_ndarray(buf)
if out is not None:
out = ensure_contiguous_ndarray(out)
# do decompression
dec = _zlib.decompress(buf)
# handle destination - Python standard library zlib module does not
# support direct decompression into buffer, so we have to copy into
# out if given
return ndarray_copy(dec, out)
def encode(self, buf):
# normalise inputs
buf = ensure_contiguous_ndarray(buf)
# do compression
return _lzma.compress(buf, format=self.format, check=self.check,
preset=self.preset, filters=self.filters)
def encode(self, buf):
arr = ensure_contiguous_ndarray(buf).view('u1')
checksum = self.checksum(arr) & 0xffffffff
enc = np.empty(arr.nbytes + 4, dtype='u1')
enc[:4].view('
def decode(self, buf, out=None):
# normalise inputs
buf = ensure_contiguous_ndarray(buf)
if out is not None:
out = ensure_contiguous_ndarray(out)
# N.B., bz2 cannot handle ndarray directly because of truth testing issues
buf = memoryview(buf)
# do decompression
dec = _bz2.decompress(buf)
# handle destination - Python standard library bz2 module does not
# support direct decompression into buffer, so we have to copy into
# out if given
return ndarray_copy(dec, out)
def encode(self, buf):
# normalise inputs
buf = ensure_contiguous_ndarray(buf)
# do compression
return _zlib.compress(buf, self.level)
def decode(self, buf, out=None):
buf = ensure_contiguous_ndarray(buf)
items = msgpack.unpackb(buf, raw=self.raw)
dec = np.empty(items[-1], dtype=items[-2])
dec[:] = items[:-2]
if out is not None:
np.copyto(out, dec)
return out
else:
return dec