voyant-api 0.12.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.

The recommended Python flow is to start from the latest heartbeat read back, modify only the fields you want to change, then call send_sdl_blocking(). This preserves the sensor's current SDL settings and, when the sensor is already streaming point clouds, applies the requested settings while transitioning through Idle, then resumes PointCloud unless the command requests Idle.

from voyant_api import (
    CarbonClient, CarbonConfig,
    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()

# Wait for the first heartbeat so sensor_state() reflects the live read back.
# See py/sdl_example.py for a complete polling helper.
state = client.sensor_state()
cmd = state.to_sdl_command()
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

# Blocks until the sensor confirms the command or the SDL timeout expires.
status = client.send_sdl_blocking(cmd)
if status == SdlStatus.Applied:
    print("Command applied.")
else:
    print(f"Command failed: {status}")

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. For read-modify-write updates, start from client.sensor_state().to_sdl_command() after the first heartbeat, then 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 SdlState, SdlRampLength

 # Assumes `client` has already been constructed, started, and has received
 # its first heartbeat before calling `sensor_state()` / `to_sdl_command()`.

state = client.sensor_state()
cmd = state.to_sdl_command()
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 – 20.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_blocking(cmd) for the recommended blocking path. It returns Applied on success or a failure status if any step fails. If the sensor is currently in PointCloud and the command requests Idle, it sends the command once and leaves the sensor in Idle. Otherwise, it sends the requested settings with req_state = Idle, then sends the same settings with req_state = PointCloud.

For event-loop or real-time callers that cannot block, use 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.13.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)

TimeSyncState

client.time_sync_state() returns a snapshot of the host↔FPGA clock synchronization health. The client keeps the FPGA clock aligned to the host in the background (enabled by default); this reports how well it is tracking. Reads SyncQuality.Unsynced until the first measurement lands.

from voyant_api import SyncQuality

ts = client.time_sync_state()

print(ts.quality)         # SyncQuality.{Unsynced, Poor, Fair, Good, Excellent}
print(ts.valid)           # True once a fresh measurement exists
print(ts.offset_ns)       # host − FPGA clock difference (ns); >0 means the FPGA is behind
print(ts.jitter_ns)       # std-dev of the offset over the rolling window (ns)
print(ts.round_trip_ns)   # register round-trip of the kept measurement (ns)
print(ts.sample_count)    # measurements in the rolling window
print(ts.age_ms)          # ms since the last fresh measurement

if ts.quality >= SyncQuality.Good:
    print("timestamps are trustworthy")

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 (blocking and non-blocking paths)

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.