blosc2.register_codec#

blosc2.register_codec(codec_name: str, id: int, encoder: Callable[[np.ndarray[np.uint8], np.ndarray[np.uint8], int, blosc2.SChunk], int] | None = None, decoder: Callable[[np.ndarray[np.uint8], np.ndarray[np.uint8], int, blosc2.SChunk], int] | None = None, version: int = 1) → None#

Register a user defined codec.

Parameters:

  • codec_name (str) – Name of the codec.

  • id (int) – Codec id, which must be between 160 and 255 (inclusive).

  • encoder (Python function or None) – A Python function that receives an input to compress as a ndarray of dtype uint8, an output to fill the compressed buffer in as a ndarray of dtype uint8, the codec meta and the SChunk instance. It must return the size of the compressed buffer in bytes. If None, the codec name indicates a dynamic plugin that must be installed.

  • decoder (Python function or None) – A Python function that receives an input to decompress as a ndarray of dtype uint8, an output to fill the decompressed buffer in as a ndarray of dtype uint8, the codec meta and the SChunk instance. It must return the size of the decompressed buffer in bytes. If None, then the codec name indicates a dynamic plugin which must be installed.

  • version (int) – The codec version. Default is 1.

Returns:

out

Return type:

None

Notes

  • Cannot use multi-threading when using a user-defined codec.

  • User-defined codecs can only be used inside an SChunk instance.

  • Both encoder and decoder functions must be given (for a Python codec), or none (for a dynamic plugin).

See also

register_filter()

Examples

# Define encoder and decoder functions
def encoder(input, output, meta, schunk):
    # Check whether the data is an arange
    step = int(input[1] - input[0])
    res = input[1:] - input[:-1]
    if np.min(res) == np.max(res):
        output[0:4] = input[0:4]  # start
        n = step.to_bytes(4, sys.byteorder)
        output[4:8] = [n[i] for i in range(4)]
        return 8
    else:
        # Not compressible, tell Blosc2 to do a memcpy
        return 0

def decoder1(input, output, meta, schunk):
    # For decoding we only have to worry about the arange case
    # (other cases are handled by Blosc2)
    output[:] = [input[0] + i * input[1] for i in range(output.size)]

    return output.size

# Register codec
codec_name = "codec1"
id = 180
blosc2.register_codec(codec_name, id, encoder, decoder)
On this page