diff --git a/src/hdf5plugin/_filters.py b/src/hdf5plugin/_filters.py index 3f6e3c3a..3604b0ec 100644 --- a/src/hdf5plugin/_filters.py +++ b/src/hdf5plugin/_filters.py @@ -75,7 +75,7 @@ class FilterBase(h5py.filters.FilterRefBase): filter_id: int filter_name: str - def _init( + def __init__( self, filter_options: tuple[int, ...] = (), config: Mapping[str, int | float | bool | str] | None = None, @@ -189,7 +189,7 @@ def __init__( if cname == "zstd": filter_options += (clevel,) config["clevel"] = clevel - self._init(filter_options, config) + super().__init__(filter_options, config) @property def nelems(self) -> int: @@ -294,7 +294,7 @@ def __init__( if shuffle not in (self.NOSHUFFLE, self.SHUFFLE, self.BITSHUFFLE): raise ValueError(f"shuffle={shuffle} is not supported") - self._init( + super().__init__( filter_options=(0, 0, 0, 0, clevel, shuffle, compression), config={"cname": cname, "clevel": clevel, "shuffle": shuffle}, ) @@ -411,7 +411,7 @@ def __init__( self.TRUNC_PREC, ): raise ValueError(f"filters={filters} is not supported") - self._init( + super().__init__( filter_options=(0, 0, 0, 0, clevel, filters, compression), config={"cname": cname, "clevel": clevel, "filters": filters}, ) @@ -480,7 +480,7 @@ def __init__(self, blocksize: int = 9): if not 1 <= blocksize <= 9: raise ValueError("blocksize must be in the range [1, 9]") - self._init( + super().__init__( filter_options=(blocksize,), config={"blocksize": blocksize}, ) @@ -525,7 +525,7 @@ def __init__(self) -> None: "The FciDecomp filter is not available as hdf5plugin was not built with C++11.\n" "You may need to reinstall hdf5plugin with a recent version of pip, or rebuild it with a newer compiler." ) - self._init(filter_options=(), config={}) + super().__init__(filter_options=(), config={}) @classmethod def _from_filter_options(cls, filter_options: tuple[int, ...]) -> FciDecomp: @@ -559,7 +559,7 @@ def __init__(self, nbytes: int = 0): nbytes = int(nbytes) if not 0 <= nbytes <= 0x7E000000: raise ValueError("clevel must be in the range [0, 2113929216]") - self._init( + super().__init__( filter_options=(nbytes,), config={"nbytes": nbytes}, ) @@ -729,7 +729,7 @@ def __init__( filter_options = () config = {} - self._init(filter_options, config) + super().__init__(filter_options, config) # From zfp.h _ZFP_MIN_BITS = 1 # minimum number of bits per block @@ -915,7 +915,7 @@ def __init__( mode = 1 quality = 16 if rate is None else rate - self._init( + super().__init__( filter_options=self.__pack_options(mode, quality, swap, missing_value_mode), config={ mode_name: quality, @@ -1126,7 +1126,7 @@ def __init__( logger.info(f"SZ mode {sz_mode} used.") logger.info(f"filter options {filter_options}") - self._init(filter_options, config) + super().__init__(filter_options, config) @classmethod def _from_filter_options(cls, filter_options: tuple[int, ...]) -> SZ: @@ -1227,7 +1227,7 @@ def __init__( if len(filter_options) != 9: raise IndexError("Invalid number of arguments") - self._init(filter_options, config) + super().__init__(filter_options, config) @classmethod def _from_filter_options(cls, filter_options: tuple[int, ...]) -> SZ3: @@ -1259,7 +1259,7 @@ def _from_filter_options(cls, filter_options: tuple[int, ...]) -> SZ3: class Zstd(FilterBase): - """``h5py.Group.create_dataset``'s compression arguments for using FciDecomp filter. + """``h5py.Group.create_dataset``'s compression arguments for using Zstd filter. .. code-block:: python @@ -1280,7 +1280,7 @@ class Zstd(FilterBase): def __init__(self, clevel: int = 3): if not 1 <= clevel <= 22: raise ValueError("clevel must be in the range [1, 22]") - self._init( + super().__init__( filter_options=(clevel,), config={"clevel": clevel}, ) diff --git a/src/hdf5plugin/_utils.py b/src/hdf5plugin/_utils.py index 6d415e1d..2afac7bd 100644 --- a/src/hdf5plugin/_utils.py +++ b/src/hdf5plugin/_utils.py @@ -197,7 +197,7 @@ def get_config() -> HDF5PluginConfig: def get_filters( filters: int | str | tuple[int | str, ...] = tuple(FILTERS.keys()), -) -> tuple[type[FilterBase], ...]: +) -> tuple[type[h5py.filters.FilterRefBase], ...]: """Returns selected filter classes. By default it returns all filter classes. @@ -232,7 +232,7 @@ def get_filters( def from_filter_options( filter_id: int | str, filter_options: tuple[int, ...] -) -> FilterBase: +) -> h5py.filters.FilterRefBase: """Returns corresponding compression filter configuration instance. .. code-block:: python @@ -281,7 +281,7 @@ def register( status = True for filter_class in filter_classes: - filter_name = filter_class.filter_name + filter_name = cast(FilterBase, filter_class).filter_name if not force and is_filter_available(filter_name) is True: logger.info(f"{filter_name} filter already loaded, skip it.") continue diff --git a/src/hdf5plugin/test.py b/src/hdf5plugin/test.py index 5592c605..0c064f20 100644 --- a/src/hdf5plugin/test.py +++ b/src/hdf5plugin/test.py @@ -31,7 +31,7 @@ import shutil import tempfile import unittest -from typing import Any +from typing import Any, cast import h5py import numpy @@ -702,7 +702,9 @@ class TestFromFilterOptionsRoundtrip(unittest.TestCase): """Test from_filter_options function roundtrip""" def _test( - self, compression_filter: _filters.FilterBase, data: numpy.ndarray[Any, Any] + self, + compression_filter: h5py.filters.FilterRefBase, + data: numpy.ndarray[Any, Any], ): with h5py.File("in_memory", "w", driver="core", backing_store=False) as h5f: h5f.create_dataset( @@ -802,7 +804,8 @@ def testGetConfigRoundtrip(self): filter_instance = filter_class() config = filter_instance.get_config() self.assertIsInstance(config, dict) - self.assertEqual(filter_instance, filter_class(**config)) + cls = cast(type[h5py.filters.FilterRefBase], filter_class) + self.assertEqual(filter_instance, cls(**config)) class TestFilterProperties(unittest.TestCase): @@ -921,6 +924,7 @@ def test_register_single_filter_by_id(self): for filter_name in BUILD_CONFIG.embedded_filters: with self.subTest(name=filter_name): filter_class = hdf5plugin.get_filters(filter_name)[0] + assert filter_class.filter_id is not None status = hdf5plugin.register(filter_class.filter_id, force=True) self.assertTrue(status) self._simple_test(filter_name) @@ -955,7 +959,7 @@ def testSelection(self): """Get selected filters""" tests: dict[ int | str | tuple[int | str, ...], - tuple[type[_filters.FilterBase], ...], + tuple[type[h5py.filters.FilterRefBase], ...], ] = { "blosc": (hdf5plugin.Blosc,), ("blosc", "zfp"): (hdf5plugin.Blosc, hdf5plugin.Zfp),