Add SZ3.from_filter_options

Author: t20100

src/hdf5plugin/_filters.py

@@ -968,6 +968,30 @@ def __init__(
 
         self.filter_options = compression_opts
 
+    @classmethod
+    def from_filter_options(cls, filter_options: tuple[int, ...]) -> SZ3:
+        if len(filter_options) < 13:
+            raise ValueError(f"Expected 13 values, got {len(filter_options)}")
+
+        sz_mode = filter_options[4]
+        if sz_mode == 0:
+            return cls(
+                absolute=cls.__unpack_float64(filter_options[5], filter_options[6])
+            )
+        if sz_mode == 1:
+            return cls(
+                relative=cls.__unpack_float64(filter_options[7], filter_options[8])
+            )
+        if sz_mode == 2:
+            return cls(
+                norm2=cls.__unpack_float64(filter_options[9], filter_options[10])
+            )
+        if sz_mode == 3:
+            psnr = cls.__unpack_float64(filter_options[11], filter_options[12])
+            return cls(peak_signal_to_noise_ratio=psnr)
+
+        raise ValueError(f"Unsupported sz_mode: {sz_mode}")
+
     @staticmethod
     def __pack_float64(error: float) -> tuple[int, int]:
         # Pack as big-endian IEEE 754 double
@@ -978,6 +1002,13 @@ def __pack_float64(error: float) -> tuple[int, int]:
         low = struct.unpack(">I", packed[4:8])[0]
         return high, low
 
+    @staticmethod
+    def __unpack_float64(high: int, low: int) -> float:
+        # Pack most-significant & least-significant bits
+        packed = struct.pack(">II", high, low)
+        # Unpack as big-endian IEEE 754 double
+        return float(struct.unpack(">d", packed)[0])
+
 
 class Zstd(FilterBase):
     """``h5py.Group.create_dataset``'s compression arguments for using FciDecomp filter.

src/hdf5plugin/test.py

@@ -598,6 +598,35 @@ def testSperr(self):
                     compression_filter.filter_options, expected_filter.filter_options
                 )
 
+    def testSZ3(self):
+        for filter_options, expected_filter in (
+            (
+                (1, 0, 0, 256, 0, 1058682594, 3944497965, 0, 0, 0, 0, 0, 0),
+                hdf5plugin.SZ3(),
+            ),
+            (
+                (1, 0, 0, 256, 0, 1051772663, 2696277389, 0, 0, 0, 0, 0, 0),
+                hdf5plugin.SZ3(absolute=1e-6),
+            ),
+            (
+                (1, 0, 0, 256, 1, 0, 0, 1062232653, 3539053052, 0, 0, 0, 0),
+                hdf5plugin.SZ3(relative=1e-3),
+            ),
+            (
+                (1, 0, 0, 256, 2, 0, 0, 0, 0, 1062232653, 3539053052, 0, 0),
+                hdf5plugin.SZ3(norm2=1e-3),
+            ),
+            (
+                (1, 0, 0, 256, 3, 0, 0, 0, 0, 0, 0, 1062232653, 3539053052),
+                hdf5plugin.SZ3(peak_signal_to_noise_ratio=1e-3),
+            ),
+        ):
+            with self.subTest(filter_options=filter_options):
+                compression_filter = hdf5plugin.SZ3.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,)
@@ -669,6 +698,11 @@ def testSperr(self):
         data = numpy.arange(256**2, dtype=numpy.float32).reshape(256, 256)
         self._test(hdf5plugin.Sperr(), data)
 
+    @unittest.skipUnless(should_test("sz3"), "SZ3 filter not available")
+    def testSZ3(self):
+        data = numpy.arange(256**2, dtype=numpy.float32).reshape(256, 256)
+        self._test(hdf5plugin.SZ3(), data)
+
     @unittest.skipUnless(should_test("zstd"), "Zstd filter not available")
     def testZstd(self):
         data = numpy.arange(256**2, dtype=numpy.float32).reshape(256, 256)