PointCloudBatch

Represents an ordered batch of point clouds.

Parameters

Field	Type	Description
point_clouds	Sequence[PointCloud | np.ndarray]	A sequence of PointCloud instances and/or bare (N, 3) np.ndarrays. A bare ndarray is treated as positions-only (no normals/colors), using PointCloud's default Draco settings.

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

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.

There's no batch-level accessor for Draco settings (use_compression, quantization_bits, ...) and no to_numpy()/data shortcut for the whole batch -- index with batch[i] to get one PointCloud and read its own compression_settings.

Methods

Method	Description
PointCloudBatch.coerce(value)	Returns value unchanged if already a PointCloudBatch; otherwise wraps a list/tuple of PointCloud/np.ndarray. Raises TypeError for anything else.

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__, per row. Logs a warning if any row matched only within tolerance. NotImplemented if other isn't a PointCloudBatch.
hash(batch)	Not supported.

Serialization

Each cloud is encoded independently (lossless or Draco) per its own use_compression setting -- see PointCloud's Serialization section for the Draco lossiness caveats.

Arrow layout:

StructArray length 1
├── positions:         list<fixed_size_list<float32>[3]> nullable  (per-row null when Draco-encoded)
├── normals:           list<fixed_size_list<float32>[3]> nullable
├── colors:            list<fixed_size_list<uint8>[3]> nullable
├── use_compression:   list<bool>
├── positions_payload: list<binary> nullable  (Draco-encoded positions[+colors])
├── normals_payload:   list<binary> nullable  (Draco-encoded normals)
├── point_count:       list<int64>
├── has_normals:       list<bool>
└── has_colors:        list<bool>

One PointCloudBatch object is serialized as one Arrow row; each field's list holds the batch's B clouds, each cloud encoded per its own use_compression setting.

Visualization

datatypes.visualize(batch, entity_path=...) logs each cloud under its own indexed child path ({entity_path}/{i}) as rr.Points3D. A list[str] label (one per cloud) renders one floating text annotation per cloud, each at that cloud's own centroid -- all clouds share one 3D view, so there's no per-cloud "tile" the way there is for ImageBatch.

Example

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

import time

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

from telekinesis import datatypes

def point_cloud_batch_example():
    """Demonstrate creation, visualization, access, indexing, rebuilding, and serialization of a PointCloudBatch."""

    # ======================= Create ============================================
    N = 4000000
    cloud_1 = datatypes.PointCloud(
        np.random.randn(N, 3).astype(np.float32) + np.array([-10.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),
        use_compression=False,
    )
    cloud_2 = np.random.randn(N, 3).astype(np.float32) + np.array(
        [10.0, 0.0, 0.0], dtype=np.float32
    )
    cloud_3 = np.random.randn(N, 3).astype(np.float32) + np.array(
        [10.0, 0.0, 0.0], dtype=np.float32
    )
    clouds = [cloud_1, cloud_2, cloud_3]
    batch = datatypes.PointCloudBatch(clouds)

    logger.info(f"Original PointCloudBatch: {batch}")

    # ======================= Visualize =========================================
    rr.init("point_cloud_batch_example", spawn=True)
    datatypes.visualize(
        batch,
        entity_path="/PointCloudBatch/my_point_cloud_batch",
        label=["My PointCloud 1", "My PointCloud 2", "My PointCloud 3"],
    )

    # ======================= Inspect ===========================================
    positions = batch.positions
    normals = batch.normals
    colors = batch.colors
    length = len(batch)

    logger.info(f"length={length}")
    logger.info(f"Underlying positions: {positions}")
    logger.info(f"Underlying normals: {normals}")
    logger.info(f"Underlying colors: {colors}")

    # ======================= Index =============================================
    index = 0
    single_cloud = batch[index]
    logger.info(f"Single PointCloud from batch: {single_cloud}")
    datatypes.visualize(
        single_cloud,
        entity_path="/PointCloudBatch/my_updated_point_cloud_1",
        label="My Updated PointCloud 1",
    )

    # ======================= Rebuild ===========================================
    index = -1
    updated_cloud = datatypes.PointCloud(
        np.random.randn(N, 3).astype(np.float32),
        normals=np.random.randn(N, 3).astype(np.float32),
        colors=np.full((N, 3), [255, 0, 0], dtype=np.uint8),
        use_compression=True,
    )
    datatypes.visualize(
        updated_cloud,
        entity_path="/PointCloudBatch/my_updated_point_cloud_2",
        label="My Updated PointCloud 2",
    )
    start = time.perf_counter()
    clouds[index] = updated_cloud
    batch = datatypes.PointCloudBatch(clouds)
    rebuild_ms = (time.perf_counter() - start) * 1000
    logger.info(f"Rebuilt PointCloudBatch: {batch}")
    logger.info(f"Rebuild time: {rebuild_ms:.3f} ms")

    # ======================= Serialize / Deserialize ===========================
    start = time.perf_counter()
    serialized = datatypes.serialize(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: {deserialized == batch}")
    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()