Codec API Reference¶

This page documents the codec implementations for both Zarr V2 and V3.

AnscombeTransformV3¶

`AnscombeTransformV3` `dataclass` ¶

Bases: ArrayArrayCodec

Zarr v3 codec for Anscombe Transform for photon-limited data.

The codec assumes input data has linear encoding with Poisson noise, typically from photon-limited imaging modalities.

Attributes:

Name	Type	Description
`zero_level`	`int`	Signal level when no photons are recorded.
`conversion_gain`	`float`	Signal intensity increase per photon.
`encoded_dtype`	`str`	Data type for encoded values (default: "uint8").
`decoded_dtype`	`str`	Data type for decoded values (default: "int16").
`is_fixed_size`	`bool`	Whether the codec produces fixed-size output (default: True).

Source code in src/anscombe_transform/codec.py

@dataclass(frozen=True, slots=True)
class AnscombeTransformV3(ArrayArrayCodec):
    """
    Zarr v3 codec for Anscombe Transform for photon-limited data.

    The codec assumes input data has linear encoding with Poisson noise,
    typically from photon-limited imaging modalities.

    Attributes
    ----------
    zero_level : int
        Signal level when no photons are recorded.
    conversion_gain : float
        Signal intensity increase per photon.
    encoded_dtype : str
        Data type for encoded values (default: "uint8").
    decoded_dtype : str
        Data type for decoded values (default: "int16").
    is_fixed_size : bool
        Whether the codec produces fixed-size output (default: True).
    """

    zero_level: int
    conversion_gain: float
    encoded_dtype: str = "uint8"
    decoded_dtype: str = "int16"
    is_fixed_size: bool = True

    @classmethod
    def from_dict(cls, data: dict) -> Self:
        """
        Create codec instance from configuration dictionary.

        Parameters
        ----------
        data : dict
            Configuration dictionary with 'configuration' key containing codec parameters.

        Returns
        -------
        AnscombeTransformV3
            New codec instance.
        """
        config = data.get("configuration", {})
        return cls(
            zero_level=config["zero_level"],
            conversion_gain=config["conversion_gain"],
            encoded_dtype=config.get("encoded_dtype", "uint8"),
            decoded_dtype=config.get("decoded_dtype", "int16"),
        )

    def to_dict(self) -> dict:
        """
        Convert codec to configuration dictionary.

        Returns
        -------
        dict
            Configuration dictionary with codec name and parameters.
        """
        return {
            "name": "anscombe-v1",
            "configuration": {
                "zero_level": self.zero_level,
                "conversion_gain": self.conversion_gain,
                "encoded_dtype": self.encoded_dtype,
                "decoded_dtype": self.decoded_dtype,
            },
        }

    def resolve_metadata(self, chunk_spec: ArraySpec) -> ArraySpec:
        """
        Resolve metadata for encoded output.

        Parameters
        ----------
        chunk_spec : ArraySpec
            Input chunk specification.

        Returns
        -------
        ArraySpec
            Output chunk specification with updated dtype.
        """
        return ArraySpec(
            shape=chunk_spec.shape,
            dtype=parse_dtype(np.dtype(self.encoded_dtype), zarr_format=3),
            fill_value=chunk_spec.fill_value,
            config=chunk_spec.config,
            prototype=chunk_spec.prototype,
        )

    def _encode(self, buf: np.ndarray) -> np.ndarray:
        """
        Encode data synchronously for direct use.

        Parameters
        ----------
        buf : np.ndarray
            Input array to encode.

        Returns
        -------
        np.ndarray
            Encoded array.
        """
        return encode(
            buf,
            conversion_gain=self.conversion_gain,
            zero_level=self.zero_level,
            encoded_dtype=self.encoded_dtype,
        )

    def _decode(self, buf: np.ndarray) -> np.ndarray:
        """
        Decode data synchronously for direct use.

        Parameters
        ----------
        buf : np.ndarray
            Encoded buffer to decode.

        Returns
        -------
        np.ndarray
            Decoded array.
        """
        return decode(
            buf.tobytes(),
            conversion_gain=self.conversion_gain,
            zero_level=self.zero_level,
            encoded_dtype=self.encoded_dtype,
            decoded_dtype=self.decoded_dtype,
        )

    async def _encode_single(
        self,
        chunk_array,
        chunk_spec,
    ):
        """
        Encode a single chunk using Anscombe transform.

        Parameters
        ----------
        chunk_array : NDBuffer
            Input chunk to encode.
        chunk_spec : ArraySpec
            Chunk specification.

        Returns
        -------
        NDBuffer
            Encoded chunk.
        """
        # Convert NDBuffer to numpy array
        data = chunk_array.as_numpy_array()

        # Apply encoding
        encoded = self._encode(data)

        # Return as NDBuffer
        return chunk_array.from_numpy_array(encoded)

    async def _decode_single(
        self,
        chunk_array,
        chunk_spec,
    ):
        """
        Decode a single chunk using inverse Anscombe transform.

        Parameters
        ----------
        chunk_array : NDBuffer
            Encoded chunk to decode.
        chunk_spec : ArraySpec
            Chunk specification.

        Returns
        -------
        NDBuffer
            Decoded chunk.
        """
        # Convert NDBuffer to numpy array
        data = chunk_array.as_numpy_array()

        # Apply decoding
        decoded = self._decode(data)

        # Reshape to original shape
        decoded = decoded.reshape(chunk_spec.shape)

        # Return as NDBuffer
        return chunk_array.from_numpy_array(decoded)

`from_dict(data)` `classmethod` ¶

Create codec instance from configuration dictionary.

Parameters:

Name	Type	Description	Default
`data`	`dict`	Configuration dictionary with 'configuration' key containing codec parameters.	required

Returns:

Type	Description
`AnscombeTransformV3`	New codec instance.

Source code in src/anscombe_transform/codec.py

@classmethod
def from_dict(cls, data: dict) -> Self:
    """
    Create codec instance from configuration dictionary.

    Parameters
    ----------
    data : dict
        Configuration dictionary with 'configuration' key containing codec parameters.

    Returns
    -------
    AnscombeTransformV3
        New codec instance.
    """
    config = data.get("configuration", {})
    return cls(
        zero_level=config["zero_level"],
        conversion_gain=config["conversion_gain"],
        encoded_dtype=config.get("encoded_dtype", "uint8"),
        decoded_dtype=config.get("decoded_dtype", "int16"),
    )

`to_dict()` ¶

Convert codec to configuration dictionary.

Returns:

Type	Description
`dict`	Configuration dictionary with codec name and parameters.

Source code in src/anscombe_transform/codec.py

def to_dict(self) -> dict:
    """
    Convert codec to configuration dictionary.

    Returns
    -------
    dict
        Configuration dictionary with codec name and parameters.
    """
    return {
        "name": "anscombe-v1",
        "configuration": {
            "zero_level": self.zero_level,
            "conversion_gain": self.conversion_gain,
            "encoded_dtype": self.encoded_dtype,
            "decoded_dtype": self.decoded_dtype,
        },
    }

`resolve_metadata(chunk_spec)` ¶

Resolve metadata for encoded output.

Parameters:

Name	Type	Description	Default
`chunk_spec`	`ArraySpec`	Input chunk specification.	required

Returns:

Type	Description
`ArraySpec`	Output chunk specification with updated dtype.

Source code in src/anscombe_transform/codec.py

def resolve_metadata(self, chunk_spec: ArraySpec) -> ArraySpec:
    """
    Resolve metadata for encoded output.

    Parameters
    ----------
    chunk_spec : ArraySpec
        Input chunk specification.

    Returns
    -------
    ArraySpec
        Output chunk specification with updated dtype.
    """
    return ArraySpec(
        shape=chunk_spec.shape,
        dtype=parse_dtype(np.dtype(self.encoded_dtype), zarr_format=3),
        fill_value=chunk_spec.fill_value,
        config=chunk_spec.config,
        prototype=chunk_spec.prototype,
    )

AnscombeTransformV2¶

`AnscombeTransformV2` `dataclass` ¶

Zarr v2 codec for Anscombe Transform for photon-limited data.

The codec assumes input data has linear encoding with Poisson noise, typically from photon-limited imaging modalities.

Attributes:

Name	Type	Description
`codec_id`	`str`	Codec identifier ("anscombe-v1").
`zero_level`	`int`	Signal level when no photons are recorded.
`conversion_gain`	`float`	Signal intensity increase per photon.
`encoded_dtype`	`str`	Data type for encoded values (default: "uint8").
`decoded_dtype`	`str`	Data type for decoded values (default: "int16").

Source code in src/anscombe_transform/codec.py

@dataclass(frozen=True, slots=True)
class AnscombeTransformV2:
    """
    Zarr v2 codec for Anscombe Transform for photon-limited data.

    The codec assumes input data has linear encoding with Poisson noise,
    typically from photon-limited imaging modalities.

    Attributes
    ----------
    codec_id : str
        Codec identifier ("anscombe-v1").
    zero_level : int
        Signal level when no photons are recorded.
    conversion_gain : float
        Signal intensity increase per photon.
    encoded_dtype : str
        Data type for encoded values (default: "uint8").
    decoded_dtype : str
        Data type for decoded values (default: "int16").
    """

    codec_id: ClassVar[Literal["anscombe-v1"]] = "anscombe-v1"
    zero_level: int
    conversion_gain: float
    encoded_dtype: str = "uint8"
    decoded_dtype: str = "int16"

    def encode(self, buf: np.ndarray) -> np.ndarray:
        """
        Encode data using Anscombe transform.

        Parameters
        ----------
        buf : np.ndarray
            Input array to encode.

        Returns
        -------
        np.ndarray
            Encoded array.
        """
        return encode(
            buf,
            conversion_gain=self.conversion_gain,
            zero_level=self.zero_level,
            encoded_dtype=self.encoded_dtype,
        )

    def decode(self, buf: bytes, out: object | None = None) -> np.ndarray:
        """
        Decode data using inverse Anscombe transform.

        Parameters
        ----------
        buf : bytes
            Encoded buffer to decode.
        out : object or None, optional
            Output buffer (unused), by default None.

        Returns
        -------
        np.ndarray
            Decoded array.
        """
        return decode(
            buf,
            conversion_gain=self.conversion_gain,
            zero_level=self.zero_level,
            encoded_dtype=self.encoded_dtype,
            decoded_dtype=self.decoded_dtype,
        )

    def get_config(self) -> AnscomeCodecJSON_V2:
        """
        Get codec configuration dictionary.

        Returns
        -------
        dict
            Configuration dictionary with codec ID and parameters.
        """
        return {
            "id": self.codec_id,
            "zero_level": self.zero_level,
            "conversion_gain": self.conversion_gain,
        }

    @classmethod
    def from_config(cls, config: AnscomeCodecJSON_V2) -> Self:
        """
        Create codec instance from configuration dictionary.

        Parameters
        ----------
        config : dict
            Configuration dictionary with 'zero_level' and 'conversion_gain' keys.

        Returns
        -------
        AnscombeTransformV2
            New codec instance.
        """
        return cls(zero_level=config["zero_level"], conversion_gain=config["conversion_gain"])

`encode(buf)` ¶

Encode data using Anscombe transform.

Parameters:

Name	Type	Description	Default
`buf`	`ndarray`	Input array to encode.	required

Returns:

Type	Description
`ndarray`	Encoded array.

Source code in src/anscombe_transform/codec.py

def encode(self, buf: np.ndarray) -> np.ndarray:
    """
    Encode data using Anscombe transform.

    Parameters
    ----------
    buf : np.ndarray
        Input array to encode.

    Returns
    -------
    np.ndarray
        Encoded array.
    """
    return encode(
        buf,
        conversion_gain=self.conversion_gain,
        zero_level=self.zero_level,
        encoded_dtype=self.encoded_dtype,
    )

`decode(buf, out=None)` ¶

Decode data using inverse Anscombe transform.

Parameters:

Name	Type	Description	Default
`buf`	`bytes`	Encoded buffer to decode.	required
`out`	`object or None`	Output buffer (unused), by default None.	`None`

Returns:

Type	Description
`ndarray`	Decoded array.

Source code in src/anscombe_transform/codec.py

def decode(self, buf: bytes, out: object | None = None) -> np.ndarray:
    """
    Decode data using inverse Anscombe transform.

    Parameters
    ----------
    buf : bytes
        Encoded buffer to decode.
    out : object or None, optional
        Output buffer (unused), by default None.

    Returns
    -------
    np.ndarray
        Decoded array.
    """
    return decode(
        buf,
        conversion_gain=self.conversion_gain,
        zero_level=self.zero_level,
        encoded_dtype=self.encoded_dtype,
        decoded_dtype=self.decoded_dtype,
    )

`get_config()` ¶

Get codec configuration dictionary.

Returns:

Type	Description
`dict`	Configuration dictionary with codec ID and parameters.

Source code in src/anscombe_transform/codec.py

def get_config(self) -> AnscomeCodecJSON_V2:
    """
    Get codec configuration dictionary.

    Returns
    -------
    dict
        Configuration dictionary with codec ID and parameters.
    """
    return {
        "id": self.codec_id,
        "zero_level": self.zero_level,
        "conversion_gain": self.conversion_gain,
    }

`from_config(config)` `classmethod` ¶

Create codec instance from configuration dictionary.

Parameters:

Name	Type	Description	Default
`config`	`dict`	Configuration dictionary with 'zero_level' and 'conversion_gain' keys.	required

Returns:

Type	Description
`AnscombeTransformV2`	New codec instance.

Source code in src/anscombe_transform/codec.py

@classmethod
def from_config(cls, config: AnscomeCodecJSON_V2) -> Self:
    """
    Create codec instance from configuration dictionary.

    Parameters
    ----------
    config : dict
        Configuration dictionary with 'zero_level' and 'conversion_gain' keys.

    Returns
    -------
    AnscombeTransformV2
        New codec instance.
    """
    return cls(zero_level=config["zero_level"], conversion_gain=config["conversion_gain"])

Core Functions¶

encode¶

`encode(buf, *, conversion_gain, zero_level, encoded_dtype)` ¶

Encode an array using the Anscombe transform.

Parameters:

Name	Type	Description	Default
`buf`	`ndarray`	Input array to encode.	required
`conversion_gain`	`float`	Signal intensity increase per photon.	required
`zero_level`	`int`	Signal level when no photons are recorded.	required
`encoded_dtype`	`str`	NumPy dtype string for encoded output.	required

Returns:

Type	Description
`ndarray`	Encoded array with variance-stabilized values.

Source code in src/anscombe_transform/codec.py

def encode(
    buf: np.ndarray, *, conversion_gain: float, zero_level: int, encoded_dtype: str
) -> np.ndarray:
    """
    Encode an array using the Anscombe transform.

    Parameters
    ----------
    buf : np.ndarray
        Input array to encode.
    conversion_gain : float
        Signal intensity increase per photon.
    zero_level : int
        Signal level when no photons are recorded.
    encoded_dtype : str
        NumPy dtype string for encoded output.

    Returns
    -------
    np.ndarray
        Encoded array with variance-stabilized values.
    """
    lut = make_anscombe_lookup(
        conversion_gain,
        output_type=encoded_dtype,
        zero_level=zero_level,
    )
    encoded = lookup(buf, lut)
    return encoded.astype(encoded_dtype)

decode¶

`decode(buf, *, conversion_gain, zero_level, encoded_dtype, decoded_dtype)` ¶

Decode an array using the inverse Anscombe transform.

Parameters:

Name	Type	Description	Default
`buf`	`bytes or ndarray`	Encoded buffer to decode.	required
`conversion_gain`	`float`	Signal intensity increase per photon.	required
`zero_level`	`int`	Signal level when no photons are recorded.	required
`encoded_dtype`	`DtypeLike`	NumPy dtype of encoded data.	required
`decoded_dtype`	`DtypeLike`	NumPy dtype for decoded output.	required

Returns:

Type	Description
`ndarray`	Decoded array with original value scale.

Source code in src/anscombe_transform/codec.py

def decode(
    buf: bytes | np.ndarray,
    *,
    conversion_gain: float,
    zero_level: int,
    encoded_dtype: npt.DtypeLike,
    decoded_dtype: npt.DTypeLike,
) -> np.ndarray:
    """
    Decode an array using the inverse Anscombe transform.

    Parameters
    ----------
    buf : bytes or np.ndarray
        Encoded buffer to decode.
    conversion_gain : float
        Signal intensity increase per photon.
    zero_level : int
        Signal level when no photons are recorded.
    encoded_dtype : numpy.typing.DtypeLike
        NumPy dtype of encoded data.
    decoded_dtype : numpy.typing.DtypeLike
        NumPy dtype for decoded output.

    Returns
    -------
    np.ndarray
        Decoded array with original value scale.
    """
    lookup_table = make_anscombe_lookup(
        conversion_gain,
        output_type=encoded_dtype,
        zero_level=zero_level,
    )
    inverse_table = make_inverse_lookup(lookup_table, output_type=decoded_dtype)
    decoded = np.frombuffer(buf, dtype=encoded_dtype)
    return lookup(decoded, inverse_table).astype(decoded_dtype)

Lookup Table Functions¶

make_anscombe_lookup¶

`make_anscombe_lookup(conversion_gain, input_max=32767, zero_level=0, beta=0.5, output_type='uint8')` ¶

Compute the Anscombe lookup table.

The lookup converts a linear grayscale image into a uniform variance image by applying the Anscombe variance-stabilizing transformation.

Parameters:

Name	Type	Description	Default
`conversion_gain`	`float`	Estimated signal intensity increase per quantum (e.g. photon).	required
`input_max`	`int`	The maximum value in the input data, by default 0x7FFF (32767).	`32767`
`zero_level`	`int`	Signal level when no photons are recorded, by default 0.	`0`
`beta`	`float`	The grayscale quantization step expressed in units of noise std dev, by default 0.5.	`0.5`
`output_type`	`str`	NumPy dtype string for output array, by default "uint8".	`'uint8'`

Returns:

Type	Description
`ndarray`	Lookup table array for Anscombe transformation.

Source code in src/anscombe_transform/codec.py

def make_anscombe_lookup(
    conversion_gain: float,
    input_max: int = 0x7FFF,
    zero_level: int = 0,
    beta: float = 0.5,
    output_type: str = "uint8",
) -> np.ndarray:
    """
    Compute the Anscombe lookup table.

    The lookup converts a linear grayscale image into a uniform variance image
    by applying the Anscombe variance-stabilizing transformation.

    Parameters
    ----------
    conversion_gain : float
        Estimated signal intensity increase per quantum (e.g. photon).
    input_max : int, optional
        The maximum value in the input data, by default 0x7FFF (32767).
    zero_level : int, optional
        Signal level when no photons are recorded, by default 0.
    beta : float, optional
        The grayscale quantization step expressed in units of noise std dev, by default 0.5.
    output_type : str, optional
        NumPy dtype string for output array, by default "uint8".

    Returns
    -------
    np.ndarray
        Lookup table array for Anscombe transformation.
    """
    xx = (np.r_[: input_max + 1] - zero_level) / conversion_gain  # input expressed in photon rates
    zero_slope = 1 / beta / np.sqrt(3 / 8)  # slope for negative values
    offset = zero_level * zero_slope / conversion_gain
    lookup_table = np.round(
        offset
        + (xx < 0) * (xx * zero_slope)
        + (xx >= 0) * (2.0 / beta * (np.sqrt(np.maximum(0, xx) + 3 / 8) - np.sqrt(3 / 8)))
    )
    lookup = lookup_table.astype(output_type)
    assert np.diff(lookup_table).min() >= 0, "non-monotonic lookup generated"
    return lookup

make_inverse_lookup¶

`make_inverse_lookup(lookup_table, output_type='int16')` ¶

Compute the inverse lookup table for a monotonic forward lookup table.

Parameters:

Name	Type	Description	Default
`lookup_table`	`ndarray`	Monotonic forward lookup table.	required
`output_type`	`str`	NumPy dtype string for output array, by default "int16".	`'int16'`

Returns:

Type	Description
`ndarray`	Inverse lookup table that maps encoded values back to original values.

Source code in src/anscombe_transform/codec.py

def make_inverse_lookup(lookup_table: np.ndarray, output_type="int16") -> np.ndarray:
    """
    Compute the inverse lookup table for a monotonic forward lookup table.

    Parameters
    ----------
    lookup_table : np.ndarray
        Monotonic forward lookup table.
    output_type : str, optional
        NumPy dtype string for output array, by default "int16".

    Returns
    -------
    np.ndarray
        Inverse lookup table that maps encoded values back to original values.
    """
    _, inv1 = np.unique(lookup_table, return_index=True)  # first entry
    _, inv2 = np.unique(lookup_table[::-1], return_index=True)  # last entry
    inverse = (inv1 + lookup_table.size - 1 - inv2) / 2
    return inverse.astype(output_type)

lookup¶

`lookup(movie, lookup_table)` ¶

Apply lookup table to movie with boundary clamping.

Parameters:

Name	Type	Description	Default
`movie`	`ndarray`	Input array to transform.	required
`lookup_table`	`ndarray`	Lookup table for transformation.	required

Returns:

Type	Description
`ndarray`	Transformed array with values from lookup table.

Source code in src/anscombe_transform/codec.py

def lookup(movie: np.ndarray, lookup_table: np.ndarray) -> np.ndarray:
    """
    Apply lookup table to movie with boundary clamping.

    Parameters
    ----------
    movie : np.ndarray
        Input array to transform.
    lookup_table : np.ndarray
        Lookup table for transformation.

    Returns
    -------
    np.ndarray
        Transformed array with values from lookup table.
    """
    return lookup_table[np.maximum(0, np.minimum(movie, lookup_table.size - 1))]

Codec API Reference¶

AnscombeTransformV3¶

AnscombeTransformV3 dataclass ¶

from_dict(data) classmethod ¶

to_dict() ¶

resolve_metadata(chunk_spec) ¶

AnscombeTransformV2¶

AnscombeTransformV2 dataclass ¶

encode(buf) ¶

decode(buf, out=None) ¶

get_config() ¶

from_config(config) classmethod ¶

Core Functions¶

encode¶

encode(buf, *, conversion_gain, zero_level, encoded_dtype) ¶

decode¶

decode(buf, *, conversion_gain, zero_level, encoded_dtype, decoded_dtype) ¶

Lookup Table Functions¶

make_anscombe_lookup¶

make_anscombe_lookup(conversion_gain, input_max=32767, zero_level=0, beta=0.5, output_type='uint8') ¶

make_inverse_lookup¶

make_inverse_lookup(lookup_table, output_type='int16') ¶

lookup¶

lookup(movie, lookup_table) ¶

`AnscombeTransformV3` `dataclass` ¶

`from_dict(data)` `classmethod` ¶

`to_dict()` ¶

`resolve_metadata(chunk_spec)` ¶

`AnscombeTransformV2` `dataclass` ¶

`encode(buf)` ¶

`decode(buf, out=None)` ¶

`get_config()` ¶

`from_config(config)` `classmethod` ¶

`encode(buf, *, conversion_gain, zero_level, encoded_dtype)` ¶

`decode(buf, *, conversion_gain, zero_level, encoded_dtype, decoded_dtype)` ¶

`make_anscombe_lookup(conversion_gain, input_max=32767, zero_level=0, beta=0.5, output_type='uint8')` ¶

`make_inverse_lookup(lookup_table, output_type='int16')` ¶

`lookup(movie, lookup_table)` ¶