WIP: add from options function and class methods

Author: t20100

src/hdf5plugin/__init__.py

@@ -52,7 +52,13 @@
     Zfp,
     Zstd,
 )
-from ._utils import PLUGIN_PATH, get_config, get_filters, register  # noqa
+from ._utils import (  # noqa
+    PLUGIN_PATH,
+    filter_from_options,
+    get_config,
+    get_filters,
+    register,
+)
 from ._version import version  # noqa
 
 # Backward compatibility

src/hdf5plugin/_filters.py

@@ -74,6 +74,14 @@ class FilterBase(h5py.filters.FilterRefBase):
     filter_id: int
     filter_name: str
 
+    @classmethod
+    def from_options(cls, filter_options: tuple[int, ...]) -> FilterBase:
+        """Returns compression arguments from HDF5 compression filters options
+
+        :raises ValueError: Unsupported filter_options
+        """
+        raise NotImplementedError()
+
 
 class Bitshuffle(FilterBase):
     """``h5py.Group.create_dataset``'s compression arguments for using bitshuffle filter.
@@ -148,6 +156,44 @@ def __init__(
         else:
             self.filter_options = (nelems, self.__COMPRESSIONS[cname])
 
+    @classmethod
+    def from_options(cls, filter_options: tuple[int, ...]) -> Bitshuffle:
+        """Returns compression arguments from HDF5 compression filters options
+
+        :param filter_options: Expected format:
+
+          - Zstd: (_, _, _, nelems, compression_id=3, compression_level)
+          - Other: (_, _, _, nelems, compression_id)
+
+        :raises ValueError: Unsupported filter_options
+        """
+        if len(filter_options) not in (5, 6):
+            raise ValueError(
+                f"Expected 5 or 6 values in filter_options, got {len(filter_options)}"
+            )
+
+        nelems = filter_options[3]
+
+        compression_id = filter_options[4]
+        for cname, cid in cls.__COMPRESSIONS.items():
+            if compression_id == cid:
+                break
+        else:
+            raise ValueError(f"Unsupported compression id: {compression_id}")
+
+        if cname == "zstd":
+            if len(filter_options) != 6:
+                raise ValueError(
+                    f"Expected 6 values in filter_options, got {len(filter_options)}"
+                )
+            return cls(nelems, cname, clevel=filter_options[5])
+        else:
+            if len(filter_options) != 5:
+                raise ValueError(
+                    f"Expected 5 values in filter_options, got {len(filter_options)}"
+                )
+            return cls(nelems, cname)
+
 
 class Blosc(FilterBase):
     """``h5py.Group.create_dataset``'s compression arguments for using blosc filter.
@@ -204,6 +250,31 @@ def __init__(self, cname: str = "lz4", clevel: int = 5, shuffle: int = SHUFFLE):
             raise ValueError(f"shuffle={shuffle} is not supported")
         self.filter_options = (0, 0, 0, 0, clevel, shuffle, compression)
 
+    @classmethod
+    def from_options(cls, filter_options: tuple[int, ...]) -> Blosc:
+        """Returns compression arguments from HDF5 compression filters options
+
+        :param filter_options: Expected format: (_, _, _, _, clevel, shuffle, compression)
+        :raises ValueError: Unsupported filter_options
+        """
+        if len(filter_options) != 7:
+            raise ValueError(
+                f"Expected 7 values in filter_options, got {len(filter_options)}"
+            )
+
+        compression_id = filter_options[6]
+        for cname, cid in cls.__COMPRESSIONS.items():
+            if compression_id == cid:
+                break
+        else:
+            raise ValueError(f"Unsupported compression id: {compression_id}")
+
+        return cls(
+            cname,
+            clevel=filter_options[4],
+            shuffle=filter_options[5],
+        )
+
 
 class Blosc2(FilterBase):
     """``h5py.Group.create_dataset``'s compression arguments for using blosc2 filter.
@@ -272,6 +343,31 @@ def __init__(self, cname: str = "blosclz", clevel: int = 5, filters: int = SHUFF
             raise ValueError(f"filters={filters} is not supported")
         self.filter_options = (0, 0, 0, 0, clevel, filters, compression)
 
+    @classmethod
+    def from_options(cls, filter_options: tuple[int, ...]) -> Blosc2:
+        """Returns compression arguments from HDF5 compression filters options
+
+        :param filter_options: Expected format: (_, _, _, _, clevel, filters, compression)
+        :raises ValueError: Unsupported filter_options
+        """
+        if len(filter_options) != 7:
+            raise ValueError(
+                f"Expected 7 values in filter_options, got {len(filter_options)}"
+            )
+
+        compression_id = filter_options[6]
+        for cname, cid in cls.__COMPRESSIONS.items():
+            if compression_id == cid:
+                break
+        else:
+            raise ValueError(f"Unsupported compression id: {compression_id}")
+
+        return cls(
+            cname,
+            clevel=filter_options[4],
+            filters=filter_options[5],
+        )
+
 
 class BZip2(FilterBase):
     """``h5py.Group.create_dataset``'s compression arguments for using BZip2 filter.
@@ -297,6 +393,19 @@ def __init__(self, blocksize: int = 9):
             raise ValueError("blocksize must be in the range [1, 9]")
         self.filter_options = (blocksize,)
 
+    @classmethod
+    def from_options(cls, filter_options: tuple[int, ...]) -> BZip2:
+        """Returns compression arguments from HDF5 compression filters options
+
+        :param filter_options: Expected format: (blocksize,)
+        :raises ValueError: Unsupported filter_options
+        """
+        if len(filter_options) != 1:
+            raise ValueError(
+                f"Expected 1 value in filter_options, got {len(filter_options)}"
+            )
+        return cls(blocksize=filter_options[0])
+
 
 class FciDecomp(FilterBase):
     """``h5py.Group.create_dataset``'s compression arguments for using FciDecomp filter.
@@ -322,6 +431,18 @@ def __init__(self) -> None:
                 "You may need to reinstall hdf5plugin with a recent version of pip, or rebuild it with a newer compiler."
             )
 
+    @classmethod
+    def from_options(cls, filter_options: tuple[int, ...]) -> FciDecomp:
+        """Returns compression arguments from HDF5 compression filters options
+
+        :raises ValueError: Unsupported filter_options
+        """
+        if len(filter_options) != 0:
+            raise ValueError(
+                f"Expected empty filter_options, got {len(filter_options)} values"
+            )
+        return cls()
+
 
 class LZ4(FilterBase):
     """``h5py.Group.create_dataset``'s compression arguments for using lz4 filter.
@@ -348,6 +469,19 @@ def __init__(self, nbytes: int = 0):
             raise ValueError("clevel must be in the range [0, 2113929216]")
         self.filter_options = (nbytes,)
 
+    @classmethod
+    def from_options(cls, filter_options: tuple[int, ...]) -> LZ4:
+        """Returns compression arguments from HDF5 compression filters options
+
+        :param filter_options: Expected format: (nbytes,)
+        :raises ValueError: Unsupported filter_options
+        """
+        if len(filter_options) != 1:
+            raise ValueError(
+                f"Expected 1 value in filter_options, got {len(filter_options)}"
+            )
+        return cls(nbytes=filter_options[0])
+
 
 class Zfp(FilterBase):
     """``h5py.Group.create_dataset``'s compression arguments for using ZFP filter.
@@ -823,6 +957,19 @@ def __init__(self, clevel: int = 3):
             raise ValueError("clevel must be in the range [1, 22]")
         self.filter_options = (clevel,)
 
+    @classmethod
+    def from_options(cls, filter_options: tuple[int, ...]) -> Zstd:
+        """Returns compression arguments from HDF5 compression filters options
+
+        :param filter_options: Expected format: (clevel,)
+        :raises ValueError: Unsupported filter_options
+        """
+        if len(filter_options) != 1:
+            raise ValueError(
+                f"Expected 1 value in filter_options, got {len(filter_options)}"
+            )
+        return cls(clevel=filter_options[0])
+
 
 FILTER_CLASSES: Tuple[type[FilterBase], ...] = (
     Bitshuffle,

src/hdf5plugin/_utils.py

@@ -230,6 +230,20 @@ def get_filters(
     return tuple(filter_classes)
 
 
+def filter_from_options(filter_id: int, filter_options: tuple[int, ...]) -> FilterBase:
+    """Returns corresponding compression filter configuration instance.
+
+    :param filter_id: HDF5 compression filter ID
+    :param filter_options: Compression filter configuration as stored in HDF5 datasets
+    :raises ValueError: Unsupported or invalid filter_id, filter_options combination
+    """
+    for filter_cls in FILTER_CLASSES:
+        if filter_id == filter_cls.filter_id:
+            return filter_cls.from_options(filter_options)
+
+    raise ValueError(f"Unsupported filter id: {filter_id}")
+
+
 def register(
     filters: int | str | tuple[int | str, ...] = tuple(FILTERS.keys()),
     force: bool = True,