voyant-api 0.14.0

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("239.255.48.84")
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("239.255.48.84")
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, SdlStatus,
    init_voyant_logging,
)

init_voyant_logging()

config = CarbonConfig()
config.set_bind_addr("0.0.0.0:5678")
config.set_group_addr("239.255.48.84")
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

# 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("239.255.48.84")
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")

# Observer-only: passively receive the point stream without sending anything to the
# sensor (no SDL, time-sync, or comms-health). Use this when another (primary) client
# already owns the sensor, so multiple clients can share one multicast stream. Exactly
# one primary (non-observer) client should be connected.
config.set_observer_only(True)

# Stream transport: how the sensor delivers its push stream. Unicast (the default) makes
# the sensor reply only to this client and arrives reliably even when the host has several
# network connections active at once. Multicast streams to a shared group so several clients
# can view one sensor (use observer-only for the extra ones).
from voyant_api import StreamTransport
config.set_stream_transport(StreamTransport.Multicast)

# 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: "239.255.48.84", interface_addr: "192.168.1.100" }, 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

 # 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)

(ramp_length is firmware-fixed at 16.384 µs, so it is read-only — exposed as cmd.ramp_length for inspection but not settable.)

You can also build a command in one validated call with configure():

cmd = SdlCommand()
cmd.configure(SdlState.PointCloud, 10.0, 60.0, 0.0, 6.0)
#             req_state, frame_rate_fps, hfov_deg, hfov_center_deg, ramp_bandwidth_ghz

Static line (linescan). Use the linescan() constructor to park the beam (HFOV 0°). There is no frame-rate argument — the firmware fixes the line rate at 900 fps. The per-field hfov_deg/frame_rate_fps setters raise if used to switch between swept and static-line mode; use linescan()/configure() for that. cmd.is_linescan reports whether a command is a static line.

Note: Linescan beam steering is temporarily unsupported (MCU firmware v2.5.0 / FPGA v1.3.3) — the sensor centers the mirror regardless — so hfov_center_deg must be 0.0; send_sdl rejects a non-zero center.

cmd = SdlCommand.linescan(SdlState.PointCloud, 6.0)
#                         req_state, ramp_bandwidth_ghz

In linescan mode cmd.frame_rate_fps reads back the firmware-fixed line rate 900.0 — the wire field carries a sentinel, not a real fps. A static-line SdlDeviceState reports the same frame_rate_fps == 900.0 on read-back.

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.

Background-noise calibration

Two blocking operations configure the sensor's background-noise mask. The box is identified from the sensor's heartbeat — no serial argument needed. Both require the sensor to be in Idle and leave it in Idle (they do not start streaming):

• client.wait_for_heartbeat() — block until the first heartbeat arrives (needed for the box serial). Raises RuntimeError on timeout.
• client.ensure_idle_for_calibration() — drive the sensor to Idle from any state (both operations require it). Raises RuntimeError if the transition does not confirm.
• client.apply_default_background_noise() — write the compiled-in default mask for the connected box. Fast (a few seconds); does not change the sensor's state.
• client.refine_background_noise(iterations=None) — refine the mask from a covered-window capture. The sensor window must be covered first (the capture assumes background noise only). Cycles the sensor internally and returns it to Idle (~10–20 s). iterations defaults when omitted.

# Both operations require a heartbeat (for the box serial) and Idle on entry.
client.wait_for_heartbeat()
client.ensure_idle_for_calibration()

# Apply the built-in default calibration.
client.apply_default_background_noise()

# Or refine it (cover the sensor window first).
client.refine_background_noise()

All four raise RuntimeError if a precondition fails (sensor not Idle, no calibration embedded for the box, no heartbeat, or the operation itself fails).

While a calibration is running, SDL commands and peak dumps on the same client are rejected (the client drives the sensor itself during refine). Do not run a second client against the same sensor while calibrating — the two would issue conflicting commands.

Peak dumps

Dump the raw peak stream to a CSV file while the point cloud keeps streaming. The client must be running (start() first); the writer runs on its own thread, so start_peak_dump() returns immediately.

• client.start_peak_dump(path, max_frames=None, max_peaks=None, add_timestamp=False) — begin a dump. The first of max_frames / max_peaks to be hit ends it; with neither it runs until stop_peak_dump(). add_timestamp inserts a timestamp into the filename. Raises RuntimeError if the client is not running or a dump is already in progress.
• client.stop_peak_dump() — request the active dump to stop. Returns immediately; the file ends on a whole frame.
• client.is_peak_dumping() — True while a dump is running.

import time

client.start()
client.start_peak_dump("peaks.csv", max_frames=100)

# The point cloud keeps flowing during the dump; keep draining frames until the
# bounded dump completes on its own. To end an unbounded dump, call
# client.stop_peak_dump() and poll is_peak_dumping() until it returns False.
while client.is_peak_dumping():
    frame = client.try_receive_frame()
    if frame is None:
        time.sleep(0.001)

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 v1.0.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("239.255.48.84")
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)
• peak_dump_example.py — Dumping the raw peak stream to CSV while streaming

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.