voyant-api 0.9.2

Python bindings for Voyant Photonics, Inc. sensors

Voyant API - Python Bindings

Python bindings for Voyant Photonics LiDAR sensors, providing high-performance access to point cloud data.

Installation

pip install voyant-api

Sensor compatibility

Sensor	Client
Carbon	CarbonClient + CarbonConfig
Meadowlark	VoyantClient (deprecated — see below)

Quick Start

Receiving live data (Carbon)

import time
from voyant_api import CarbonClient, CarbonConfig, init_voyant_logging

init_voyant_logging()

config = CarbonConfig()
config.set_bind_addr("0.0.0.0:5678")
config.set_group_addr("224.0.0.0")
config.set_interface_addr("192.168.1.100")

client = CarbonClient(config)
client.start()

# Press Ctrl+C to stop
while client.is_running():
    frame = client.try_receive_frame()
    if frame is not None:
        print(frame)

        # Get point cloud as numpy array (N x 4: x, y, z, radial_vel)
        xyzv = frame.xyzv()
        print(f"Points shape: {xyzv.shape}")
    else:
        time.sleep(0.001)

Config can also be loaded from a JSON file — see CarbonConfig JSON format below.

Recording data

import time
from voyant_api import CarbonClient, CarbonConfig, VoyantRecorder, RecordStatus, init_voyant_logging

init_voyant_logging()

config = CarbonConfig()
config.set_bind_addr("0.0.0.0:5678")
config.set_group_addr("224.0.0.0")
config.set_interface_addr("192.168.1.100")

client = CarbonClient(config)
client.start()

with VoyantRecorder(
    output_path="my_recording.bin",
    timestamp_filename=True,
    max_total_frames=1000,  # Optional: stop after 1000 frames
) as recorder:
    while client.is_running():
        frame = client.try_receive_frame()
        if frame is not None:
            status = recorder.record_frame(frame)
            if status == RecordStatus.STOP:
                break
        else:
            time.sleep(0.001)

Playing back recordings

from voyant_api import VoyantPlayback, init_voyant_logging
from voyant_api.pandas_utils import frame_to_dataframe

init_voyant_logging()

with VoyantPlayback(filter_points=True) as playback:
    playback.open("my_recording.bin")

    for frame in playback:
        if frame is None:
            break

        print(frame)

        # Convert to pandas DataFrame
        df = frame_to_dataframe(frame)
        print(df.head())

Converting recordings to PCD

from voyant_api import VoyantPlayback, init_voyant_logging
from voyant_api.pcd_utils import save_frame_to_pcd, frame_to_extended_pcd

init_voyant_logging()

with VoyantPlayback(filter_points=True) as playback:
    playback.open("my_recording.bin")

    for frame in playback:
        if frame is None:
            break

        # Save directly to .pcd file
        save_frame_to_pcd(frame, f"frame_{frame.frame_index}.pcd")

        # Or get a PointCloud object for further processing
        pc = frame_to_extended_pcd(frame)
        pc.save(f"frame_{frame.frame_index}.pcd")

Sending SDL commands

SDL (Software Defined Lidar) commands configure the sensor at runtime — changing operating state, field of view, frame rate, and waveform parameters.

import time
from voyant_api import (
    CarbonClient, CarbonConfig,
    SdlCommand, SdlState, SdlRampLength, SdlStatus,
    init_voyant_logging,
)

init_voyant_logging()

config = CarbonConfig()
config.set_bind_addr("0.0.0.0:5678")
config.set_group_addr("224.0.0.0")
config.set_interface_addr("192.168.1.100")

client = CarbonClient(config)
client.start()

# Build a command — construct with defaults, override only what you need.
cmd = SdlCommand()
cmd.req_state = SdlState.PointCloud
cmd.hfov_deg = 60.0
cmd.hfov_center_deg = 0.0
cmd.frame_rate_fps = 10.0
cmd.ramp_length = SdlRampLength.V16_384us

# Send — returns immediately. Pending means the UDP packet was accepted.
status = client.send_sdl(cmd)
if status != SdlStatus.Pending:
    print(f"Command rejected before sending: {status}")

# Poll for confirmation inside your frame loop.
# The sensor confirms via heartbeat, so poll every iteration.
while client.is_running():
    frame = client.try_receive_frame()
    if frame is None:
        time.sleep(0.001)

    status = client.poll_sdl()
    if status == SdlStatus.Applied:
        print("Command applied.")
        break
    elif status not in (SdlStatus.Idle, SdlStatus.Pending):
        print(f"Command failed: {status}")
        break

API Overview

CarbonClient

Receives live data from Carbon sensors. Construct with a CarbonConfig and call start() before polling for frames.

config = CarbonConfig()
config.set_bind_addr("0.0.0.0:5678")
config.set_group_addr("224.0.0.0")
config.set_interface_addr("192.168.1.100")
config.set_range_max(50.0)
config.set_pfa(1e-4)

client = CarbonClient(config)
client.start()

CarbonConfig

Configuration for the Carbon pipeline. Construct with defaults and setters, or load from JSON.

# From defaults
config = CarbonConfig()
config.set_bind_addr("0.0.0.0:5678")

# From a JSON file
config = CarbonConfig.from_json("config.json")

CarbonConfig JSON format

All fields are optional and fall back to their defaults when omitted. To see all available fields and their current defaults, run:

from voyant_api import CarbonConfig
print(CarbonConfig())

This prints the full nested config with all current defaults, for example:

CarbonConfig { receiver: ReceiverConfig { multicast: MulticastReceiverConfig { bind_addr: "0.0.0.0:5678", group_addr: "224.0.0.0", interface_addr: "127.0.0.1" }, batch_size: 32, ... }, dsp: DspConfig { pfa: None, bandwidth_hz: None, elevation_fov_deg: 30.0, ... }, ... }

Any field shown in that output can be set in the JSON file. Fields showing None are unset and use sensor defaults.

SdlCommand

Configures sensor parameters at runtime. Construct with defaults and override only the fields you need. All setters validate the value immediately and raise ValueError if it is out of range.

from voyant_api import SdlCommand, SdlState, SdlRampLength

cmd = SdlCommand()
cmd.req_state = SdlState.PointCloud        # Operating state
cmd.hfov_deg = 60.0                        # Horizontal FOV (0.0 – 120.0°)
cmd.hfov_center_deg = 0.0                  # FOV center (−60.0 – 60.0°)
cmd.frame_rate_fps = 10.0                  # Frame rate (1.0 – 19.0 fps)
cmd.ramp_bandwidth_ghz = 6.0               # Ramp bandwidth (0.5 – 10.0 GHz)
cmd.ramp_length = SdlRampLength.V16_384us  # Ramp length

Send with client.send_sdl(cmd) and poll for confirmation with client.poll_sdl().

send_sdl returns a status immediately:

Status	Meaning
Pending	Command sent, awaiting heartbeat confirmation
InvalidParameter	A value is out of range — not sent
BadFovCenterCombo	FOV/center combination is invalid — not sent
FovFpsError	FOV/FPS combination exceeds hardware limits — not sent
SendFailed	UDP send failed — check logs
PreviousCommandPending	Another command is already in flight — wait for it to resolve

poll_sdl returns Idle when nothing is in flight, Pending while waiting for confirmation, and a resolved status once the sensor confirms or the command times out:

Status	Meaning
Idle	No command is currently in flight
Pending	Waiting for heartbeat confirmation
Applied	Sensor confirmed the command was applied
Timeout	No heartbeat confirmation within the timeout window
MaxRetriesExceeded	Retransmitted the maximum number of times without confirmation
StreamReset	Heartbeat frame counter jumped backwards — stream was reset
Any rejection status	Sensor rejected the command

Note: Only one SDL command can be in flight at a time. Idle means nothing is pending and you can send immediately. While a command is in flight, poll_sdl returns Pending — keep polling until you receive a terminal status before sending another command.

SensorState

client.sensor_state() returns a snapshot of the latest heartbeat-derived sensor state in physical units. It is updated on each heartbeat, independently of the frame pipeline — it is not synchronized with any particular frame. Values are zero/default until the first heartbeat arrives; check state.last_heartbeat_frame > 0 before treating contents as valid.

Note: In v0.10.0, sensor state will be accessible directly from each frame. The standalone client.sensor_state() call is a transitional API.

Typical uses are sensor health monitoring, confirming SDL commands were applied, and logging device identity — not per-frame processing.

state = client.sensor_state()

# Device identity
print(state.device.device_id)        # e.g. "CAR-30-005"
print(state.device.fpga_version)     # e.g. "v1.2.3"
print(state.device.mcu_version)      # e.g. "v1.0.0"

# Current SDL configuration confirmed by sensor
print(state.sdl.device_state)        # e.g. SdlState.PointCloud
print(state.sdl.frame_rate_fps)      # e.g. 10.0
print(state.sdl.hfov_deg)            # e.g. 60.0

# Health / temperatures
print(state.health.fpga_temp_c)
print(state.health.clarity_board_temp_c)

# Frame counters
print(state.counters.total_frame_count)
print(state.counters.total_drops_count)
print(state.counters.any_drops_sticky)

# Timing
print(state.peaks_per_frame)
print(state.last_heartbeat_frame)

VoyantRecorder

Records frames to binary files with automatic splitting options.

recorder = VoyantRecorder(
    output_path="recording.bin",
    timestamp_filename=True,         # Add timestamp to filename
    frames_per_file=None,            # Split after N frames
    duration_per_file=None,          # Split after N seconds
    size_per_file_mb=None,           # Split after N megabytes
    max_total_frames=None,           # Stop after N total frames
    max_total_duration=None,         # Stop after N total seconds
    max_total_size_mb=None,          # Stop after N total megabytes
)

VoyantPlayback

Plays back recorded data with rate control.

playback = VoyantPlayback(
    rate=1.0,              # Playback speed (None = as fast as possible)
    loopback=False,        # Loop continuously
    filter_points=True,    # Remove invalid points
)
playback.open("recording.bin")

Frame data access

# NumPy arrays
xyz   = frame.xyz()            # (N x 3): [x, y, z]
xyzv  = frame.xyzv()           # (N x 4): [x, y, z, radial_vel]
sph   = frame.spherical()      # (N x 3): [range, azimuth, elevation]

# Pandas DataFrames (via voyant_api.pandas_utils)
from voyant_api.pandas_utils import frame_to_dataframe, frame_to_extended_dataframe
df          = frame_to_dataframe(frame)           # 7 columns
df_extended = frame_to_extended_dataframe(frame)  # 11 columns

# PCD PointCloud objects (via voyant_api.pcd_utils)
from voyant_api.pcd_utils import frame_to_pcd, frame_to_extended_pcd, save_frame_to_pcd
pc = frame_to_pcd(frame)           # 7 fields
pc = frame_to_extended_pcd(frame)  # 11 fields
save_frame_to_pcd(frame, "out.pcd")

# Frame metadata
print(frame.frame_index)
print(frame.timestamp)
print(frame.n_points)
print(frame.n_valid_points)

# Sensor state (health, SDL config, device info, calibration)
state = client.sensor_state()
print(state.device)      # DeviceInfo: serial, product, firmware versions
print(state.sdl)         # SdlDeviceState: confirmed FOV, fps, state
print(state.health)      # HealthState: temperatures, error words
print(state.counters)    # CounterState: frame/ramp/drop counts
print(state.calibration) # CalibrationState: doppler, chirp bandwidth, datum
print(state.dsp_header)  # DspHeaderState: timestamp, frame start toggle

---

Migrating from VoyantClient

VoyantClient is deprecated as of v0.5.0 and will be removed in a future release. Carbon sensor users should migrate to CarbonClient.

VoyantClient remains functional for Meadowlark sensors but will receive no new features or fixes.

The main differences:

	VoyantClient (deprecated)	CarbonClient
Config	Constructor kwargs	CarbonConfig object
Lifecycle	No explicit start	client.start() required
Shutdown	—	client.stop() / client.wait_for_shutdown()
Timestamps	use_msg_stamps	set_use_msg_timestamp() — default True for Carbon

Before:

client = VoyantClient(
    bind_addr="0.0.0.0:4444",
    group_addr="224.0.0.0",
    interface_addr="192.168.1.100",
    filter_points=True,
    use_msg_stamps=True,
)

while True:
    frame = client.try_receive_frame()
    if frame is not None:
        process(frame)

After:

config = CarbonConfig()
config.set_bind_addr("0.0.0.0:5678")
config.set_group_addr("224.0.0.0")
config.set_interface_addr("192.168.1.100")

client = CarbonClient(config)
client.start()

# Press Ctrl+C to stop
while client.is_running():
    frame = client.try_receive_frame()
    if frame is not None:
        process(frame)
    else:
        time.sleep(0.001)

client.stop()

---

Features

• High performance: Rust-based implementation with zero-copy data access
• NumPy integration: Direct conversion to NumPy arrays via frame.xyzv()
• Pandas support: DataFrame conversion via voyant_api.pandas_utils
• PCD support: Point Cloud Data export via voyant_api.pcd_utils
• Type hints: Full type annotations for IDE support (.pyi stubs included)
• Recording & playback: Save and replay sensor data with timestamp preservation
• Network streaming: Multicast UDP support for live sensor data
• SDL commands: Runtime sensor configuration via SdlCommand

Complete examples

Full example scripts are available in the voyant-sdk repository:

• client_example.py — Live data streaming with CarbonClient
• recorder_example.py — Recording with all options
• playback_example.py — Playback and processing
• pcd_conversion_example.py — Converting recordings to PCD files
• sdl_example.py — Sending SDL commands and polling for confirmation

System requirements

• Python: 3.9 or later
• Dependencies: NumPy 2.0+, Pandas 2.0+, pypcd4 1.4+
• Platforms: Linux, Windows, macOS
• Hardware: Carbon sensors require v0.5.0+. Meadowlark sensors use the deprecated VoyantClient.

Documentation

• Full documentation: https://voyant-photonics.github.io/
• Examples repository: https://github.com/Voyant-Photonics/voyant-sdk

Support

• Issues: https://github.com/Voyant-Photonics/voyant-sdk/issues
• Email: [e-mail]

License

Proprietary — for use with Voyant Photonics hardware products only.

Copyright © 2025 Voyant Photonics, Inc. All rights reserved.