PointCloud

SUMMARY

A 3D point cloud with optional per-point normals and colors.

from telekinesis import datatypes
import numpy as np
point_cloud = datatypes.PointCloud(np.zeros((10, 3), dtype=np.float32))

API Reference

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

View Reference →

Parameters

Parameter	Type	Default	Description
positions	np.ndarray | list | tuple	Required	Point positions with shape (N, 3). A (3,) value is accepted as one point, and empty input creates an empty point cloud.
normals	np.ndarray | list | tuple | None	None	Optional per-point normals with the same number of points as positions.
colors	np.ndarray | list | tuple | None	None	Optional per-point RGB colors with the same number of points as positions.
compression	PointCloudCompression | int	PointCloudCompression.NONE	Compression codec used during serialization; PointCloudCompression.DRACO uses lossy compression.
quantization_bits	int	14	Number of Draco quantization bits per component, in the range [1, 30].
compression_level	int	7	Draco compression effort, in the range [0, 10]. Higher values generally produce smaller output but require more encoding time.
quantization_range	float	-1.0	Size of the Draco quantization bounding cube. Set to -1 to let Draco determine the range automatically.
quantization_origin	np.ndarray | list | tuple | None	None	Optional origin of the Draco quantization bounding cube, with shape (3,). If None, Draco determines the origin automatically.
create_metadata	bool	False	Whether to embed attribute metadata in the Draco payload.

Raises

Exception	Condition
TypeError	positions/normals/colors isn't array-like convertible to the required dtype, or 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; compression isn't a valid PointCloudCompression member/matching int; 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.
compression	PointCloudCompression	Codec used for serialization (NONE or DRACO). Fixed at construction; build a new PointCloud to change it.
compression_settings	dict[str, Any]	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 compression=PointCloudCompression.DRACO. 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. Shared defaults used by the constructor and set_compression_parameters.

Methods

Method	Type	Description
PointCloud.coerce(value)	PointCloud	Converts array-like point positions into a PointCloud, running the same validation as the constructor. If value is already a PointCloud, it is returned unchanged.
PointCloud.from_path(path)	PointCloud	Loads a point cloud from a .ply file. Alpha channels in vertex colors are dropped, and NaN/Inf values in the file are not checked.
PointCloud.from_url(url, *, cache_dir=None, use_cache=True, connect_timeout=5.0, read_timeout=30.0)	PointCloud	Downloads (or reuses a cached copy of) a .ply file and loads it the same way as from_path.
to_numpy(copy=True)	np.ndarray	Returns the point positions as a plain array (not normals/colors). Pass copy=False to get a direct reference instead, so mutating it also mutates the PointCloud.
copy()	PointCloud	Returns a new, independent PointCloud with the same positions, normals, colors, and compression settings.
save_to_path(path)	None	Writes a binary PLY file with an XYZ vertex list, plus normals/colors fields when present.
set_compression_parameters(quantization_bits=14, compression_level=7, quantization_range=-1.0, quantization_origin=None, create_metadata=False)	None	Reconfigures the Draco compression tuning parameters. They only take effect once compression is PointCloudCompression.DRACO. If compression is currently NONE, you'll get a warning instead of an error.

Operators

Operation	Behavior
pc == other	True only if other is a PointCloud with equal colors (exact) and equal positions/normals. Comparison is bit-exact unless either side has compression=PointCloudCompression.DRACO, in which case positions/normals 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) as an np.ndarray; NumPy functions accept a PointCloud directly. Passing copy=False raises ValueError.

Visualization

import rerun as rr

# Your code block
# ....

rr.init("point_cloud_example", spawn=True)
datatypes.visualize(point_cloud, entity_path="/point_cloud", label="PointCloud")

Example

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

import time
from pathlib import Path

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

from telekinesis import datatypes

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

    # ======================= Create ============================================
    N = 2000
    positions = np.random.randn(N, 3).astype(np.float32)
    point_cloud = datatypes.PointCloud(positions)
    logger.info(f"Created PointCloud: {point_cloud}")

    normals = np.random.randn(N, 3).astype(np.float32)
    colors = np.random.randint(0, 255, (N, 3), dtype=np.uint8)
    point_cloud = datatypes.PointCloud(
        positions,
        normals=normals,
        colors=colors,
        compression=datatypes.PointCloudCompression.DRACO,
    )
    logger.info(f"PointCloud with normals, colors, and compression: {point_cloud}")

    point_cloud_from_coerce = datatypes.PointCloud.coerce(positions)
    logger.info(f"PointCloud created via coerce: {point_cloud_from_coerce}")

    url = "https://assets.telekinesis.ai/examples/v1/point_clouds/zivid_bottles_10_preprocessed.ply"
    point_cloud_from_url = datatypes.PointCloud.from_url(url=url)
    logger.info(f"PointCloud loaded from URL: {point_cloud_from_url}")

    cached_path = Path.home() / ".cache" / "telekinesis" / "point_clouds" / Path(url).name
    point_cloud_from_path = datatypes.PointCloud.from_path(cached_path)
    logger.info(f"PointCloud loaded from path: {point_cloud_from_path}")

    # ======================= Inspect ===========================================
    logger.info(f"positions={point_cloud.positions}")
    logger.info(f"normals={point_cloud.normals}")
    logger.info(f"colors={point_cloud.colors}")
    logger.info(f"has_normals={point_cloud.has_normals}")
    logger.info(f"has_colors={point_cloud.has_colors}")
    logger.info(f"compression={point_cloud.compression}")
    logger.info(f"compression_settings={point_cloud.compression_settings}")
    logger.info(f"draco_atol={point_cloud.draco_atol}")

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

    point_cloud.normals = np.random.randn(N, 3).astype(np.float32)
    logger.info(f"Updated normals: {point_cloud}")

    point_cloud.colors = np.random.randint(0, 255, (N, 3), dtype=np.uint8)
    logger.info(f"Updated colors: {point_cloud}")

    # `compression` is fixed at construction; build a new PointCloud to change it.
    point_cloud_no_compression = datatypes.PointCloud(
        point_cloud.positions,
        normals=point_cloud.normals,
        colors=point_cloud.colors,
        compression=datatypes.PointCloudCompression.NONE,
    )
    logger.info(f"Rebuilt with compression=NONE: {point_cloud_no_compression.compression}")

    point_cloud.set_compression_parameters(compression_level=5, quantization_bits=12)
    logger.info(f"Updated compression settings: {point_cloud.compression_settings}")

    point_cloud_copy = point_cloud.copy()
    logger.info(f"Copied PointCloud: {point_cloud_copy}")

    point_cloud_numpy = point_cloud.to_numpy(copy=True)
    logger.info(f"NumPy positions:\n{point_cloud_numpy}")

    array_data = np.asarray(point_cloud)
    centroid = np.mean(point_cloud, axis=0)
    logger.info(f"As array: {array_data}")
    logger.info(f"Centroid: {centroid}")

    logger.info(f"length={len(point_cloud)}")

    save_path = "results/point_cloud_example.ply"
    point_cloud.save_to_path(save_path)
    logger.info(f"Saved PointCloud to {save_path}")

    # ======================= Visualize =========================================
    rr.init("point_cloud_example", spawn=True)
    datatypes.visualize(
        point_cloud, entity_path="/point_cloud/updated", label="Updated PointCloud"
    )
    datatypes.visualize(
        point_cloud_from_url, entity_path="/point_cloud/from_url", label="URL PointCloud"
    )

    # ======================= Serialize / Deserialize ===========================
    start = time.perf_counter()
    serialized = datatypes.serialize(point_cloud)
    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 PointCloud: {deserialized}")
    logger.info(f"Round-trip successful: {point_cloud == deserialized}")
    logger.info(f"Serialization time: {serialization_ms:.3f} ms")
    logger.info(f"Deserialization time: {deserialization_ms:.3f} ms")


if __name__ == "__main__":
    point_cloud_example()