code and tests for decoding the flag in the blosc header

Blosc · Mar 12, 2018 · 6a9064d · 6a9064d
1 parent 9e73280
commit 6a9064d
Show file tree

Hide file tree

Showing 3 changed files with 47 additions and 0 deletions.
diff --git a/bloscpack/constants.py b/bloscpack/constants.py
@@ -37,3 +37,10 @@
 
 # Codecs available from Blosc
 CNAME_AVAIL = blosc.compressor_list()
+CNAME_MAPPING = {
+    0: 'blosclz',
+    1: 'lz4',
+    2: 'snappy',
+    3: 'zlib',
+    4: 'zstd',
+}
diff --git a/bloscpack/headers.py b/bloscpack/headers.py
@@ -27,6 +27,7 @@
                         MAX_CLEVEL,
                         BLOSCPACK_HEADER_LENGTH,
                         MAX_META_SIZE,
+                        CNAME_MAPPING,
                         )
 from .defaults import (DEFAULT_OFFSETS,
                        )
@@ -220,6 +221,15 @@ def decode_blosc_header(buffer_):
                         ('ctbytes', decode_uint32(buffer_[12:16]))))
 
 
+def decode_blosc_flags(byte_):
+    return OrderedDict((('byte_shuffle',  bool(byte_ & 1)),
+                        ('pure_memcpy',   bool(byte_ >> 1 & 1)),
+                        ('bit_shuffle',   bool(byte_ >> 2 & 1)),
+                        ('split_blocks',  bool(byte_ >> 4 & 1)),
+                        ('codec',         CNAME_MAPPING[byte_ >> 5 & 7]),
+                        ))
+
+
 class BloscpackHeader(MutableMappingObject):
     """ The Bloscpack header.
 

diff --git a/test/test_headers.py b/test/test_headers.py
@@ -15,6 +15,7 @@
 
 from bloscpack.args import (BloscArgs,
                             )
+from bloscpack.compat_util import OrderedDict
 from bloscpack.constants import (MAGIC,
                                  FORMAT_VERSION,
                                  MAX_FORMAT_VERSION,
@@ -33,6 +34,7 @@
                                decode_metadata_options,
                                check_range,
                                decode_blosc_header,
+                               decode_blosc_flags,
                                )
 
 
@@ -191,6 +193,34 @@ def test_decode_blosc_header_uncompressible_data_dont_split_false():
     nt.assert_equal(expected, header)
 
 
+def test_decode_blosc_flags():
+
+    def gen_expected(new):
+        it = OrderedDict((
+            ('byte_shuffle', False),
+            ('pure_memcpy', False),
+            ('bit_shuffle', False),
+            ('split_blocks', False),
+            ('codec', 'blosclz'),
+        ))
+        it.update(new)
+        return it
+    for input_byte, new_params in [
+            (0b00000000, {}),
+            (0b00000001, {'byte_shuffle': True}),
+            (0b00000010, {'pure_memcpy': True}),
+            (0b00000100, {'bit_shuffle': True}),
+            (0b00010000, {'split_blocks': True}),
+            (0b00100000, {'codec': 'lz4'}),
+            (0b01000000, {'codec': 'snappy'}),
+            (0b01100000, {'codec': 'zlib'}),
+            (0b10000000, {'codec': 'zstd'}),
+            ]:
+        yield (nt.assert_equal,
+               decode_blosc_flags(input_byte),
+               gen_expected(new_params))
+
+
 def test_BloscPackHeader_constructor_exceptions():
     # uses nose test generators