A small script to obtain the information of all the available compression plugins on hdf5 plugin package

extract_plugging_info_hdf5.py

import hdf5plugin
import h5py
import numpy as np
from hdmf.backends.hdf5.h5_utils import H5DataIO
from pynwb.testing.mock.file import mock_NWBFile
from pathlib import Path
from pynwb import NWBHDF5IO
from pynwb.file import TimeSeries
from neuroconv.tools.nwb_helpers import AVAILABLE_HDF5_COMPRESSION_METHODS

def test_compression_method(method_name):
    """Test a specific compression method and return filter information."""
    print(f"\n{'='*60}")
    print(f"Testing compression method: {method_name}")
    print(f"{'='*60}")
    
    try:
        # Create test data
        data = np.random.rand(100, 10).astype(np.float32)
        
        # Wrap data with H5DataIO using the compression method
        compression_class = AVAILABLE_HDF5_COMPRESSION_METHODS[method_name]
        
        if isinstance(compression_class, str):
            compression_instance = dict(compression=method_name, compression_opts=None)
        else:
            compression_instance = compression_class(**dict())

        wrapped_data = H5DataIO(
            data=data,
            **compression_instance,
            allow_plugin_filters=True,
        )
        
        # Create TimeSeries with compressed data
        test_ts = TimeSeries(
            name="test_compressed_timeseries",
            data=wrapped_data,
            unit="SIunit",
            starting_time=0.0,
            rate=10.0,
        )
        
        # Create NWB file
        nwbfile = mock_NWBFile()
        nwbfile.add_acquisition(test_ts)
        
        # Write to file
        nwbfile_path = Path(f"./test_{method_name}.nwb")
        with NWBHDF5IO(nwbfile_path, "w") as io:
            io.write(nwbfile)
        
        # Read back and analyze compression
        with h5py.File(nwbfile_path, "r") as file:
            dataset = file["/acquisition/test_compressed_timeseries/data"]
            
            print(f"Dataset compression: {dataset.compression}")
            print(f"Dataset compression_opts: {dataset.compression_opts}")
            print(f"Dataset chunks: {dataset.chunks}")
            
            # Get filter information from the dataset creation property list
            dcpl = dataset.id.get_create_plist()
            nfilters = dcpl.get_nfilters()
            print(f"Number of filters: {nfilters}")
            
            for i in range(nfilters):
                filter_info = dcpl.get_filter(i)
                print(f"Filter {i} info: {filter_info}")
        
        # Clean up
        nwbfile_path.unlink()
        
        return True
        
    except Exception as e:
        print(f"ERROR testing {method_name}: {e}")
        return False

def main():
    """Test all available HDF5 compression methods."""
    print("Testing all available HDF5 compression methods...")
    print(f"Available methods: {list(AVAILABLE_HDF5_COMPRESSION_METHODS.keys())}")
    
    for method_name in AVAILABLE_HDF5_COMPRESSION_METHODS.keys():
        test_compression_method(method_name)

if __name__ == "__main__":
    main()