biopb.image

biopb.image

Ops

Bases: object

A server of named ops over tensors and JSON.

There is no call deadline: a client times out on silence between events, and stops a call by cancelling it. A failure inside a call is the RPC's status: INVALID_ARGUMENT for arguments the op refuses, INTERNAL otherwise.

OpsServicer

Bases: object

A server of named ops over tensors and JSON.

There is no call deadline: a client times out on silence between events, and stops a call by cancelling it. A failure inside a call is the RPC's status: INVALID_ARGUMENT for arguments the op refuses, INTERNAL otherwise.

Call

Call(request, context)

Missing associated documentation comment in .proto file.

Source code in src/main/python/biopb/image/rpc_ops_pb2_grpc.py

def Call(self, request, context):
    """Missing associated documentation comment in .proto file."""
    context.set_code(grpc.StatusCode.UNIMPLEMENTED)
    context.set_details('Method not implemented!')
    raise NotImplementedError('Method not implemented!')

Describe

Describe(request, context)

Missing associated documentation comment in .proto file.

Source code in src/main/python/biopb/image/rpc_ops_pb2_grpc.py

def Describe(self, request, context):
    """Missing associated documentation comment in .proto file."""
    context.set_code(grpc.StatusCode.UNIMPLEMENTED)
    context.set_details('Method not implemented!')
    raise NotImplementedError('Method not implemented!')

OpsStub

OpsStub(channel)

Bases: object

A server of named ops over tensors and JSON.

There is no call deadline: a client times out on silence between events, and stops a call by cancelling it. A failure inside a call is the RPC's status: INVALID_ARGUMENT for arguments the op refuses, INTERNAL otherwise.

Constructor.

Parameters:

Name	Type	Description	Default
channel		A grpc.Channel.	required

Source code in src/main/python/biopb/image/rpc_ops_pb2_grpc.py

def __init__(self, channel):
    """Constructor.

    Args:
        channel: A grpc.Channel.
    """
    self.Call = channel.unary_stream(
            '/biopb.image.Ops/Call',
            request_serializer=biopb_dot_image_dot_rpc__ops__pb2.Call.SerializeToString,
            response_deserializer=biopb_dot_image_dot_rpc__ops__pb2.Event.FromString,
            _registered_method=True)
    self.Describe = channel.unary_unary(
            '/biopb.image.Ops/Describe',
            request_serializer=google_dot_protobuf_dot_empty__pb2.Empty.SerializeToString,
            response_deserializer=biopb_dot_image_dot_rpc__ops__pb2.OpList.FromString,
            _registered_method=True)

deserialize_image_data

deserialize_image_data(
    image_data: ImageData, *, cache_bytes: int = 1000000000
) -> Union[np.ndarray, da.Array]

Convert protobuf ImageData to a numpy array or dask array.

Handles both eager (inline) and lazy (Flight server) data representations. Also handles legacy pixels field for backward compatibility.

Parameters:

Name	Type	Description	Default
image_data	ImageData	protobuf ImageData message	required

Keyword Args: cache_bytes: Cache size for lazy_data chunk cache (default 1GB).

Returns:

Type	Description
Union[ndarray, Array]	numpy array for eager_data, or dask array for lazy_data.

Source code in src/main/python/biopb/image/_utils.py

def deserialize_image_data(
    image_data: ImageData,
    *,
    cache_bytes: int = 1_000_000_000,
) -> Union[np.ndarray, da.Array]:
    """Convert protobuf ImageData to a numpy array or dask array.

    Handles both eager (inline) and lazy (Flight server) data representations.
    Also handles legacy pixels field for backward compatibility.

    Args:
        image_data: protobuf ImageData message
    Keyword Args:
        cache_bytes: Cache size for lazy_data chunk cache (default 1GB).

    Returns:
        numpy array for eager_data, or dask array for lazy_data.
    """
    # Check oneof field
    data_type = image_data.WhichOneof("data")

    if data_type == "eager_data":
        return _np_from_pb(image_data.eager_data)
    elif data_type == "lazy_data":
        from biopb.tensor.client import TensorFlightClient

        return TensorFlightClient.tensor_from_pb(
            image_data.lazy_data,
            cache_bytes=cache_bytes,
        )
    elif data_type is None:
        # Legacy fallback: check deprecated pixels field
        if image_data.HasField("pixels"):
            dim_order = image_data.pixels.dimension_order
            return _deserialize_to_numpy(
                image_data.pixels,
                np_index_order=dim_order[
                    ::-1
                ],  # dimension_order is F-order, so reverse for C-order index
            )
        else:
            raise ValueError("ImageData has no data field set")

    else:
        raise ValueError(f"Unknown ImageData data type: {data_type}")

get_image_data_dim_labels

get_image_data_dim_labels(
    image_data: ImageData, *, implicit: bool = False
) -> Optional[Sequence[str]]

Get a copy of the dimension labels from ImageData.

Parameters:

Name	Type	Description	Default
image_data	ImageData	ImageData protobuf message	required
implicit	bool	If True and dim_labels field is unset, return heuristic labels based on shape. Heuristics: - 2D -> ['y', 'x'] - 3D and dim[0] <= 4 -> ['c', 'y', 'x'] - 3D and dim[-1] <= 4 -> ['y', 'x', 'c'] - All other cases -> None	False

Returns:

Type	Description
Optional[Sequence[str]]	List of dimension labels, or None if unset and implicit=False or
Optional[Sequence[str]]	heuristics don't apply.

Source code in src/main/python/biopb/image/_utils.py

def get_image_data_dim_labels(
    image_data: ImageData, *, implicit: bool = False
) -> Optional[Sequence[str]]:
    """Get a copy of the dimension labels from ImageData.

    Args:
        image_data: ImageData protobuf message
        implicit: If True and dim_labels field is unset, return heuristic labels
            based on shape. Heuristics:
            - 2D -> ['y', 'x']
            - 3D and dim[0] <= 4 -> ['c', 'y', 'x']
            - 3D and dim[-1] <= 4 -> ['y', 'x', 'c']
            - All other cases -> None

    Returns:
        List of dimension labels, or None if unset and implicit=False or
        heuristics don't apply.
    """
    data_type = image_data.WhichOneof("data")
    if data_type == "eager_data":
        dim_labels = image_data.eager_data.dim_labels
    elif data_type == "lazy_data":
        from biopb.tensor.client import TensorFlightClient

        dim_labels = TensorFlightClient.descriptor_from_pb(
            image_data.lazy_data
        ).dim_labels
    else:
        dim_labels = None

    if dim_labels:
        return list(dim_labels)  # is a copy

    if not implicit:
        return None

    # Apply heuristic when dim_labels is unset and implicit=True
    shape = get_image_data_shape(image_data)
    if shape is None:
        return None

    ndim = len(shape)

    if ndim == 2:
        return ["y", "x"]
    elif ndim == 3:
        if shape[0] <= 4:
            return ["c", "y", "x"]
        elif shape[-1] <= 4:
            return ["y", "x", "c"]
        else:
            return None
    else:
        return None

mask_to_roi

mask_to_roi(mask: ndarray, *, bitorder: str = 'big') -> ROI

Convert a binary mask to a ROI protobuf.

Parameters:

Name	Type	Description	Default
mask	ndarray	Binary mask as numpy array (2D or 3D).	required
bitorder	str	Bit order for packing ('big' or 'little'). Defaults to 'big'.	'big'

Returns:

Type	Description
ROI	ROI protobuf message containing the mask data.

Source code in src/main/python/biopb/image/_utils.py

def mask_to_roi(mask: np.ndarray, *, bitorder: str = "big") -> ROI:
    """Convert a binary mask to a ROI protobuf.

    Args:
        mask: Binary mask as numpy array (2D or 3D).
        bitorder: Bit order for packing ('big' or 'little'). Defaults to 'big'.

    Returns:
        ROI protobuf message containing the mask data.
    """
    dim = mask.ndim

    if dim == 2:
        yp, xp = np.where(mask)
        ymin, xmin = yp.min(), xp.min()
        ymax, xmax = yp.max() + 1, xp.max() + 1
        rect = Rectangle(
            top_left=Point(y=ymin, x=xmin),
            bottom_right=Point(y=ymax, x=xmax),
        )
        pixels = mask[ymin:ymax, xmin:xmax]

    elif dim == 3:
        zp, yp, xp = np.where(mask)
        zmin, ymin, xmin = zp.min(), yp.min(), xp.min()
        zmax, ymax, xmax = zp.max() + 1, yp.max() + 1, xp.max() + 1
        rect = Rectangle(
            top_left=Point(z=zmin, y=ymin, x=xmin),
            bottom_right=Point(z=zmax, y=ymax, x=xmax),
        )
        pixels = mask[zmin:zmax, ymin:ymax, xmin:xmax]

    roi = ROI(
        mask=Mask(
            rectangle=rect,
            bin_data=BinData(
                data=np.packbits(pixels, bitorder=bitorder).tobytes(),
                endianness=0 if bitorder == "big" else 1,
            ),
        ),
    )

    return roi

normalize_array_dims

normalize_array_dims(
    arr: Union[ndarray, Array],
    dim_labels: Sequence[str],
    target_dim_labels: Sequence[str],
) -> Union[np.ndarray, da.Array]

Normalize array dimensions to match target dimension labels.

Transposes, squeezes, and adds singleton dimensions to convert an array with given dimension labels to match the target dimension order.

Parameters:

Name	Type	Description	Default
arr	Union[ndarray, Array]	Input array (numpy or dask)	required
dim_labels	Sequence[str]	Current dimension labels for each axis of arr. If None, raises ValueError.	required
target_dim_labels	Sequence[str]	Target dimension labels for the output array.	required

Returns:

Type	Description
Union[ndarray, Array]	Array with dimensions reordered to match target_dim_labels.
Union[ndarray, Array]	The output will have len(target_dim_labels) dimensions.

Raises:

Type	Description
ValueError	If dim_labels is None, if dim_labels length doesn't match array ndim, if dim_labels or target_dim_labels has duplicates, or if a dimension exists in dim_labels but not target_dim_labels with size > 1 (cannot squeeze non-singleton dimension).

Source code in src/main/python/biopb/image/_utils.py

def normalize_array_dims(
    arr: Union[np.ndarray, da.Array],
    dim_labels: Sequence[str],
    target_dim_labels: Sequence[str],
) -> Union[np.ndarray, da.Array]:
    """Normalize array dimensions to match target dimension labels.

    Transposes, squeezes, and adds singleton dimensions to convert an array
    with given dimension labels to match the target dimension order.

    Args:
        arr: Input array (numpy or dask)
        dim_labels: Current dimension labels for each axis of arr.
            If None, raises ValueError.
        target_dim_labels: Target dimension labels for the output array.

    Returns:
        Array with dimensions reordered to match target_dim_labels.
        The output will have len(target_dim_labels) dimensions.

    Raises:
        ValueError: If dim_labels is None,
            if dim_labels length doesn't match array ndim,
            if dim_labels or target_dim_labels has duplicates,
            or if a dimension exists in dim_labels but not target_dim_labels
            with size > 1 (cannot squeeze non-singleton dimension).
    """
    if not dim_labels:
        raise ValueError("no dim_labels, cannot normalize array dimensions")

    dim_labels = list(dim_labels)
    target_dim_labels = list(target_dim_labels)

    # Use uppercase for case-insensitive comparison
    dim_labels_upper = [label.upper() for label in dim_labels]
    target_dim_labels_upper = [label.upper() for label in target_dim_labels]

    if len(dim_labels) != arr.ndim:
        raise ValueError(
            f"dim_labels length ({len(dim_labels)}) does not match "
            f"array dimensions ({arr.ndim})"
        )

    if len(set(dim_labels_upper)) != len(dim_labels_upper):
        raise ValueError(f"dim_labels has duplicate labels: {dim_labels}")
    if len(set(target_dim_labels_upper)) != len(target_dim_labels_upper):
        raise ValueError(f"target_dim_labels has duplicate labels: {target_dim_labels}")

    shape_dict = {
        label.upper(): size for label, size in zip(dim_labels, arr.shape, strict=True)
    }

    squeeze_dims = set(dim_labels_upper) - set(target_dim_labels_upper)
    for label_upper in squeeze_dims:
        if shape_dict[label_upper] != 1:
            original_label = dim_labels[dim_labels_upper.index(label_upper)]
            raise ValueError(
                f"Dimension '{original_label}' has size {shape_dict[label_upper]} but is not in "
                f"target_dim_labels {target_dim_labels}. Cannot squeeze non-singleton dimension."
            )

    expand_dims = set(target_dim_labels_upper) - set(dim_labels_upper)

    # Step 1: Squeeze dimensions not in target
    if squeeze_dims:
        squeeze_axes = tuple(
            i for i, label in enumerate(dim_labels_upper) if label in squeeze_dims
        )
        arr = arr.squeeze(axis=squeeze_axes)
        dim_labels = [
            label for i, label in enumerate(dim_labels) if i not in squeeze_axes
        ]
        dim_labels_upper = [
            label for label in dim_labels_upper if label not in squeeze_dims
        ]

    # Step 2: Insert singleton dimensions for labels in target but not source
    for target_label in target_dim_labels:
        if target_label.upper() in expand_dims:
            insert_pos = target_dim_labels.index(target_label)
            if _is_dask_array(arr):
                import dask.array as da  # already imported; see _is_dask_array

                arr = da.expand_dims(arr, axis=insert_pos)
            else:
                arr = np.expand_dims(arr, axis=insert_pos)
            dim_labels.insert(insert_pos, target_label)
            dim_labels_upper.insert(insert_pos, target_label.upper())

    # Step 3: Transpose to match target order
    if dim_labels_upper != target_dim_labels_upper:
        transpose_axes = tuple(
            dim_labels_upper.index(label) for label in target_dim_labels_upper
        )
        arr = arr.transpose(transpose_axes)

    return arr

roi_to_mask

roi_to_mask(roi: ROI, mask: ndarray) -> np.ndarray

Convert a ROI protobuf to a binary mask.

Parameters:

Name	Type	Description	Default
roi	ROI	ROI protobuf message containing shape (point, rectangle, polygon, or mask).	required
mask	ndarray	Template numpy array defining output shape and dtype.	required

Returns:

Type	Description
ndarray	Binary mask as numpy array with same shape/dtype as input mask.

Raises:

Type	Description
ValueError	If mask dimension is not 2 or 3.
NotImplementedError	For unsupported ROI types or 3D polygons.

Source code in src/main/python/biopb/image/_utils.py

def roi_to_mask(roi: ROI, mask: np.ndarray) -> np.ndarray:
    """Convert a ROI protobuf to a binary mask.

    Args:
        roi: ROI protobuf message containing shape (point, rectangle, polygon, or mask).
        mask: Template numpy array defining output shape and dtype.

    Returns:
        Binary mask as numpy array with same shape/dtype as input mask.

    Raises:
        ValueError: If mask dimension is not 2 or 3.
        NotImplementedError: For unsupported ROI types or 3D polygons.
    """
    mask_ = np.zeros_like(mask, dtype="uint8")
    dim = mask_.ndim

    if dim not in (2, 3):
        raise ValueError(f"Illegal mask dimension {dim}.")

    def _get_int_point(p):
        return (int(p.z), int(p.y), int(p.x))

    roi_type = roi.WhichOneof("shape")
    if roi_type == "point":
        if dim == 3:
            mask_[_get_int_point(roi.point)] = 1
        else:
            mask_[_get_int_point(roi.point)[1:]] = 1

    elif roi_type == "rectangle":
        tl = _get_int_point(roi.rectangle.top_left)
        br = _get_int_point(roi.rectangle.bottom_right)
        if dim == 3:
            mask_[tl[0] : br[0], tl[1] : br[1], tl[2] : br[2]] = 1
        else:
            mask_[tl[1] : br[1], tl[2] : br[2]] = 1

    elif roi_type == "polygon":
        if dim != 2:
            raise NotImplementedError("3D polygon not supported")

        try:
            import cv2
        except ImportError as e:
            raise ImportError(
                "cv2 (opencv-python) is required for polygon ROI conversion. "
                "Install it with: pip install opencv-python"
            ) from e

        points = np.array([_get_int_point(p)[1:] for p in roi.polygon.points])
        points = points.reshape(-1, 1, 2)[:, :, ::-1]  # reverse x, y

        cv2.fillPoly(mask_, [points], color=1)

    elif roi_type == "mask":
        tl = _get_int_point(roi.mask.rectangle.top_left)
        br = _get_int_point(roi.mask.rectangle.bottom_right)

        bitorder = "big" if roi.mask.bin_data.endianness == 0 else "little"
        data = np.frombuffer(roi.mask.bin_data.data, dtype="uint8")
        data = np.unpackbits(data, bitorder=bitorder)

        if dim == 3:
            rect = mask_[tl[0] : br[0], tl[1] : br[1], tl[2] : br[2]]
        else:
            rect = mask_[tl[1] : br[1], tl[2] : br[2]]

        rect[:] = data[: rect.size].reshape(rect.shape)

    else:
        raise NotImplementedError(f"ROI type: {roi_type}")

    return mask_.astype(mask.dtype)

serialize_from_numpy_to_image_data

serialize_from_numpy_to_image_data(
    np_img: ndarray,
    *,
    dim_labels: Optional[Sequence[str]] = None
) -> ImageData

Convert numpy array to protobuf ImageData with eager_data.

Parameters:

Name	Type	Description	Default
np_img	ndarray	image as numpy array (any memory order is accepted)	required

Keyword Args: dim_labels: Dimension labels for the tensor. Must be same length as np_img.ndim. Returns: protobuf ImageData with eager_data set.

Source code in src/main/python/biopb/image/_utils.py

def serialize_from_numpy_to_image_data(
    np_img: np.ndarray,
    *,
    dim_labels: Optional[Sequence[str]] = None,
) -> ImageData:
    """Convert numpy array to protobuf ImageData with eager_data.

    Args:
        np_img: image as numpy array (any memory order is accepted)
    Keyword Args:
        dim_labels: Dimension labels for the tensor. Must be same length as np_img.ndim.
    Returns:
        protobuf ImageData with eager_data set.
    """
    tensor = _pb_from_np(
        np_img,
        dim_labels=dim_labels,
    )
    return ImageData(eager_data=tensor)