######################################################################## # # Created: September 22, 2010 # Author: The Blosc development team - blosc@blosc.org # ######################################################################## import os import pickle import subprocess import sys from ._version import LooseVersion from blosc import blosc_extension as _ext import blosc def detect_number_of_cores(): """ detect_number_of_cores() Detect the number of cores in this system. Returns ------- out : int The number of cores in this system. """ # Linux, Unix and MacOS: if hasattr(os, "sysconf"): if "SC_NPROCESSORS_ONLN" in os.sysconf_names: # Linux & Unix: ncpus = os.sysconf("SC_NPROCESSORS_ONLN") if isinstance(ncpus, int) and ncpus > 0: return ncpus else: # OSX: completed = subprocess.run(("sysctl", "-n", "hw.ncpu"), stdout=subprocess.PIPE) if completed.returncode == 0: ncpus = int(completed.stdout) if ncpus > 0: return ncpus # Windows: elif "NUMBER_OF_PROCESSORS" in os.environ: ncpus = int(os.environ["NUMBER_OF_PROCESSORS"]) if ncpus > 0: return ncpus return 1 # Default def set_nthreads(nthreads): """ set_nthreads(nthreads) Set the number of threads to be used during Blosc operation. Parameters ---------- nthreads : int The number of threads to be used during Blosc operation. Returns ------- out : int The previous number of used threads. Raises ------ ValueError If nthreads is larger that the maximum number of threads blosc can use. Notes ----- The number of threads for Blosc is the maximum number of cores detected on your machine (via `detect_number_of_cores`). In some cases Blosc gets better results if you set the number of threads to a value slightly below than your number of cores. Examples -------- Set the number of threads to 2 and then to 1: >>> oldn = blosc.set_nthreads(2) >>> blosc.set_nthreads(1) 2 """ if nthreads > blosc.MAX_THREADS: raise ValueError("the number of threads cannot be larger than %d" % blosc.MAX_THREADS) blosc.nthreads = nthreads return _ext.set_nthreads(nthreads) def set_blocksize(blocksize): """set_blocksize(blocksize) Force the use of a specific blocksize. If 0, an automatic blocksize will be used (the default). Notes ----- This is a low-level function and is recommended for expert users only. Examples -------- >>> blosc.set_blocksize(512) >>> blosc.set_blocksize(0) """ _ext.set_blocksize(blocksize) def get_blocksize(): """ get_blocksize() Get the blocksize currently used. Examples -------- >>> blosc.set_blocksize(0) >>> blosc.get_blocksize() 0 """ return _ext.get_blocksize() def set_releasegil(gilstate): """ set_releasegil( gitstate ) Sets a boolean on whether to release the Python global inter-lock (GIL) during c-blosc compress and decompress operations or not. This defaults to False. Notes ----- Designed to be used with larger chunk sizes and a ThreadPool. There is a small performance penalty with releasing the GIL that will more harshly penalize small block sizes. Examples -------- >>> oldReleaseState = blosc.set_releasegil(True) """ gilstate = bool(gilstate) return _ext.set_releasegil(gilstate) def compressor_list(): """ compressor_list() Returns a list of compressors available in C library. Parameters ---------- None Returns ------- out : list The list of names. """ return _ext.compressor_list().split(',') def code_to_name(code): """ code_to_name(code) Return the compressor name of a compressor code. Parameters ---------- code : int The compressor code. Returns ------- out : str The compressor name. """ return _ext.code_to_name(code) def name_to_code(name): """ name_to_code(name) Return the compressor code of a compressor name. Parameters ---------- name : str The compressor name. Returns ------- out : int The compressor code. """ return _ext.name_to_code(name) def clib_info(cname): """ clib_info(cname) Return info for compression libraries in C library. Parameters ---------- cname : str The compressor name. Returns ------- out : tuple The associated library name and version. """ return _ext.clib_info(cname) def get_clib(bytesobj): """ get_clib(bytesobj) Return the name of the compression library for Blosc `bytesobj` buffer. Parameters ---------- bytesobj : str / bytes The compressed buffer. Returns ------- out : str The name of the compression library. """ _check_bytesobj(bytesobj) return _ext.get_clib(bytesobj) def get_cbuffer_sizes(bytesobj): """ get_cbuffer_sizes(bytesobj) Return information about a compressed buffer: (number of uncompressed bytes, number of compressed bytes, blocksize). Parameters ---------- bytesobj : str / bytes The compressed buffer. Returns ------- out : tuple The associated uncompressed bytes, compressed bytes and blocksize Examples -------- >>> s = b'0123456789' >>> c = blosc.compress(s, typesize=1) >>> blosc.get_cbuffer_sizes(c) (10, 26, 10) """ return _ext.get_cbuffer_sizes(bytesobj) def cbuffer_validate(bytesobj): """ validate_cbuffer(bytesobj) Validate the cbuffer. Check that the cbuffer is safe to compress. Note that this does not guarantee that the blosc will be able to decompress the buffer successfully, only that it is safe to attempt to do so. Parameters ---------- bytesobj : str / bytes The compressed buffer. Returns ------- result : boolean True if safe, False if not. Examples -------- >>> s = b'0123456789' >>> c = blosc.compress(s, typesize=1) >>> blosc.cbuffer_validate(c) True """ return _ext.cbuffer_validate(bytesobj) def free_resources(): """ free_resources() Free possible memory temporaries and thread resources. Returns ------- out : None Notes ----- Blosc maintain a pool of threads waiting for work as well as some temporary space. You can use this function to release these resources when you are not going to use Blosc for a long while. Examples -------- >>> blosc.free_resources() >>> """ _ext.free_resources() def _check_shuffle(shuffle): if shuffle not in (blosc.NOSHUFFLE, blosc.SHUFFLE, blosc.BITSHUFFLE): raise ValueError("shuffle can only be one of NOSHUFFLE, SHUFFLE" " and BITSHUFFLE.") if (shuffle == blosc.BITSHUFFLE and LooseVersion(blosc.blosclib_version) < LooseVersion("1.8.0")): raise ValueError("You need C-Blosc 1.8.0 or higher for using" " BITSHUFFLE.") def _check_clevel(clevel): if not 0 <= clevel <= 9: raise ValueError("clevel can only be in the 0-9 range.") def _check_cname(cname): list_cnames = compressor_list() if cname not in list_cnames: raise ValueError("cname can only be one of: %s, not '%s'" % (list_cnames, cname)) def _check_typesize(typesize): if not 1 <= typesize <= blosc.MAX_TYPESIZE: raise ValueError("typesize can only be in the 1-%d range." % blosc.MAX_TYPESIZE) def _check_bytesobj(bytesobj): if not isinstance(bytesobj, bytes): raise TypeError("only string (2.x) or bytes (3.x) objects " "supported as input") def _check_byteslike(bytes_like): try: memoryview(bytes_like) except Exception: raise TypeError("Input type %s must be a bytes-like object that supports Python Buffer Protocol" % type(bytes_like)) def _check_input_length(input_name, input_len): if input_len > blosc.MAX_BUFFERSIZE: raise ValueError("%s cannot be larger than %d bytes" % (input_name, blosc.MAX_BUFFERSIZE)) def _check_address(address): if not isinstance(address, int): raise TypeError("only int or long objects are supported as address") def compress(bytesobj, typesize=8, clevel=9, shuffle=blosc.SHUFFLE, cname='blosclz'): """compress(bytesobj[, typesize=8, clevel=9, shuffle=blosc.SHUFFLE, cname='blosclz']]) Compress bytesobj, with a given type size. Parameters ---------- bytesobj : bytes-like object (supporting the buffer interface) The data to be compressed. typesize : int The data type size. clevel : int (optional) The compression level from 0 (no compression) to 9 (maximum compression). The default is 9. shuffle : int (optional) The shuffle filter to be activated. Allowed values are blosc.NOSHUFFLE, blosc.SHUFFLE and blosc.BITSHUFFLE. The default is blosc.SHUFFLE. cname : string (optional) The name of the compressor used internally in Blosc. It can be any of the supported by Blosc ('blosclz', 'lz4', 'lz4hc', 'snappy', 'zlib', 'zstd' and maybe others too). The default is 'blosclz'. Returns ------- out : str / bytes The compressed data in form of a Python str / bytes object. Raises ------ TypeError If bytesobj doesn't support the buffer interface. ValueError If bytesobj is too long. If typesize is not within the allowed range. If clevel is not within the allowed range. If cname is not a valid codec. Examples -------- >>> import array, sys >>> a = array.array('i', range(1000*1000)) >>> a_bytesobj = a.tobytes() >>> c_bytesobj = blosc.compress(a_bytesobj, typesize=4) >>> len(c_bytesobj) < len(a_bytesobj) True """ _check_input_length('bytesobj', len(bytesobj)) _check_typesize(typesize) _check_shuffle(shuffle) _check_clevel(clevel) _check_cname(cname) return _ext.compress(bytesobj, typesize, clevel, shuffle, cname) def compress_ptr(address, items, typesize=8, clevel=9, shuffle=blosc.SHUFFLE, cname='blosclz'): """compress_ptr(address, items[, typesize=8, clevel=9, shuffle=blosc.SHUFFLE, cname='blosclz']]) Compress the data at address with given items and typesize. Parameters ---------- address : int or long the pointer to the data to be compressed items : int The number of items (of typesize) to be compressed. typesize : int The data type size. clevel : int (optional) The compression level from 0 (no compression) to 9 (maximum compression). The default is 9. shuffle : int (optional) The shuffle filter to be activated. Allowed values are blosc.NOSHUFFLE, blosc.SHUFFLE and blosc.BITSHUFFLE. The default is blosc.SHUFFLE. cname : string (optional) The name of the compressor used internally in Blosc. It can be any of the supported by Blosc ('blosclz', 'lz4', 'lz4hc', 'snappy', 'zlib', 'zstd' and maybe others too). The default is 'blosclz'. Returns ------- out : str / bytes The compressed data in form of a Python str / bytes object. Raises ------ TypeError If address is not of type int or long. ValueError If items * typesize is larger than the maximum allowed buffer size. If typesize is not within the allowed range. If clevel is not within the allowed range. If cname is not within the supported compressors. Notes ----- This function can be used anywhere that a memory address is available in Python. For example the Numpy "__array_interface__['data'][0]" construct, or when using the ctypes modules. Importantly, the user is responsible for making sure that the memory address is valid and that the memory pointed to is contiguous. Passing a non-valid address has a high likelihood of crashing the interpreter by segfault. Examples -------- >>> import numpy >>> items = 7 >>> np_array = numpy.arange(items) >>> c = blosc.compress_ptr(np_array.__array_interface__['data'][0], \ items, np_array.dtype.itemsize) >>> d = blosc.decompress(c) >>> np_ans = numpy.fromstring(d, dtype=np_array.dtype) >>> bool((np_array == np_ans).all()) True >>> import ctypes >>> typesize = 8 >>> data = [float(i) for i in range(items)] >>> Array = ctypes.c_double * items >>> a = Array(*data) >>> c = blosc.compress_ptr(ctypes.addressof(a), items, typesize) >>> d = blosc.decompress(c) >>> import struct >>> ans = [struct.unpack('d', d[i:i+typesize])[0] \ for i in range(0, items*typesize, typesize)] >>> data == ans True """ _check_address(address) if items < 0: raise ValueError("items cannot be negative") length = items * typesize _check_input_length('length', length) _check_typesize(typesize) _check_shuffle(shuffle) _check_clevel(clevel) _check_cname(cname) return _ext.compress_ptr(address, length, typesize, clevel, shuffle, cname) def decompress(bytes_like, as_bytearray=False): """decompress(bytes_like) Decompresses a bytes-like compressed object. Parameters ---------- bytes_like : bytes-like object The data to be decompressed. Must be a bytes-like object that supports the Python Buffer Protocol, like bytes, bytearray, memoryview, or numpy.ndarray. as_bytearray : bool, optional If this flag is True then the return type will be a bytearray object instead of a bytesobject. Returns ------- out : str / bytes or bytearray The decompressed data in form of a Python str / bytes object. If as_bytearray is True then this will be a bytearray object, otherwise this will be a str/ bytes object. Raises ------ TypeError If bytes_like does not support Buffer Protocol Examples -------- >>> import array, sys >>> a = array.array('i', range(1000*1000)) >>> a_bytesobj = a.tobytes() >>> c_bytesobj = blosc.compress(a_bytesobj, typesize=4) >>> a_bytesobj2 = blosc.decompress(c_bytesobj) >>> a_bytesobj == a_bytesobj2 True >>> b"" == blosc.decompress(blosc.compress(b"", 1)) True >>> b"1"*7 == blosc.decompress(blosc.compress(b"1"*7, 8)) True >>> type(blosc.decompress(blosc.compress(b"1"*7, 8), ... as_bytearray=True)) is bytearray True """ return _ext.decompress(bytes_like, as_bytearray) def decompress_ptr(bytes_like, address): """decompress_ptr(bytes_like, address) Decompresses a bytes_like compressed object into the memory at address. Parameters ---------- bytes_like : bytes-like object The data to be decompressed. Must be a bytes-like object that supports the Python Buffer Protocol, like bytes, bytearray, memoryview, or numpy.ndarray. address : int or long The address at which to write the decompressed data Returns ------- nbytes : int the number of bytes written to the buffer Raises ------ TypeError If bytesobj is not of type bytes or string. If address is not of type int or long. Notes ----- This function can be used anywhere that a memory address is available in Python. For example the Numpy "__array_interface__['data'][0]" construct, or when using the ctypes modules. Importantly, the user is responsible for making sure that the memory address is valid and that the memory pointed to is contiguous and can be written to. Passing a non-valid address has a high likelihood of crashing the interpreter by segfault. Examples -------- >>> import numpy >>> items = 7 >>> np_array = numpy.arange(items) >>> c = blosc.compress_ptr(np_array.__array_interface__['data'][0], \ items, np_array.dtype.itemsize) >>> np_ans = numpy.empty(items, dtype=np_array.dtype) >>> nbytes = blosc.decompress_ptr(c, np_ans.__array_interface__['data'][0]) >>> bool((np_array == np_ans).all()) True >>> nbytes == items * np_array.dtype.itemsize True >>> import ctypes >>> typesize = 8 >>> data = [float(i) for i in range(items)] >>> Array = ctypes.c_double * items >>> in_array = Array(*data) >>> c = blosc.compress_ptr(ctypes.addressof(in_array), items, typesize) >>> out_array = ctypes.create_string_buffer(items*typesize) >>> nbytes = blosc.decompress_ptr(c, ctypes.addressof(out_array)) >>> import struct >>> ans = [struct.unpack('d', out_array[i:i+typesize])[0] \ for i in range(0, items*typesize, typesize)] >>> data == ans True >>> nbytes == items * typesize True """ _check_byteslike(bytes_like) _check_address(address) return _ext.decompress_ptr(bytes_like, address) def pack_array(array, clevel=9, shuffle=blosc.SHUFFLE, cname='blosclz'): """pack_array(array[, clevel=9, shuffle=blosc.SHUFFLE, cname='blosclz']]) Pack (compress) a NumPy array. Parameters ---------- array : ndarray The NumPy array to be packed. clevel : int (optional) The compression level from 0 (no compression) to 9 (maximum compression). The default is 9. shuffle : int (optional) The shuffle filter to be activated. Allowed values are blosc.NOSHUFFLE, blosc.SHUFFLE and blosc.BITSHUFFLE. The default is blosc.SHUFFLE. cname : string (optional) The name of the compressor used internally in Blosc. It can be any of the supported by Blosc ('blosclz', 'lz4', 'lz4hc', 'snappy', 'zlib', 'zstd' and maybe others too). The default is 'blosclz'. Returns ------- out : str / bytes The packed array in form of a Python str / bytes object. Raises ------ TypeError If array does not quack like a numpy ndarray. ValueError If array.itemsize * array.size is larger than the maximum allowed buffer size. If typesize is not within the allowed range. If clevel is not within the allowed range. If cname is not within the supported compressors. Examples -------- >>> import numpy >>> a = numpy.arange(1e6) >>> parray = blosc.pack_array(a) >>> len(parray) < a.size*a.itemsize True """ if not (hasattr(array, 'dtype') and hasattr(array, 'shape')): # This does not quack like an ndarray raise TypeError( "only NumPy ndarray objects supported as input") itemsize = array.itemsize _check_input_length('array size', array.size*itemsize) _check_typesize(array.itemsize) _check_shuffle(shuffle) _check_clevel(clevel) _check_cname(cname) # Use the fastest pickle available pickled_array = pickle.dumps(array, pickle.HIGHEST_PROTOCOL) # ... and compress the pickle packed_array = compress(pickled_array, itemsize, clevel, shuffle, cname) return packed_array def unpack_array(packed_array, **kwargs): """unpack_array(packed_array) Unpack (decompress) a packed NumPy array. Parameters ---------- packed_array : str / bytes The packed array to be decompressed. **kwargs : fix_imports / encoding / errors Optional parameters that can be passed to the pickle.loads API https://docs.python.org/3/library/pickle.html#pickle.loads Returns ------- out : ndarray The decompressed data in form of a NumPy array. Raises ------ TypeError If packed_array is not of type bytes or string. Examples -------- >>> import numpy >>> a = numpy.arange(1e6) >>> parray = blosc.pack_array(a) >>> len(parray) < a.size*a.itemsize True >>> a2 = blosc.unpack_array(parray) >>> bool(numpy.all(a == a2)) True >>> a = numpy.array(['å', 'ç', 'ø']) >>> parray = blosc.pack_array(a) >>> a2 = blosc.unpack_array(parray) >>> bool(numpy.all(a == a2)) True """ _check_bytesobj(packed_array) # First decompress the pickle pickled_array = _ext.decompress(packed_array, False) # ... and unpickle if kwargs: array = pickle.loads(pickled_array, **kwargs) if all(isinstance(x, bytes) for x in array.tolist()): import numpy array = numpy.array([x.decode('utf-8') for x in array.tolist()]) else: array = pickle.loads(pickled_array) return array # For the load tests protocol: # https://docs.python.org/3/library/unittest.html#load-tests-protocol def load_tests(loader, tests, pattern): import doctest tests.addTests(doctest.DocTestSuite()) return tests def os_release_pretty_name(): for p in ('/etc/os-release', '/usr/lib/os-release'): try: f = open(p) for line in f: name, _, value = line.rstrip().partition('=') if name == 'PRETTY_NAME': if len(value) >= 2 and value[0] in '"\'' and value[0] == value[-1]: value = value[1:-1] return value except OSError: pass return None def print_versions(): """Print all the versions of software that python-blosc relies on.""" import platform print("-=" * 38) print("python-blosc version: %s" % blosc.__version__) print("Blosc version: %s" % blosc.blosclib_version) print("Compressors available: %s" % blosc.cnames) print("Compressor library versions:") for clib in sorted(blosc.clib_versions.keys()): print(" %s: %s" % (clib, blosc.clib_versions[clib])) print("Python version: %s" % sys.version) (sysname, nodename, release, version, machine, processor) = platform.uname() print("Platform: %s-%s-%s (%s)" % (sysname, release, machine, version)) if sysname == "Linux": distro = os_release_pretty_name() if distro: print("Linux dist:", distro) if not processor: processor = "not recognized" print("Processor: %s" % processor) print("Byte-ordering: %s" % sys.byteorder) print("Detected cores: %s" % blosc.ncores) print("Number of threads to use by default: %s" % blosc.nthreads) print("-=" * 38) if __name__ == '__main__': # test myself import doctest print_versions() nfail, ntests = doctest.testmod() if nfail == 0: print("All %d tests passed successfully!" % ntests)