iqz

A simple lossless compression codec for I/Q radar data.

Each chunk of n chirps (int16, any per-chirp shape) is encoded by subtracting the chunk's rounded mean chirp, zigzag-encoding and byte-shuffling the residuals, and compressing with zstd; see the format specification for details.

• Functional API: encode, decode, decode_into, info.
• Streaming API: Encoder and Decoder, which re-chunk chirps pushed or read in batches of any size.

Two implementations are provided, and can be selected with the backend argument of each function or class: native (Rust) and numpy. Both produce and accept identical blobs.

iqz.Decoder

Streaming decoder, which re-chunks decoded chirps into any size.

Push blobs with push, each containing any number of chirps, and read chirps back out with read or read_into in batches of any size, regardless of where the chunk boundaries fall.

decoder = iqz.Decoder(shape=(3, 4, 1024))
out = np.empty((16, 3, 4, 1024), dtype=np.int16)
for blob in blobs:
    decoder.push(blob)
    while decoder.available >= 16:
        decoder.read_into(out)  # reuses `out`; no allocation

Blobs are only decoded when read: chunks which fit entirely in a read are decoded directly into the output, and only chunks which are split between reads go through an internal buffer.

Warning

Pushed blobs are referenced, not copied, until they have been read; they must not be modified in the meantime.

Parameters:

Name	Type	Description	Default
shape	tuple[int, ...] | None	per-chirp sample shape, excluding the chirps axis. If None, chirps are flattened to (M,), where M is taken from the first blob.	None
backend	Literal['native', 'numpy'] | None	implementation to use; if None, uses native when the extension is available, and numpy otherwise.	None

Source code in src/iqz/_stream.py

class Decoder:
    """Streaming decoder, which re-chunks decoded chirps into any size.

    Push blobs with [`push`][.], each containing any number of chirps, and
    read chirps back out with [`read`][.] or [`read_into`][.] in batches of
    any size, regardless of where the chunk boundaries fall.

    ```python
    decoder = iqz.Decoder(shape=(3, 4, 1024))
    out = np.empty((16, 3, 4, 1024), dtype=np.int16)
    for blob in blobs:
        decoder.push(blob)
        while decoder.available >= 16:
            decoder.read_into(out)  # reuses `out`; no allocation
    ```

    Blobs are only decoded when read: chunks which fit entirely in a read
    are decoded directly into the output, and only chunks which are split
    between reads go through an internal buffer.

    !!! warning

        Pushed blobs are referenced, not copied, until they have been read;
        they must not be modified in the meantime.

    Args:
        shape: per-chirp sample shape, excluding the chirps axis. If `None`,
            chirps are flattened to `(M,)`, where `M` is taken from the first
            blob.
        backend: implementation to use; if `None`, uses `native` when the
            extension is available, and `numpy` otherwise.
    """

    def __init__(
        self, shape: tuple[int, ...] | None = None, *,
        backend: Literal["native", "numpy"] | None = None
    ) -> None:
        _api._impl(backend)
        self.backend: Literal["native", "numpy"] | None = backend
        self._shape = None if shape is None else tuple(shape)
        self._blobs: deque[tuple[Buffer, int]] = deque()
        self._available = 0
        # Decoded head chunk, if it was split by a previous read.
        self._scratch = np.empty(0, dtype=np.int16)
        self._head: Int16[np.ndarray, "n M"] | None = None
        self._offset = 0

    @property
    def shape(self) -> tuple[int, ...] | None:
        """Per-chirp sample shape, or `None` if not yet known."""
        return self._shape

    @property
    def available(self) -> int:
        """Number of chirps pushed which have not yet been read."""
        return self._available

    def push(self, blob: Buffer) -> None:
        """Queue a blob for reading.

        Only the blob's header is checked here; it is decoded when read.

        Args:
            blob: encoded chunk with any number of chirps.

        Raises:
            ValueError: if the blob's header is malformed, or does not match
                the stream's shape.
        """
        n, M = _api.info(blob)
        self._shape = _api._sample_shape(self._shape, M)
        self._blobs.append((blob, n))
        self._available += n

    def read_into(
        self, out: Int16[np.ndarray, "chirps *sample"]
    ) -> Int16[np.ndarray, "chirps *sample"]:
        """Read the next `len(out)` chirps into a buffer provided by the caller.

        Args:
            out: writable, C-contiguous int16 array of shape `(k, *shape)`.

        Returns:
            `out`.

        Raises:
            TypeError: if `out` is not an int16 array.
            ValueError: if fewer than `k` chirps are available, `out` has
                the wrong shape, or a blob fails to decode.
        """
        if not isinstance(out, np.ndarray) or out.dtype != np.int16:
            raise TypeError("Output must be an int16 array.")
        if out.ndim < 1 or out.shape[0] > self._available:
            raise ValueError(
                f"Cannot read {out.shape[0] if out.ndim else '?'} chirps; "
                f"{self._available} available.")
        if self._shape is not None and out.shape[1:] != self._shape:
            raise ValueError(
                f"Output shape {out.shape[1:]} (excluding chirps) does not "
                f"match the stream's shape {self._shape}.")
        if not out.flags.c_contiguous or not out.flags.writeable:
            raise ValueError("Output must be writable and C-contiguous.")

        k = out.shape[0]
        flat = out.reshape(k, -1)
        pos = 0
        while pos < k:
            blob, n = self._blobs[0]
            if self._head is None and n <= k - pos:
                # Whole chunk: decode straight into the output.
                _api.decode_into(
                    blob, flat[pos:pos + n], backend=self.backend)
                pos += n
                self._blobs.popleft()
                continue
            if self._head is None:
                M = flat.shape[1]
                if self._scratch.size < n * M:
                    self._scratch = np.empty(n * M, dtype=np.int16)
                head = self._scratch[:n * M].reshape(n, M)
                _api.decode_into(blob, head, backend=self.backend)
                self._head = head
            take = min(n - self._offset, k - pos)
            flat[pos:pos + take] = self._head[self._offset:self._offset + take]
            pos += take
            self._offset += take
            if self._offset == n:
                self._blobs.popleft()
                self._head = None
                self._offset = 0
        self._available -= k
        return out

    @overload
    def read(
        self, k: int | None = None, *, format: Literal["numpy"] = "numpy"
    ) -> Int16[np.ndarray, "chirps *sample"]: ...

    @overload
    def read(
        self, k: int | None = None, *, format: Literal["bytes"]
    ) -> bytes: ...

    def read(
        self, k: int | None = None, *,
        format: Literal["numpy", "bytes"] = "numpy"
    ) -> Int16[np.ndarray, "chirps *sample"] | bytes:
        """Read decoded chirps into a new array.

        Allocates a new output for every read; use [`read_into`][^.] to
        reuse a buffer instead.

        Args:
            k: number of chirps to read; if `None`, read all available.
            format: `numpy` to return an int16 array of shape `(k, *shape)`,
                or `bytes` to return the raw little-endian int16 samples.

        Returns:
            `k` chirps.

        Raises:
            ValueError: if fewer than `k` chirps are available, or a blob
                fails to decode.
        """
        if format not in ("numpy", "bytes"):
            raise ValueError(f"Unknown format: {format!r}")
        count = self._available if k is None else k
        if count < 0 or count > self._available:
            raise ValueError(
                f"Cannot read {count} chirps; {self._available} available.")
        out = np.empty((count, *(self._shape or (0,))), dtype=np.int16)
        if count > 0:
            self.read_into(out)
        return out if format == "numpy" else out.tobytes()

shape property

shape: tuple[int, ...] | None

Per-chirp sample shape, or None if not yet known.

available property

available: int

Number of chirps pushed which have not yet been read.

push

push(blob: Buffer) -> None

Queue a blob for reading.

Only the blob's header is checked here; it is decoded when read.

Parameters:

Name	Type	Description	Default
blob	Buffer	encoded chunk with any number of chirps.	required

Raises:

Type	Description
ValueError	if the blob's header is malformed, or does not match the stream's shape.

Source code in src/iqz/_stream.py

def push(self, blob: Buffer) -> None:
    """Queue a blob for reading.

    Only the blob's header is checked here; it is decoded when read.

    Args:
        blob: encoded chunk with any number of chirps.

    Raises:
        ValueError: if the blob's header is malformed, or does not match
            the stream's shape.
    """
    n, M = _api.info(blob)
    self._shape = _api._sample_shape(self._shape, M)
    self._blobs.append((blob, n))
    self._available += n

read_into

read_into(
    out: Int16[ndarray, chirps * sample],
) -> Int16[ndarray, chirps * sample]

Read the next len(out) chirps into a buffer provided by the caller.

Parameters:

Name	Type	Description	Default
out	Int16[ndarray, chirps * sample]	writable, C-contiguous int16 array of shape (k, *shape).	required

Returns:

Type	Description
Int16[ndarray, chirps * sample]	out.

Raises:

Type	Description
TypeError	if out is not an int16 array.
ValueError	if fewer than k chirps are available, out has the wrong shape, or a blob fails to decode.

Source code in src/iqz/_stream.py

def read_into(
    self, out: Int16[np.ndarray, "chirps *sample"]
) -> Int16[np.ndarray, "chirps *sample"]:
    """Read the next `len(out)` chirps into a buffer provided by the caller.

    Args:
        out: writable, C-contiguous int16 array of shape `(k, *shape)`.

    Returns:
        `out`.

    Raises:
        TypeError: if `out` is not an int16 array.
        ValueError: if fewer than `k` chirps are available, `out` has
            the wrong shape, or a blob fails to decode.
    """
    if not isinstance(out, np.ndarray) or out.dtype != np.int16:
        raise TypeError("Output must be an int16 array.")
    if out.ndim < 1 or out.shape[0] > self._available:
        raise ValueError(
            f"Cannot read {out.shape[0] if out.ndim else '?'} chirps; "
            f"{self._available} available.")
    if self._shape is not None and out.shape[1:] != self._shape:
        raise ValueError(
            f"Output shape {out.shape[1:]} (excluding chirps) does not "
            f"match the stream's shape {self._shape}.")
    if not out.flags.c_contiguous or not out.flags.writeable:
        raise ValueError("Output must be writable and C-contiguous.")

    k = out.shape[0]
    flat = out.reshape(k, -1)
    pos = 0
    while pos < k:
        blob, n = self._blobs[0]
        if self._head is None and n <= k - pos:
            # Whole chunk: decode straight into the output.
            _api.decode_into(
                blob, flat[pos:pos + n], backend=self.backend)
            pos += n
            self._blobs.popleft()
            continue
        if self._head is None:
            M = flat.shape[1]
            if self._scratch.size < n * M:
                self._scratch = np.empty(n * M, dtype=np.int16)
            head = self._scratch[:n * M].reshape(n, M)
            _api.decode_into(blob, head, backend=self.backend)
            self._head = head
        take = min(n - self._offset, k - pos)
        flat[pos:pos + take] = self._head[self._offset:self._offset + take]
        pos += take
        self._offset += take
        if self._offset == n:
            self._blobs.popleft()
            self._head = None
            self._offset = 0
    self._available -= k
    return out

read

read(
    k: int | None = None, *, format: Literal["numpy"] = "numpy"
) -> Int16[ndarray, chirps * sample]

read(k: int | None = None, *, format: Literal['bytes']) -> bytes

read(
    k: int | None = None, *, format: Literal["numpy", "bytes"] = "numpy"
) -> Int16[ndarray, chirps * sample] | bytes

Read decoded chirps into a new array.

Allocates a new output for every read; use read_into to reuse a buffer instead.

Parameters:

Name	Type	Description	Default
k	int | None	number of chirps to read; if None, read all available.	None
format	Literal['numpy', 'bytes']	numpy to return an int16 array of shape (k, *shape), or bytes to return the raw little-endian int16 samples.	'numpy'

Returns:

Type	Description
Int16[ndarray, chirps * sample] | bytes	k chirps.

Raises:

Type	Description
ValueError	if fewer than k chirps are available, or a blob fails to decode.

Source code in src/iqz/_stream.py

def read(
    self, k: int | None = None, *,
    format: Literal["numpy", "bytes"] = "numpy"
) -> Int16[np.ndarray, "chirps *sample"] | bytes:
    """Read decoded chirps into a new array.

    Allocates a new output for every read; use [`read_into`][^.] to
    reuse a buffer instead.

    Args:
        k: number of chirps to read; if `None`, read all available.
        format: `numpy` to return an int16 array of shape `(k, *shape)`,
            or `bytes` to return the raw little-endian int16 samples.

    Returns:
        `k` chirps.

    Raises:
        ValueError: if fewer than `k` chirps are available, or a blob
            fails to decode.
    """
    if format not in ("numpy", "bytes"):
        raise ValueError(f"Unknown format: {format!r}")
    count = self._available if k is None else k
    if count < 0 or count > self._available:
        raise ValueError(
            f"Cannot read {count} chirps; {self._available} available.")
    out = np.empty((count, *(self._shape or (0,))), dtype=np.int16)
    if count > 0:
        self.read_into(out)
    return out if format == "numpy" else out.tobytes()

iqz.Encoder

Streaming encoder, which re-chunks chirps into fixed-size windows.

Chirps can be pushed in batches of any size; each completed window of window chirps is emitted as one blob. Call flush at the end of the stream to encode any remaining chirps as a final, shorter chunk.

encoder = iqz.Encoder(window=64)
blobs = []
for frame in frames:  # e.g. (1, 3, 4, 1024) int16 each
    blobs.extend(encoder.push(frame))
blobs.extend(encoder.flush())

Parameters:

Name	Type	Description	Default
window	int	number of chirps per encoded chunk.	64
shape	tuple[int, ...] | None	per-chirp sample shape, excluding the chirps axis. If None, this is taken from the first array pushed; buffer input requires the shape to be known.	None
level	int	zstd compression level.	3
backend	Literal['native', 'numpy'] | None	implementation to use; if None, uses native when the extension is available, and numpy otherwise.	None

Source code in src/iqz/_stream.py

class Encoder:
    """Streaming encoder, which re-chunks chirps into fixed-size windows.

    Chirps can be pushed in batches of any size; each completed window of
    `window` chirps is emitted as one blob. Call [`flush`][.] at the end of
    the stream to encode any remaining chirps as a final, shorter chunk.

    ```python
    encoder = iqz.Encoder(window=64)
    blobs = []
    for frame in frames:  # e.g. (1, 3, 4, 1024) int16 each
        blobs.extend(encoder.push(frame))
    blobs.extend(encoder.flush())
    ```

    Args:
        window: number of chirps per encoded chunk.
        shape: per-chirp sample shape, excluding the chirps axis. If `None`,
            this is taken from the first array pushed; buffer input requires
            the shape to be known.
        level: zstd compression level.
        backend: implementation to use; if `None`, uses `native` when the
            extension is available, and `numpy` otherwise.
    """

    def __init__(
        self, window: int = 64, shape: tuple[int, ...] | None = None, *,
        level: int = 3, backend: Literal["native", "numpy"] | None = None
    ) -> None:
        if window < 1:
            raise ValueError(f"`window` must be positive; got {window}.")
        self._impl = _api._impl(backend)
        self.window = window
        self.level = level
        self._shape = None if shape is None else tuple(shape)
        self._stage: Int16[np.ndarray, "window M"] | None = None
        self._fill = 0

    @property
    def shape(self) -> tuple[int, ...] | None:
        """Per-chirp sample shape, or `None` if not yet known."""
        return self._shape

    @property
    def pending(self) -> int:
        """Number of chirps buffered, but not yet encoded."""
        return self._fill

    def _encode(self, x: Int16[np.ndarray, "n M"]) -> bytes:
        return self._impl.encode_chunk(x, self.level)

    def push(
        self, x: Int16[np.ndarray, "chirps *sample"] | Buffer
    ) -> list[bytes]:
        """Add chirps to the stream.

        Args:
            x: one or more int16 chirps, either as an array with a leading
                chirps axis, or as a buffer of little-endian int16 samples.
                The data is copied (or encoded) before returning, so the
                caller may reuse it.

        Returns:
            Blobs for each window completed by this push (possibly none).

        Raises:
            TypeError: if `x` is an array, but not int16.
            ValueError: if `x` does not match the stream's shape, or is a
                buffer pushed before the shape is known.
        """
        flat, shape = _api._as_chunk(x, self._shape)
        if self._stage is None:
            self._shape = shape
            self._stage = np.empty(
                (self.window, math.prod(shape)), dtype=np.int16)
        stage = self._stage

        blobs = []
        i, k = 0, flat.shape[0]
        if self._fill > 0:
            take = min(self.window - self._fill, k)
            stage[self._fill:self._fill + take] = flat[:take]
            self._fill += take
            i = take
            if self._fill == self.window:
                blobs.append(self._encode(stage))
                self._fill = 0

        while k - i >= self.window:
            blobs.append(self._encode(flat[i:i + self.window]))
            i += self.window

        if i < k:
            stage[:k - i] = flat[i:]
            self._fill = k - i
        return blobs

    def flush(self) -> list[bytes]:
        """Encode any buffered chirps as a final (shorter) chunk.

        The encoder can be reused for a new stream (with the same shape)
        afterwards.

        Returns:
            One blob if any chirps were pending; otherwise, an empty list.
        """
        if self._fill == 0 or self._stage is None:
            return []
        blob = self._encode(self._stage[:self._fill])
        self._fill = 0
        return [blob]

shape property

shape: tuple[int, ...] | None

Per-chirp sample shape, or None if not yet known.

pending property

pending: int

Number of chirps buffered, but not yet encoded.

push

push(x: Int16[ndarray, chirps * sample] | Buffer) -> list[bytes]

Add chirps to the stream.

Parameters:

Name	Type	Description	Default
x	Int16[ndarray, chirps * sample] | Buffer	one or more int16 chirps, either as an array with a leading chirps axis, or as a buffer of little-endian int16 samples. The data is copied (or encoded) before returning, so the caller may reuse it.	required

Returns:

Type	Description
list[bytes]	Blobs for each window completed by this push (possibly none).

Raises:

Type	Description
TypeError	if x is an array, but not int16.
ValueError	if x does not match the stream's shape, or is a buffer pushed before the shape is known.

Source code in src/iqz/_stream.py

def push(
    self, x: Int16[np.ndarray, "chirps *sample"] | Buffer
) -> list[bytes]:
    """Add chirps to the stream.

    Args:
        x: one or more int16 chirps, either as an array with a leading
            chirps axis, or as a buffer of little-endian int16 samples.
            The data is copied (or encoded) before returning, so the
            caller may reuse it.

    Returns:
        Blobs for each window completed by this push (possibly none).

    Raises:
        TypeError: if `x` is an array, but not int16.
        ValueError: if `x` does not match the stream's shape, or is a
            buffer pushed before the shape is known.
    """
    flat, shape = _api._as_chunk(x, self._shape)
    if self._stage is None:
        self._shape = shape
        self._stage = np.empty(
            (self.window, math.prod(shape)), dtype=np.int16)
    stage = self._stage

    blobs = []
    i, k = 0, flat.shape[0]
    if self._fill > 0:
        take = min(self.window - self._fill, k)
        stage[self._fill:self._fill + take] = flat[:take]
        self._fill += take
        i = take
        if self._fill == self.window:
            blobs.append(self._encode(stage))
            self._fill = 0

    while k - i >= self.window:
        blobs.append(self._encode(flat[i:i + self.window]))
        i += self.window

    if i < k:
        stage[:k - i] = flat[i:]
        self._fill = k - i
    return blobs

flush

flush() -> list[bytes]

Encode any buffered chirps as a final (shorter) chunk.

The encoder can be reused for a new stream (with the same shape) afterwards.

Returns:

Type	Description
list[bytes]	One blob if any chirps were pending; otherwise, an empty list.

Source code in src/iqz/_stream.py

def flush(self) -> list[bytes]:
    """Encode any buffered chirps as a final (shorter) chunk.

    The encoder can be reused for a new stream (with the same shape)
    afterwards.

    Returns:
        One blob if any chirps were pending; otherwise, an empty list.
    """
    if self._fill == 0 or self._stage is None:
        return []
    blob = self._encode(self._stage[:self._fill])
    self._fill = 0
    return [blob]

iqz.decode

decode(
    blob: Buffer,
    shape: tuple[int, ...] | None = None,
    *,
    format: Literal["numpy"] = "numpy",
    backend: Literal["native", "numpy"] | None = None,
) -> Int16[ndarray, chirps * sample]

decode(
    blob: Buffer,
    shape: tuple[int, ...] | None = None,
    *,
    format: Literal["bytes"],
    backend: Literal["native", "numpy"] | None = None,
) -> bytes

decode(
    blob: Buffer,
    shape: tuple[int, ...] | None = None,
    *,
    format: Literal["numpy", "bytes"] = "numpy",
    backend: Literal["native", "numpy"] | None = None,
) -> Int16[ndarray, chirps * sample] | bytes

Decode one chunk.

Parameters:

Name	Type	Description	Default
blob	Buffer	encoded chunk; any buffer, e.g. bytes, memoryview, or a slice of an mmap.	required
shape	tuple[int, ...] | None	per-chirp sample shape, excluding the chirps axis. If None, chirps are returned flattened, with shape (M,).	None
format	Literal['numpy', 'bytes']	numpy to return an int16 array of shape (n, *shape), or bytes to return the raw little-endian int16 samples.	'numpy'
backend	Literal['native', 'numpy'] | None	implementation to use; if None, uses native when the extension is available, and numpy otherwise.	None

Returns:

Type	Description
Int16[ndarray, chirps * sample] | bytes	Decoded chirps.

Raises:

Type	Description
ValueError	if the blob is malformed, or does not match shape.

Source code in src/iqz/_api.py

def decode(
    blob: Buffer, shape: tuple[int, ...] | None = None, *,
    format: Literal["numpy", "bytes"] = "numpy",
    backend: Literal["native", "numpy"] | None = None
) -> Int16[np.ndarray, "chirps *sample"] | bytes:
    """Decode one chunk.

    Args:
        blob: encoded chunk; any buffer, e.g. `bytes`, `memoryview`, or a
            slice of an `mmap`.
        shape: per-chirp sample shape, excluding the chirps axis. If `None`,
            chirps are returned flattened, with shape `(M,)`.
        format: `numpy` to return an int16 array of shape `(n, *shape)`, or
            `bytes` to return the raw little-endian int16 samples.
        backend: implementation to use; if `None`, uses `native` when the
            extension is available, and `numpy` otherwise.

    Returns:
        Decoded chirps.

    Raises:
        ValueError: if the blob is malformed, or does not match `shape`.
    """
    if format not in ("numpy", "bytes"):
        raise ValueError(f"Unknown format: {format!r}")
    n, M = info(blob)
    out = np.empty((n, *_sample_shape(shape, M)), dtype=np.int16)
    _decode(blob, out.reshape(n, M), backend)
    return out if format == "numpy" else out.tobytes()

iqz.decode_into

decode_into(
    blob: Buffer,
    out: Int16[ndarray, chirps * sample],
    *,
    backend: Literal["native", "numpy"] | None = None,
) -> Int16[ndarray, chirps * sample]

decode_into(
    blob: Buffer,
    out: Buffer,
    *,
    backend: Literal["native", "numpy"] | None = None,
) -> memoryview

decode_into(
    blob: Buffer,
    out: Int16[ndarray, chirps * sample] | Buffer,
    *,
    backend: Literal["native", "numpy"] | None = None,
) -> Int16[ndarray, chirps * sample] | memoryview

Decode one chunk into a buffer provided by the caller.

Parameters:

Name	Type	Description	Default
blob	Buffer	encoded chunk; any buffer, e.g. bytes, memoryview, or a slice of an mmap.	required
out	Int16[ndarray, chirps * sample] | Buffer	destination; either a writable, C-contiguous int16 array with exactly n * M elements (of any shape), or any writable, C-contiguous buffer of exactly 2 * n * M bytes, which receives little-endian int16 samples.	required
backend	Literal['native', 'numpy'] | None	implementation to use; if None, uses native when the extension is available, and numpy otherwise.	None

Returns:

Type	Description
Int16[ndarray, chirps * sample] | memoryview	out itself if it is an array; otherwise, a byte memoryview of out.

Raises:

Type	Description
TypeError	if out is an array, but not int16.
ValueError	if the blob is malformed, or out has the wrong size, is read-only, or is not C-contiguous.

Source code in src/iqz/_api.py

def decode_into(
    blob: Buffer, out: Int16[np.ndarray, "chirps *sample"] | Buffer, *,
    backend: Literal["native", "numpy"] | None = None
) -> Int16[np.ndarray, "chirps *sample"] | memoryview:
    """Decode one chunk into a buffer provided by the caller.

    Args:
        blob: encoded chunk; any buffer, e.g. `bytes`, `memoryview`, or a
            slice of an `mmap`.
        out: destination; either a writable, C-contiguous int16 array with
            exactly `n * M` elements (of any shape), or any writable,
            C-contiguous buffer of exactly `2 * n * M` bytes, which receives
            little-endian int16 samples.
        backend: implementation to use; if `None`, uses `native` when the
            extension is available, and `numpy` otherwise.

    Returns:
        `out` itself if it is an array; otherwise, a byte `memoryview` of
            `out`.

    Raises:
        TypeError: if `out` is an array, but not int16.
        ValueError: if the blob is malformed, or `out` has the wrong size,
            is read-only, or is not C-contiguous.
    """
    n, M = info(blob)
    if isinstance(out, np.ndarray):
        if out.dtype != np.int16:
            raise TypeError(f"Expected an int16 array; got {out.dtype}.")
        if out.size != n * M:
            raise ValueError(
                f"Output has {out.size} elements; the blob decodes to "
                f"{n} x {M} = {n * M}.")
        if not out.flags.c_contiguous or not out.flags.writeable:
            raise ValueError("Output must be writable and C-contiguous.")
        _decode(blob, out.reshape(n, M), backend)
        return out
    else:
        mv = memoryview(out)
        if mv.readonly or not mv.c_contiguous:
            raise ValueError("Output must be writable and C-contiguous.")
        mv = mv.cast("B")
        if mv.nbytes != 2 * n * M:
            raise ValueError(
                f"Output has {mv.nbytes} bytes; the blob decodes to "
                f"{n} x {M} x 2 = {2 * n * M}.")
        _decode(blob, np.frombuffer(mv, np.int16).reshape(n, M), backend)
        return mv

iqz.encode

encode(
    x: Int16[ndarray, chirps * sample] | Buffer,
    shape: tuple[int, ...] | None = None,
    *,
    level: int = 3,
    backend: Literal["native", "numpy"] | None = None,
) -> bytes

Encode one chunk of chirps.

Info

The codec is designed for chunks of 64 chirps; any n >= 1 works, though shorter chunks compress worse.

Parameters:

Name	Type	Description	Default
x	Int16[ndarray, chirps * sample] | Buffer	int16 chirps, either as an array with a leading chirps axis, or as any buffer of little-endian int16 samples (with shape).	required
shape	tuple[int, ...] | None	per-chirp sample shape, excluding the leading chirps axis. If x is an array, this is optional, and is checked against x.shape[1:] when provided. Required if x is a buffer.	None
level	int	zstd compression level.	3
backend	Literal['native', 'numpy'] | None	implementation to use; if None, uses native when the extension is available, and numpy otherwise.	None

Returns:

Type	Description
bytes	Encoded blob.

Raises:

Type	Description
TypeError	if x is an array, but not int16.
ValueError	if x is empty, its shape does not match shape, or shape is missing for buffer input.

Source code in src/iqz/_api.py

def encode(
    x: Int16[np.ndarray, "chirps *sample"] | Buffer,
    shape: tuple[int, ...] | None = None, *,
    level: int = 3, backend: Literal["native", "numpy"] | None = None
) -> bytes:
    """Encode one chunk of chirps.

    !!! info

        The codec is designed for chunks of 64 chirps; any `n >= 1` works,
        though shorter chunks compress worse.

    Args:
        x: int16 chirps, either as an array with a leading chirps axis, or as
            any buffer of little-endian int16 samples (with `shape`).
        shape: per-chirp sample shape, excluding the leading chirps axis. If
            `x` is an array, this is optional, and is checked against
            `x.shape[1:]` when provided. Required if `x` is a buffer.
        level: zstd compression level.
        backend: implementation to use; if `None`, uses `native` when the
            extension is available, and `numpy` otherwise.

    Returns:
        Encoded blob.

    Raises:
        TypeError: if `x` is an array, but not int16.
        ValueError: if `x` is empty, its shape does not match `shape`, or
            `shape` is missing for buffer input.
    """
    flat, _ = _as_chunk(x, shape)
    return _impl(backend).encode_chunk(flat, level)

iqz.info

info(blob: Buffer) -> tuple[int, int]

Get the shape of an encoded chunk.

The shape is recovered from the content sizes recorded in the blob's two zstd frame headers; nothing is decompressed.

Parameters:

Name	Type	Description	Default
blob	Buffer	encoded chunk.	required

Returns:

Type	Description
tuple[int, int]	(n, M): the number of chirps, and the number of samples per chirp.

Raises:

Type	Description
ValueError	if the blob is malformed.

Source code in src/iqz/_api.py

def info(blob: Buffer) -> tuple[int, int]:
    """Get the shape of an encoded chunk.

    The shape is recovered from the content sizes recorded in the blob's two
    zstd frame headers; nothing is decompressed.

    Args:
        blob: encoded chunk.

    Returns:
        `(n, M)`: the number of chirps, and the number of samples per chirp.

    Raises:
        ValueError: if the blob is malformed.
    """
    parts = _format.IQZBlob.from_buffer(blob)
    return parts.chirps, parts.samples