Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
def benchmark(data, hcdata=None):
number = 100
size = len(data)
hcdata = hcdata or data
for modname, func in [("pylz4", pylz4.compress), ("rustlz4", rustlz4.compress)]:
timer = timeit.Timer(functools.partial(func, data))
elapsed = timer.timeit(number=number)
perf = size * number / elapsed / 1e6
name = "%s.%s" % (modname, func.__name__)
print("%24s: %8.2f MB/s" % (name, perf))
for modname, func in [("pylz4", pylz4.compressHC), ("rustlz4", rustlz4.compresshc)]:
timer = timeit.Timer(functools.partial(func, hcdata))
elapsed = timer.timeit(number=number)
perf = size * number / elapsed / 1e6
name = "%s.%s" % (modname, func.__name__)
print("%24s: %8.2f MB/s" % (name, perf))
data = pylz4.compress(data)
for modname, func in [("pylz4", pylz4.decompress), ("rustlz4", rustlz4.decompress)]:
timer = timeit.Timer(functools.partial(func, data))
elapsed = timer.timeit(number=number)
perf = size * number / elapsed / 1e6
name = "%s.%s" % (modname, func.__name__)
print("%24s: %8.2f MB/s" % (name, perf))
def chain_comp(data):
return brotli.compress(lzma.compress(bz2.compress(zlib.compress(zlib.compress(lzo.compress(lz4.compressHC(lz4.compress(data))),1),9),9)))
def roundtrip(size=None):
if size is None:
size = struct.unpack(">I", b"\0" + os.urandom(3))[0]
data = os.urandom(size)
assert rustlz4.decompress(pylz4.compress(data)) == data
assert pylz4.decompress(buffer(rustlz4.compress(data))) == data
assert rustlz4.decompress(pylz4.compressHC(data)) == data
assert pylz4.decompress(buffer(rustlz4.compresshc(data))) == data
def zip_compress(plain, level=9):
if not USE_LZ4:
compressed = zlib.compress(plain, level)
return compressed[2:]
else:
compressed = lz4.compress(plain) if level < 9 else lz4.compressHC(plain)
return compressed[4:]
compress = lambda o: lz4.compressHC( pickle.dumps( o ) )
except:
compress = lambda o: lz4.compressHC( pickle.dumps( o ) )
except: