PointCloudBatch

SUMMARY

A batch of 3D point clouds.

from telekinesis import datatypes
import numpy as np
batch = datatypes.PointCloudBatch([np.zeros((10, 3), dtype=np.float32)])

API Reference

Complete API documentation for PointCloudBatch, including parameters, attributes, and methods.

View Reference →

Parameters

Parameter	Type	Default	Description
point_clouds	Sequence[PointCloud | np.ndarray]	Required	Sequence of PointCloud instances or position arrays with shape (N_i, 3). Bare arrays create positions-only point clouds.
compression	PointCloudCompression | int | None	None	Optional compression override applied to every point cloud. If None, existing PointCloud codecs are preserved and bare arrays use PointCloudCompression.NONE.

Raises

Exception	Condition
TypeError	point_clouds isn't a list/tuple, or an element isn't a PointCloud/np.ndarray
ValueError	An element's positions/normals/colors have an unsupported shape, or compression isn't a valid PointCloudCompression member/matching int

Attributes

Attribute	Type	Description
positions	list[np.ndarray]	Defensive copy of each cloud's positions, (N_i, 3) float32, one entry per cloud.
normals	list[np.ndarray | None]	Defensive copy of each cloud's normals, or None per cloud.
colors	list[np.ndarray | None]	Defensive copy of each cloud's colors, or None per cloud.
compressions	list[PointCloudCompression]	Each cloud's PointCloudCompression codec, in order, one entry per cloud.

The compressions attribute exposes each cloud's codec, but there's no batch-level accessor for the other Draco tuning parameters (quantization_bits, compression_level, ...). Index with batch[i] to get one PointCloud and read its own compression_settings.

Methods

Method	Type	Description
PointCloudBatch.coerce(value)	PointCloudBatch	Converts a sequence of point clouds into a PointCloudBatch. Accepts a list or tuple mixing PointCloud instances and bare (N, 3) arrays. If value is already a PointCloudBatch, it is returned unchanged.
to_numpy(copy=True)	list[np.ndarray]	Returns each cloud's positions as a list of plain arrays. With the default copy=True you get independent copies; pass copy=False to get direct references to the batch's internal position arrays instead. Normals, colors, and compression settings aren't included; read those from their own properties or by indexing into the batch.
copy()	PointCloudBatch	Returns a new, independent PointCloudBatch with its own storage for every cloud's positions, normals, and colors, keeping each cloud's Draco compression settings.

Operators

Operation	Behavior
len(batch)	Number of point clouds B.
batch[i]	int returns a materialized PointCloud (defensive copy; IndexError if out of range); slice or boolean np.ndarray mask returns a new PointCloudBatch sub-batch. Anything else raises TypeError.
batch == other	True only if other is a PointCloudBatch with the same number of clouds and row-for-row equal positions/normals/colors, using the same draco_atol tolerance rule as PointCloud.__eq__. Logs a warning if any row matched only within tolerance. NotImplemented if other isn't a PointCloudBatch.

Visualization

import rerun as rr

# Your code block
# ....

rr.init("point_cloud_batch_example", spawn=True)
datatypes.visualize(batch, entity_path="/batch", label="PointCloudBatch")

Example

"""Demonstrates the Telekinesis PointCloudBatch datatype."""

import time

import numpy as np
import rerun as rr
from loguru import logger

from telekinesis import datatypes

def point_cloud_batch_example():
    """Demonstrate creation, inspection, operations, visualization, and serialization."""

    # ======================= Create ============================================
    N = 2000
    cloud_1 = datatypes.PointCloud(
        np.random.randn(N, 3).astype(np.float32) + np.array([-5.0, 0.0, 0.0], dtype=np.float32),
        normals=np.random.randn(N, 3).astype(np.float32),
        colors=np.random.randint(0, 255, (N, 3), dtype=np.uint8),
    )
    cloud_2 = np.random.randn(N, 3).astype(np.float32) + np.array(
        [5.0, 0.0, 0.0], dtype=np.float32
    )
    point_cloud_batch = datatypes.PointCloudBatch([cloud_1, cloud_2])
    logger.info(f"Created PointCloudBatch: {point_cloud_batch}")

    point_cloud_batch_from_coerce = datatypes.PointCloudBatch.coerce([cloud_1, cloud_2])
    logger.info(f"PointCloudBatch created via coerce: {point_cloud_batch_from_coerce}")

    # ======================= Inspect ===========================================
    logger.info(f"positions={point_cloud_batch.positions}")
    logger.info(f"normals={point_cloud_batch.normals}")
    logger.info(f"colors={point_cloud_batch.colors}")
    logger.info(f"compressions={point_cloud_batch.compressions}")
    logger.info(f"length={len(point_cloud_batch)}")

    # ======================= Operations =========================================
    single_cloud = point_cloud_batch[0]
    logger.info(f"Single PointCloud at index 0: {single_cloud}")

    sliced_batch = point_cloud_batch[0:1]
    logger.info(f"Sliced PointCloudBatch: {sliced_batch}")

    mask = np.array([True, False])
    masked_batch = point_cloud_batch[mask]
    logger.info(f"Masked PointCloudBatch: {masked_batch}")

    point_cloud_batch_copy = point_cloud_batch.copy()
    logger.info(f"Copied PointCloudBatch: {point_cloud_batch_copy}")

    point_cloud_batch_numpy = point_cloud_batch.to_numpy(copy=True)
    logger.info(f"NumPy positions per cloud: {[array.shape for array in point_cloud_batch_numpy]}")

    updated_cloud = datatypes.PointCloud(
        np.random.randn(N, 3).astype(np.float32),
        colors=np.full((N, 3), [255, 0, 0], dtype=np.uint8),
    )
    rebuilt_batch = datatypes.PointCloudBatch([cloud_1, updated_cloud])
    logger.info(f"Rebuilt PointCloudBatch: {rebuilt_batch}")

    # ======================= Visualize =========================================
    rr.init("point_cloud_batch_example", spawn=True)
    datatypes.visualize(
        point_cloud_batch,
        entity_path="/point_cloud_batch/original",
        label=["Cloud 1", "Cloud 2"],
    )
    datatypes.visualize(
        rebuilt_batch,
        entity_path="/point_cloud_batch/rebuilt",
        label=["Cloud 1", "Updated Cloud 2"],
    )

    # ======================= Serialize / Deserialize ===========================
    start = time.perf_counter()
    serialized = datatypes.serialize(point_cloud_batch)
    serialization_ms = (time.perf_counter() - start) * 1000

    start = time.perf_counter()
    deserialized = datatypes.deserialize(serialized)["param_0"]
    deserialization_ms = (time.perf_counter() - start) * 1000

    logger.info(f"Deserialized PointCloudBatch: {deserialized}")
    logger.info(f"Round-trip successful: {point_cloud_batch == deserialized}")
    logger.info(f"Serialization time: {serialization_ms:.3f} ms")
    logger.info(f"Deserialization time: {deserialization_ms:.3f} ms")


if __name__ == "__main__":
    point_cloud_batch_example()