Core Connected Components

def connected_components(
    data: NDArray[Any],
    max_labels: int = -1,
    connectivity: Literal[4, 6, 8, 18, 26] = 26,
    return_N: bool = False,
    delta: float = 0,
    out_dtype: DTypeLike = None,
    out_file: Union[str, BinaryIO, None] = None,
    periodic_boundary: bool = False,
    binary_image: bool = False,
) -> Union[NDArray[Union[np.uint16, np.uint32, np.uint64]], tuple[NDArray[Union[np.uint16, np.uint32, np.uint64]], int]]:
    """
    Connected components applied to 3D images with handling for multiple labels.

    Parameters:
    - data: Input weights in a 2D or 3D numpy array
    - max_labels: Save memory by predicting the maximum number of possible labels
    - connectivity: For 3D images, 6 (faces), 18 (+edges), or 26 (+corners). For 2D images, 4 (faces) or 8 (+corners)
    - return_N: If True, also return the number of connected components
    - delta: Connect together values whose difference is <= delta (for continuous images)
    - out_dtype: Output data type (np.uint16, np.uint32, or np.uint64)
    - out_file: If specified, output array will be memory-mapped to this file
    - periodic_boundary: The boundary edges wrap around
    - binary_image: Treat input as binary (foreground > 0, background == 0)

    Returns:
    - NDArray: Labeled components numbered from 1 to N
    - tuple[NDArray, int]: If return_N=True, returns (labeled_array, num_components)
    """

import cc3d
import numpy as np

# Basic usage with default 26-connectivity
labels_in = np.random.randint(0, 5, (100, 100, 100), dtype=np.int32)
labels_out = cc3d.connected_components(labels_in)

# Specify connectivity (6, 18, or 26 for 3D; 4 or 8 for 2D)
labels_out = cc3d.connected_components(labels_in, connectivity=6)

# Get number of components
labels_out, N = cc3d.connected_components(labels_in, return_N=True)
print(f"Found {N} connected components")

# Binary image processing
binary_image = (labels_in > 0).astype(np.uint8)
labels_out = cc3d.connected_components(binary_image, binary_image=True)

# Continuous value CCL with delta threshold
grayscale = np.random.random((50, 50, 50)) * 255
labels_out = cc3d.connected_components(grayscale, delta=10)

# Periodic boundary conditions (for simulations)
labels_out = cc3d.connected_components(
    labels_in, connectivity=6, periodic_boundary=True
)

# Memory-mapped output for large datasets
labels_out = cc3d.connected_components(
    labels_in, out_file="large_output.bin"
)

# Control output data type
labels_out = cc3d.connected_components(
    labels_in, out_dtype=np.uint64
)

def connected_components_stack(
    stacked_images: typing.Iterable[NDArray[typing.Any]], 
    connectivity: Literal[6,26] = 26,
    return_N: bool = False,
    out_dtype: DTypeLike = None,
    binary_image: bool = False,
) -> Union[CrackleArray, tuple[CrackleArray, int]]:
    """
    Connected component labeling on arrays larger than RAM.

    Parameters:
    - stacked_images: Sequence of 3D images in sequential z-order
    - connectivity: 6 (faces) or 26 (faces+edges+corners)  
    - return_N: If True, return (CrackleArray, num_components)
    - out_dtype: Output data type
    - binary_image: Treat input images as binary

    Returns:
    - CrackleArray: Compressed array with random access by z-slice
    - tuple[CrackleArray, int]: If return_N=True, includes component count
    """

import cc3d
import numpy as np

def create_image_sections(full_image):
    """Generator that yields thick Z slices sequentially."""
    for z in range(0, full_image.shape[2], 100):
        yield full_image[:, :, z:z+100]

# Process very large dataset
# Note: requires 'pip install connected-components-3d[stack]'
large_image = np.random.randint(0, 1000, (2000, 2000, 2000))
compressed_labels = cc3d.connected_components_stack(
    create_image_sections(large_image),
    connectivity=26
)

# Access parts of the result
slice_data = compressed_labels[:, :, 500:600]

# Save compressed result
compressed_labels.save("huge_ccl_result.ckl")

# Convert to numpy (only if it fits in RAM)
full_result = compressed_labels.numpy()

def estimate_provisional_labels(
    data: NDArray[Any],
) -> tuple[int, int, int]:
    """Estimate the number of provisional labels required for connected components."""

# Estimate memory requirements before processing
estimates = cc3d.estimate_provisional_labels(large_array)
x_transitions, total_estimate, static_estimate = estimates
print(f"Estimated provisional labels: {total_estimate}")