Add Sperr.from_filter_options

Author: t20100

src/hdf5plugin/_filters.py

@@ -711,6 +711,30 @@ def __init__(
             mode, quality, swap, missing_value_mode
         )
 
+    @classmethod
+    def from_filter_options(cls, filter_options: tuple[int, ...]) -> Sperr:
+        if len(filter_options) < 2:
+            raise ValueError(f"Expected at least 2 values, got {len(filter_options)}")
+
+        mode, quality, swap, missing_value_mode = cls.__unpack_options(
+            meta=filter_options[0], ret=filter_options[1]
+        )
+
+        if mode == 2:
+            return cls(
+                peak_signal_to_noise_ratio=quality,
+                swap=swap,
+                missing_value_mode=missing_value_mode,
+            )
+        if mode == 3:
+            return cls(
+                absolute=quality, swap=swap, missing_value_mode=missing_value_mode
+            )
+        if mode == 1:
+            return cls(rate=quality, swap=swap, missing_value_mode=missing_value_mode)
+
+        raise ValueError(f"Mode must be in [1, 3], got {mode}")
+
     @classmethod
     def __pack_options(
         cls, mode: int, quality: float, swap: bool, missing_value_mode: int
@@ -746,6 +770,40 @@ def __pack_options(
 
         return ret, missing_value_mode
 
+    @classmethod
+    def __unpack_options(cls, meta: int, ret: int) -> tuple[int, float, bool, int]:
+        # Unpack missing value mode from packed_info bits 6-9
+        # See h5zsperr_unpack_extra_info
+        missing_value_mode = (meta >> 6) & 0b1111
+
+        # Unpack other fields from ret
+        # See H5Z_SPERR_decode_cd_values
+        swap = bool(ret >> (cls._INTEGER_BITS + cls._FRACTIONAL_BITS + 3))
+
+        bit1 = (ret >> (cls._INTEGER_BITS + cls._FRACTIONAL_BITS)) & 1
+        bit2 = (ret >> (cls._INTEGER_BITS + cls._FRACTIONAL_BITS + 1)) & 1
+        if bit1 and not bit2:
+            mode = 1
+        elif not bit1 and bit2:
+            mode = 2
+        elif bit1 and bit2:
+            mode = 3
+        else:
+            raise ValueError("Mode must be in [1, 3], got 0")
+
+        negative = bool((ret >> (cls._INTEGER_BITS + cls._FRACTIONAL_BITS - 1)) & 1)
+
+        mask = 1 << (cls._INTEGER_BITS + cls._FRACTIONAL_BITS - 1)
+        masked_ret = ret & (mask - 1)
+
+        quality = float(masked_ret) / float(1 << cls._FRACTIONAL_BITS)
+        if negative:
+            quality *= -1.0
+        if mode == 3:
+            quality = math.exp2(quality)
+
+        return mode, quality, swap, missing_value_mode
+
 
 class SZ(FilterBase):
     """``h5py.Group.create_dataset``'s compression arguments for using SZ2 filter.

src/hdf5plugin/test.py

@@ -577,6 +577,27 @@ def testLZ4(self):
                 compression_filter = hdf5plugin.LZ4.from_filter_options(filter_options)
                 self.assertEqual(compression_filter.filter_options, expected_options)
 
+    def testSperr(self):
+        for filter_options, expected_filter in (
+            ((1043, 269484032, 128, 0, 0), hdf5plugin.Sperr()),
+            (
+                (1107, 2418016256, 256, 0, 0),
+                hdf5plugin.Sperr(rate=32, swap=True, missing_value_mode=1),
+            ),
+            ((1043, 940177214, 256, 0, 0), hdf5plugin.Sperr(absolute=1e-3)),
+            (
+                (1171, 537001984, 256, 0, 0),
+                hdf5plugin.Sperr(peak_signal_to_noise_ratio=2.0, missing_value_mode=2),
+            ),
+        ):
+            with self.subTest(filter_options=filter_options):
+                compression_filter = hdf5plugin.Sperr.from_filter_options(
+                    filter_options
+                )
+                self.assertEqual(
+                    compression_filter.filter_options, expected_filter.filter_options
+                )
+
     def testZstd(self):
         for filter_options, expected_options in (
             # (clevel,)
@@ -643,6 +664,11 @@ def testLZ4(self):
         data = numpy.arange(256**2, dtype=numpy.float32).reshape(256, 256)
         self._test(hdf5plugin.LZ4(), data)
 
+    @unittest.skipUnless(should_test("sperr"), "Sperr filter not available")
+    def testSperr(self):
+        data = numpy.arange(256**2, dtype=numpy.float32).reshape(256, 256)
+        self._test(hdf5plugin.Sperr(), data)
+
     @unittest.skipUnless(should_test("zstd"), "Zstd filter not available")
     def testZstd(self):
         data = numpy.arange(256**2, dtype=numpy.float32).reshape(256, 256)