PointCloud

Represents a 3D point cloud.

Parameters

Field	Type	Description
positions	np.ndarray | list | tuple	Required point positions, shape (N, 3) (or (3,) for a single point, or empty for an empty cloud). Converted to float32.
normals	np.ndarray | list | tuple | None	Optional per-point normals, same shape rules as positions. Converted to float32. Default None.
colors	np.ndarray | list | tuple | None	Optional per-point RGB colors, same shape rules as positions. Converted to uint8. Default None.
use_compression	bool	If True, to_pyarrow emits a Draco-compressed payload (lossy). If False (default), it emits the lossless layout.
quantization_bits	int	Draco quantization bits per component, 1-30. Default 14.
compression_level	int	Draco compression effort, 0-10. Default 7.
quantization_range	float	Size of the bounding cube used for quantization; -1 (default) lets Draco infer it from the data.
quantization_origin	np.ndarray | list | tuple | None	Origin of the quantization bounding box, shape (3,). None (default) lets Draco infer it.
create_metadata	bool	Whether Draco embeds attribute metadata in the payload. Default False.

Raises

Exception	Condition
TypeError	positions/normals/colors isn't array-like convertible to the required dtype, or use_compression/create_metadata isn't a bool
ValueError	positions/normals/colors isn't shape (N, 3) after normalization; normals/colors has a different point count than positions; quantization_bits is outside [1, 30]; compression_level is outside [0, 10]; quantization_range isn't convertible to float; or quantization_origin isn't shape (3,)

Attributes

Attribute	Type	Description
positions	np.ndarray	Defensive copy, shape (N, 3) float32. Settable; the new value must have the same point count as the current cloud (build a new PointCloud to change N).
has_normals	bool	Whether normals is not None.
normals	np.ndarray | None	Defensive copy, shape (N, 3) float32, or None. Settable (pass None to clear); the new value must match the current point count.
has_colors	bool	Whether colors is not None.
colors	np.ndarray | None	Defensive copy, shape (N, 3) uint8, or None. Settable (pass None to clear); the new value must match the current point count.
use_compression	bool	Whether to_pyarrow emits a Draco payload. Settable independently of the other Draco parameters.
compression_settings	dict[str, Any]	use_compression, quantization_bits, compression_level, quantization_range, quantization_origin (copy, or None), and create_metadata.
draco_atol	ClassVar[float]	1e-2. Absolute tolerance __eq__ uses to compare positions/normals when either operand has use_compression=True. Shared across all instances; can be reassigned at the class level.
default_quantization_bits, default_compression_level, default_quantization_range	ClassVar[int | float]	14, 7, -1.0 -- the shared defaults used by the constructor, set_compression_parameters, and encode_bytes.

Methods

Method	Description
set_compression_parameters(quantization_bits=14, compression_level=7, quantization_range=-1.0, quantization_origin=None, create_metadata=False)	Reconfigures the Draco tuning parameters. Takes effect only once use_compression is True; logs a warning if compression is currently disabled.
to_numpy(copy=True)	Returns positions as np.ndarray. Mirrors the positions property but with a copy=False zero-copy option. Does not include normals/colors.
save_to_path(path)	Writes a binary PLY file with an XYZ vertex list, plus normals/colors fields when present.
PointCloud.from_path(path)	Loads a PointCloud from a .ply file. Alpha channels in vertex colors are dropped; NaN/Inf in the file are not checked.
PointCloud.from_url(url, *, cache_dir=None, use_cache=True, connect_timeout=5.0, read_timeout=30.0)	Downloads (or reuses a cached copy of) a .ply file and loads it via from_path.
PointCloud.encode_bytes(positions, normals, colors, use_compression, ...)	Static helper producing the on-wire payloads used by to_pyarrow: raw arrays when use_compression=False, Draco-encoded bytes when True (colors are embedded in the positions payload under Draco).
PointCloud.decode_bytes(positions_payload, normals_payload, colors_payload, point_count, has_normals, has_colors, use_compression)	Reverse of encode_bytes, used by from_pyarrow.
PointCloud.coerce(value)	Returns value unchanged if already a PointCloud; otherwise wraps an array-like of (N, 3) positions. Raises TypeError for anything else.

Operators

Operation	Behavior
pc == other	True only if other is a PointCloud with equal colors (exact) and equal positions/normals -- bit-exact, unless either side has use_compression=True, in which case those two fields are compared within draco_atol since Draco is lossy. Logs a warning when equality only held within tolerance. NotImplemented if other isn't a PointCloud.
len(pc)	Number of points N.
np.asarray(pc)	Returns a copy of positions only (not normals/colors). copy=False raises ValueError; use to_numpy(copy=False) for a zero-copy view.
hash(pc)	Not supported.

Serialization

Draco compression (use_compression=True) is lossy: to_pyarrow quantizes positions/normals to quantization_bits levels across the bounding box, so a round-trip does not reproduce the exact input floats (roughly 1e-4-1e-3 absolute error at the default quantization_bits=14, growing as the bit count drops). colors, when present, are embedded inside the Draco positions payload at full uint8 precision and are unaffected.

Arrow layout:

StructArray length 1
├── use_compression:   bool
├── positions_payload: binary
├── normals_payload:   binary nullable  (null when normals absent)
├── colors_payload:    binary nullable  (null when colors absent, or always null under Draco)
├── point_count:       int64
├── has_normals:       bool
└── has_colors:        bool

Visualization

datatypes.visualize(point_cloud, entity_path=...) logs positions/colors as rr.Points3D. A string label renders as a floating text annotation at the cloud's centroid (mean position).

Example

"""Demonstrates the Telekinesis PointCloud datatype."""

from pathlib import Path
import time

import numpy as np
from loguru import logger
import rerun as rr
import rerun.blueprint as rrb

from telekinesis import datatypes

ROOT_PATH = Path(__file__).parent.parent

def point_cloud_example():
    """Demonstrate creation of compressed and uncompressed point clouds, access, visualization, update, compression settings, loading/saving .ply files, NumPy interop, and serialization."""

    # ======================= Create ============================================
    N = 4000000
    positions = np.random.randn(N, 3).astype(np.float32)
    normals = np.random.randn(N, 3).astype(np.float32)
    colors = np.random.randint(0, 255, (N, 3), dtype=np.uint8)

    uncompressed = datatypes.PointCloud(
        positions=positions, normals=normals, colors=colors, use_compression=False
    )
    compressed = datatypes.PointCloud(
        positions=positions, normals=normals, colors=colors, use_compression=True
    )
    logger.info(f"Original Uncompressed PointCloud: {uncompressed}")
    logger.info(f"Original Compressed PointCloud: {compressed}")

    # ======================= Inspect ===========================================
    positions = uncompressed.positions
    normals = uncompressed.normals
    colors = uncompressed.colors
    compression_settings = uncompressed.compression_settings

    logger.info(f"Underlying positions: {positions}")
    logger.info(f"Underlying normals: {normals}")
    logger.info(f"Underlying colors: {colors}")
    logger.info(f"Compression settings: {compression_settings}")

    # ======================= Visualize =========================================
    blueprint = rrb.Blueprint(
        rrb.Grid(
            rrb.Spatial3DView(name="My PointCloud", origin="/PointCloud/my_pointcloud"),
            rrb.Spatial3DView(name="Updated PointCloud", origin="/PointCloud/updated"),
            rrb.Spatial3DView(
                name="Updated PointCloud Colors", origin="/PointCloud/updated_colors"
            ),
        )
    )
    rr.init("point_cloud_example", spawn=True, default_blueprint=blueprint)
    rr.send_blueprint(blueprint, make_active=True)
    datatypes.visualize(
        uncompressed, entity_path="/PointCloud/my_pointcloud", label="My PointCloud"
    )

    # ======================= Update ============================================
    updated_positions = np.random.randn(N, 3).astype(np.float32)
    uncompressed.positions = updated_positions
    logger.info(f"Updated positions: {uncompressed}")

    updated_colors = np.random.randint(0, 255, (N, 3), dtype=np.uint8)
    uncompressed.colors = updated_colors
    logger.info(f"Updated colors: {uncompressed.colors}")
    datatypes.visualize(
        uncompressed,
        entity_path="/PointCloud/updated_colors",
        label="Updated PointCloud Colors",
    )

    # ======================= Compression =======================================
    uncompressed.use_compression = True
    uncompressed.set_compression_parameters(compression_level=5, quantization_bits=12)
    logger.info(f"Updated compression settings: {uncompressed}")

    # ======================= Load / Save =======================================
    url = "https://assets.telekinesis.ai/examples/v1/point_clouds/zivid_bottles_10_preprocessed.ply"
    from_url = datatypes.PointCloud.from_url(url=url)
    logger.info(f"PointCloud from .ply: {from_url}")
    datatypes.visualize(from_url, entity_path="/PointCloud/from_url", label="URL PointCloud")

    from_url.save_to_path("results/my_point_cloud_saved.ply")
    logger.info("Saved PointCloud to disk as .ply file.")

    # ======================= NumPy Interop =====================================
    numpy_data = uncompressed.to_numpy()
    array_data = np.asarray(uncompressed)
    centroid = np.mean(uncompressed, axis=0)
    logger.info(f"NumPy array: {numpy_data}")
    logger.info(f"As array: {array_data}")
    logger.info(f"Centroid: {centroid}")

    # ======================= Serialize / Deserialize ===========================
    start = time.perf_counter()
    serialized_uncompressed = datatypes.serialize(uncompressed)
    uncompressed_serialization_ms = (time.perf_counter() - start) * 1000
    uncompressed_serialized_size = len(serialized_uncompressed)

    start = time.perf_counter()
    deserialized = datatypes.deserialize(serialized_uncompressed)["param_0"]
    uncompressed_deserialization_ms = (time.perf_counter() - start) * 1000
    uncompressed_deserialized_size = len(serialized_uncompressed)

    logger.info(f"Deserialized Uncompressed PointCloud: {deserialized}")
    logger.info(f"Round-trip successful: {deserialized == uncompressed}")

    start = time.perf_counter()
    serialized_compressed = datatypes.serialize(compressed)
    compressed_serialization_ms = (time.perf_counter() - start) * 1000
    compressed_serialized_size = len(serialized_compressed)

    start = time.perf_counter()
    deserialized = datatypes.deserialize(serialized_compressed)["param_0"]
    compressed_deserialization_ms = (time.perf_counter() - start) * 1000
    compressed_deserialized_size = len(serialized_compressed)

    logger.info(f"Deserialized Compressed PointCloud: {deserialized}")
    logger.info(f"Round-trip successful: {deserialized == compressed}")

    logger.info(
        f"Uncompressed: serialize={uncompressed_serialization_ms:.3f} ms "
        f"({uncompressed_serialized_size / 1024 / 1024:.3f} MB), "
        f"deserialize={uncompressed_deserialization_ms:.3f} ms "
        f"({uncompressed_deserialized_size / 1024 / 1024:.3f} MB)"
    )
    logger.info(
        f"Compressed: serialize={compressed_serialization_ms:.3f} ms "
        f"({compressed_serialized_size / 1024 / 1024:.3f} MB), "
        f"deserialize={compressed_deserialization_ms:.3f} ms "
        f"({compressed_deserialized_size / 1024 / 1024:.3f} MB)"
    )


if __name__ == "__main__":
    point_cloud_example()