voyant_api

Voyant API - Python bindings for Voyant point cloud data and services.

 1"""Voyant API - Python bindings for Voyant point cloud data and services."""
 2
 3# Import the native extension module first (this loads the Rust code)
 4from .voyant_api import VoyantFrame
 5from .voyant_api import VoyantPlayback
 6from .voyant_api import VoyantClient
 7from .voyant_api import VoyantRecorder
 8from .voyant_api import RecordStatus
 9from .voyant_api import init_voyant_logging
10from .voyant_api import create_default_frame
11from .voyant_api import create_test_frame
12from .voyant_api import CarbonClient
13from .voyant_api import CarbonConfig
14from .voyant_api import SdlCommand
15from .voyant_api import SdlState
16from .voyant_api import SdlRampLength
17from .voyant_api import SdlStatus
18from .voyant_api import DspFrameStartToggle
19from .voyant_api import SyncQuality
20from .voyant_api import TimeSyncState
21from .voyant_api import StreamTransport
22
23# Utility submodules are not imported here to avoid importing heavier
24# dependencies (such as pandas and pypcd4) unless they are actually needed.
25# Import explicitly as needed:
26#   from voyant_api import pandas_utils
27#   from voyant_api import pcd_utils
28
29__version__ = "0.16.0"
30__all__ = [
31    "CarbonClient",
32    "CarbonConfig",
33    "SdlCommand",
34    "SdlState",
35    "SdlRampLength",
36    "SdlStatus",
37    "SyncQuality",
38    "TimeSyncState",
39    "VoyantFrame",
40    "VoyantPlayback",
41    "VoyantClient",
42    "VoyantRecorder",
43    "RecordStatus",
44    "init_voyant_logging",
45    "create_default_frame",
46    "create_test_frame",
47    "DspFrameStartToggle",
48    "StreamTransport",
49    # "protos",
50    # "playback",
51    # "client",
52]
class CarbonClient:

Python client for receiving frames from a Carbon LiDAR sensor.

Example:

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() while client.is_running(): frame = client.try_receive_frame() if frame: process(frame) client.stop()

CarbonClient(config: CarbonConfig)

Create a CarbonClient from a CarbonConfig. Does not start receiving — call start().

def start(self) -> None:

Start receiving and processing data.

def stop(self) -> None:

Stop receiving and processing data. Safe to call multiple times. The client can be restarted by calling start() again.

def try_receive_frame(self) -> VoyantFrame | None:

Try to get the next frame from the buffer.

Returns:

VoyantFrame if a new frame is available, None otherwise. Frames are dropped silently with a log warning if the buffer fills up (i.e. this is not called fast enough to keep up with the sensor rate).

def sensor_state(self) -> voyant_api.SensorState:

Get the latest sensor state from the most recent heartbeat.

Returns a snapshot — call again to get updated values.

def time_sync_state(self) -> TimeSyncState:

Get the latest host↔FPGA time-sync state (quality, offset, jitter).

Returns a snapshot — call again to get updated values. Reads Unsynced until the background time-sync task records its first measurement (or if time-sync is disabled).

def is_running(self) -> bool:

Returns true if the client is running and no OS shutdown has been requested.

def wait_for_shutdown(self) -> None:

Block the calling thread until an OS shutdown signal or stop() is called.

def send_sdl(self, cmd: SdlCommand) -> SdlStatus:

Send an SDL command to the sensor FPGA without blocking.

This is the non-blocking path; drive poll_sdl() to confirm. For commands where blocking is acceptable, prefer send_sdl_blocking().

A PointCloud request is applied transparently as a two-step transition (settings applied in Idle, then PointCloud resumed), advanced by poll_sdl().

Returns:

SdlStatus.Pending — command accepted, awaiting heartbeat confirmation. SdlStatus.SendFailed — passed validation but the UDP send failed. SdlStatus.PreviousCommandPending — a command is already in flight. Any other SdlStatus — rejected before sending (e.g. SdlStatus.InvalidParameter).

def poll_sdl(self) -> SdlStatus:

Poll for SDL command confirmation after send_sdl().

This belongs with the lower-level non-blocking send_sdl() path. For normal SDL commands, prefer send_sdl_blocking().

Returns:

SdlStatus.Idle — no command is currently in flight. SdlStatus.Pending — still waiting for heartbeat confirmation. Any other SdlStatus — resolved status once the heartbeat confirms or the command times out.

def send_sdl_blocking(self, cmd: SdlCommand) -> SdlStatus:

Send an SDL command and block until it is confirmed or times out.

Use this for normal SDL commands: it sends the command, applies the requested settings, and waits for the sensor to confirm. A PointCloud request is applied transparently as a two-step (settings in Idle, then PointCloud); the timeout from CarbonConfig.sdl.timeout_sec is enforced per command, so a PointCloud request can take up to roughly twice that long.

Arguments:
  • cmd: SdlCommand to send.
Returns:

SdlStatus.Applied on success, or a failure status.

def wait_for_heartbeat(self) -> None:

Block until the sensor's first heartbeat arrives, or a fixed timeout elapses.

SDL confirmation and both background-noise calibration operations need a heartbeat first — the box serial and the frame counter used for confirmation come from it.

Raises:
  • RuntimeError if no heartbeat arrives in time.
def ensure_idle_for_calibration(self) -> None:

Drive the sensor to Idle using the fixed background-noise calibration config, blocking until confirmed. Both calibration operations require Idle on entry; this reaches it from any state.

Raises:
  • RuntimeError if the transition does not confirm.
def apply_default_background_noise(self) -> None:

Apply the compiled-in default background-noise calibration for the connected box, blocking until done.

Writes the built-in default mask. The sensor must be in Idle; this does not change the sensor's state. Blocks for a few seconds.

Raises:
  • RuntimeError if the sensor is not Idle, no calibration is embedded for the box, or a
  • write fails.
def refine_background_noise(self, iterations: int | None = None) -> None:

Refine the background-noise calibration for the connected box from a covered-window capture, blocking until done.

The sensor must be in Idle on entry and the window must be covered (the capture assumes background noise only). Cycles the sensor between Idle and Point Cloud internally and returns to Idle when finished. Blocks for roughly 10–20 seconds. While it runs, SDL commands and peak dumps on this client are rejected; do not run a second client against the sensor.

Arguments:
  • iterations: Refinement iterations; defaults internally when None.
Raises:
  • RuntimeError if the sensor is not Idle, no calibration is embedded for the box, or a
  • transition/capture/write fails.
def start_peak_dump( self, path: str | os.PathLike | pathlib.Path, max_frames: int | None = None, max_peaks: int | None = None, add_timestamp: bool | None = None) -> None:

Begin dumping the raw peak stream to a CSV file while the point cloud keeps running. The client must already be running (call start() first).

The first of max_frames / max_peaks to be hit ends the dump; with neither set it runs until stop_peak_dump(). The writer runs on its own thread, so this returns immediately.

Arguments:
  • path: Output CSV path. When add_timestamp is True, a timestamp is inserted into the filename.
  • max_frames: Stop after this many whole frames (None = unbounded).
  • max_peaks: Stop after this many CSV rows (None = unbounded).
  • add_timestamp: Insert a timestamp into the filename. None is treated as False (no timestamp).
Raises:
  • RuntimeError if the client is not running or a dump is already in progress.
def stop_peak_dump(self) -> None:

Request the active peak dump to stop. Returns immediately; the file ends on a whole frame. Poll is_peak_dumping() for completion.

def is_peak_dumping(self) -> bool:

Returns True while a peak dump is running.

class CarbonConfig:

Configuration for the Carbon LiDAR pipeline.

Construct either from a JSON file or with defaults and setters.

CarbonConfig()

Create a CarbonConfig with default values.

Example:

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) client = CarbonClient(config)

def from_json(path: str) -> CarbonConfig:

Load a CarbonConfig from a JSON file. Missing fields default to their standard values.

Example:

config = CarbonConfig.from_json("config.json") client = CarbonClient(config)

def set_bind_addr(self, v: str) -> None:

Set the UDP bind address including port (e.g. "0.0.0.0:5678")

def set_group_addr(self, v: str) -> None:

Set the multicast group address (e.g. "239.255.48.84")

def set_interface_addr(self, v: str) -> None:

Set the local interface IP for the multicast join (e.g. "192.168.1.100"; "0.0.0.0" lets the OS pick)

def set_observer_only(self, v: bool) -> None:

Observer-only mode: passively receive the point stream without sending anything to the FPGA. Disables SDL commands, host<->FPGA time-sync, and comms-health probing so multiple clients can share one sensor — but only with Multicast transport; under the default unicast the sensor streams to a single client. Exactly one primary (non-observer) client should own the FPGA. Default: False.

def set_stream_transport(self, v: StreamTransport) -> None:

How the sensor delivers its push stream: StreamTransport.Unicast (default; the sensor replies only to this client and arrives reliably even when the host has several network connections active at once) or StreamTransport.Multicast (streams to a shared group so several clients can view one sensor). A primary writes this to the FPGA on connect; observers ignore it.

def set_use_msg_timestamp(self, v: bool) -> None:

Use timestamps from received packets instead of system time on receipt.

def set_batch_size(self, v: int) -> None:

Maximum number of peaks messages per batch before flushing to the pipeline.

def set_recv_buffer_size(self, v: int) -> None:

Size of the pre-allocated UDP receive buffer in bytes.

def set_receiver_channel_capacity(self, v: int) -> None:

Channel capacity for batched messages from receiver to pipeline.

def set_vel_corr_factor(self, v: float) -> None:

Velocity correction multiplier.

def set_bandwidth_hz(self, v: float) -> None:

Optionally override the chirp sweep bandwidth in Hz.

def set_elevation_fov_deg(self, v: float) -> None:

Elevation field of view in degrees.

def set_peak_combine_twin_only(self, v: bool) -> None:

Use legacy literal-twin-only pairing (skips the neighborhood algorithm). Default: false.

def set_peak_combine_cluster_min_size(self, v: int) -> None:

Minimum inlier/survivor count for the peak combiner. Default: 1.

def set_peak_combine_beat_thresh_hz(self, v: float) -> None:

Static beat-frequency threshold in Hz. Overrides dynamic derivation when set. Pass 0.0 to clear the override and restore dynamic derivation.

def set_peak_combine_beat_outlier_removal(self, v: bool) -> None:

Enable beat outlier removal. Set false to disable for A/B comparisons. Default: true.

def set_peak_combine_snr_pre_threshold_enabled(self, v: bool) -> None:

Enable the per-peak SNR pre-threshold (stage 0). Default: true. Set false to disable the gate for A/B comparisons.

def set_peak_combine_snr_pre_threshold_max_db(self, v: float) -> None:

SNR pre-threshold (dB) at DC. Default: 13.0.

def set_peak_combine_snr_pre_threshold_min_db(self, v: float) -> None:

SNR pre-threshold floor (dB) for high beat frequencies. Default: 10.0.

def set_peak_combine_snr_pre_threshold_slope_db_per_hz(self, v: float) -> None:

SNR pre-threshold slope (dB / Hz). Default: 6.0e-7 (= 0.6 dB/MHz).

def set_keep_invalid_points(self, v: bool) -> None:

Keep invalid points (drop_reason != Success) in the assembled point cloud. Default: false (invalid points are filtered out).

def set_range_min(self, v: float) -> None:

Minimum range filter in meters.

def set_range_max(self, v: float) -> None:

Maximum range filter in meters.

def set_doppler_min(self, v: float) -> None:

Minimum Doppler filter in m/s.

def set_doppler_max(self, v: float) -> None:

Maximum Doppler filter in m/s.

def set_azimuth_deg_min(self, v: float) -> None:

Minimum azimuth filter in degrees.

def set_azimuth_deg_max(self, v: float) -> None:

Maximum azimuth filter in degrees.

def set_elevation_deg_min(self, v: float) -> None:

Minimum elevation filter in degrees.

def set_elevation_deg_max(self, v: float) -> None:

Maximum elevation filter in degrees.

def set_rd_outlier_range_thresh_min_m(self, v: float) -> None:

Range tolerance (m) at the origin. Default: 0.1.

def set_rd_outlier_range_thresh_max_m(self, v: float) -> None:

Range tolerance (m) at ref_range. Default: 0.4.

def set_rd_outlier_doppler_thresh_min_mps(self, v: float) -> None:

Doppler tolerance (m/s) at the origin. Default: 0.08.

def set_rd_outlier_doppler_thresh_max_mps(self, v: float) -> None:

Doppler tolerance (m/s) at ref_range. Default: 0.2.

def set_rd_outlier_ref_range_m(self, v: float) -> None:

Reference range (m) where both thresholds reach their *_max values. Default: 150.

def set_spatial_filter_enabled(self, v: bool) -> None:

Enable the point-stage spatial outlier filter. Default: true.

def set_spatial_min_valid_neighbors(self, v: int) -> None:

Minimum valid neighbors before either spatial check runs.

def set_spatial_point_cluster_size(self, v: int) -> None:

Minimum number of range- and Doppler-close neighbors required to keep a point. Typical: 1–2.

def set_spatial_support_threshold(self, v: float) -> None:

Drop the point when its own confidence plus the sum of confidence scores over Doppler-close neighbors falls below this. Typical: 1.0.

def set_interp_range_threshold(self, v: float) -> None:

Range threshold for elevation interpolation in meters.

def set_interp_doppler_threshold(self, v: float) -> None:

Doppler threshold for elevation interpolation in m/s.

def set_missing_elevations(self, v: Sequence[int]) -> None:

Missing elevation indices to interpolate (must be even: 0, 2, 4, … 254). Setting this enables elevation interpolation. Default: disabled (empty list).

def set_eval_mode(self, v: bool) -> None:

Enable timing and drop statistics evaluation with generated reports.

def set_report_dir(self, v: str) -> None:

Directory for evaluation report files. Only used when eval mode is enabled. Default: "./reports"

def set_fpga_target_addr(self, v: str) -> None:

FPGA target address in ip:port format. Default: "192.168.1.128:1234"

def set_sdl_timeout_sec(self, v: float) -> None:

Timeout for SDL commands in seconds. Default: 10.0

def set_sdl_max_retries(self, v: int) -> None:

Maximum number of retries for SDL commands before giving up. Default: 5

class SdlCommand:

SDL command to send to the sensor FPGA.

Construct with defaults and override only the fields 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 status = client.send_sdl(cmd)

For a static line (linescan), use the linescan() constructor: the beam parks (HFOV 0°) at a firmware-fixed line rate (no frame rate):

cmd = SdlCommand.linescan(SdlState.PointCloud, 6.0)
#                         req_state, ramp_bw_ghz
status = client.send_sdl(cmd)

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.

The per-field setters refuse to switch between swept and static-line mode; use configure() or linescan() for that. cmd.is_linescan reports whether a command is a static line.

SdlCommand()

Construct an SdlCommand with sensor power-on defaults.

def linescan( req_state: SdlState, ramp_bandwidth_ghz: float, hfov_center_deg: float | None = None) -> SdlCommand:

Build a static-line (linescan) command in one validated call.

HFOV is pinned to 0° (the mirror is parked). There is no frame-rate argument — the firmware fixes the line rate at 900 fps. Raises ValueError if any value is out of range.

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 / None; a non-zero value builds but is rejected by send_sdl.

cmd = SdlCommand.linescan(SdlState.PointCloud, 6.0)
status = client.send_sdl(cmd)
def configure( self, req_state: SdlState, frame_rate_fps: float, hfov_deg: float, hfov_center_deg: float, ramp_bandwidth_ghz: float) -> None:

Configure all fields in one validated call.

When hfov_deg == 0 the command is a static line and frame_rate_fps is ignored (prefer linescan()); otherwise it must be in [1.0, 20.0]. Raises ValueError if any value is out of range, leaving the command unmodified.

ramp_bandwidth_ghz: float

Ramp bandwidth in GHz (0.5 – 10.0, quantized to 0.1 GHz). Raises ValueError if out of range.

ramp_length: SdlRampLength

Ramp length (read-only). Firmware-fixed at 16.384 µs, so it is not settable.

frame_rate_fps: float

Frame rate in fps (1.0 – 20.0, quantized to 0.5 fps).

In linescan mode (is_linescan) this reads back the firmware-fixed line rate (900.0) — the wire field carries a sentinel, not a real fps. Setting it in linescan mode, or out of range, raises ValueError.

is_linescan: bool

True when this command is a static line (HFOV pinned to 0°). Enter linescan via linescan() or configure() with hfov_deg == 0.

hfov_deg: float

Horizontal FOV in degrees (0.0 – 120.0, quantized to 0.1°). Raises ValueError if out of range.

hfov_center_deg: float

Horizontal FOV center in degrees (−60.0 – 60.0, quantized to 0.1°). Raises ValueError if out of range.

Must currently be 0: send_sdl rejects any non-zero center. Beam steering is not yet supported — with a non-zero (swept) HFOV it is not yet implemented (a deferred feature), and in static-line mode (HFOV 0°) it is temporarily blocked by firmware (MCU v2.5.0 / FPGA v1.3.3), which centers the mirror regardless.

req_state: SdlState

Requested sensor operating state.

class SdlState:

Enum for sdl::set_state register

Unknown = SdlState.Unknown
ErrorFault = SdlState.ErrorFault
Boot = SdlState.Boot
Initialization = SdlState.Initialization
Standby = SdlState.Standby
WarmUp = SdlState.WarmUp
Idle = SdlState.Idle
PointCloud = SdlState.PointCloud
CoolDown = SdlState.CoolDown
DebugCalibration = SdlState.DebugCalibration
Reserved = SdlState.Reserved
ReservedV11 = SdlState.ReservedV11
ReservedV12 = SdlState.ReservedV12
ReservedV13 = SdlState.ReservedV13
ReservedV14 = SdlState.ReservedV14
FirmwareUpdate = SdlState.FirmwareUpdate
class SdlRampLength:

Enum for sdl::set_ramp_length register

class SdlStatus:

Outcome of an SDL command, either from the sensor heartbeat or detected client-side.

The raw msg_status wire values from the sensor are mapped here in from_msg_status — the discriminants in this enum are contiguous and have no relationship to the wire protocol values.

Unknown = SdlStatus.Unknown

Unrecognized or default wire status from the sensor heartbeat. Not returned by the send/poll API in normal operation.

Idle = SdlStatus.Idle

No SDL command is currently in flight. Returned by poll_sdl() when called with nothing pending.

Pending = SdlStatus.Pending

Command sent, awaiting sensor confirmation via heartbeat. Returned by send_sdl() on a successful UDP send, or by poll_sdl() while waiting for heartbeat confirmation.

Applied = SdlStatus.Applied

Sensor confirmed the command was successfully applied.

BadFovCenterCombo = SdlStatus.BadFovCenterCombo

The FOV and center combination is not valid. May be rejected client-side before sending, or by the sensor.

InvalidParameter = SdlStatus.InvalidParameter

A parameter value is out of range or otherwise invalid. May be rejected client-side before sending, or by the sensor.

InvalidStateTransition = SdlStatus.InvalidStateTransition

The requested state transition is not permitted from the current state.

MissingCalibration = SdlStatus.MissingCalibration

Sensor cannot apply the command due to missing calibration data.

UnableToCalibrate = SdlStatus.UnableToCalibrate

Sensor attempted calibration but was unable to complete it.

ParseError = SdlStatus.ParseError

Sensor could not parse the SDL message.

ApplicationError = SdlStatus.ApplicationError

Sensor parsed the command but failed to apply it.

FovFpsError = SdlStatus.FovFpsError

The FOV/FPS combination exceeds hardware performance limits. Command was not sent.

CommandBuildFailed = SdlStatus.CommandBuildFailed

The SDL command could not be constructed from the supplied parameters.

PreviousCommandPending = SdlStatus.PreviousCommandPending

A previous command is still awaiting confirmation; wait for it to resolve before sending another.

SendFailed = SdlStatus.SendFailed

The underlying UDP send failed. Check logs for detail.

Timeout = SdlStatus.Timeout

No heartbeat confirmation arrived within the configured timeout window.

MaxRetriesExceeded = SdlStatus.MaxRetriesExceeded

Command was retransmitted the maximum number of times without confirmation.

StreamReset = SdlStatus.StreamReset

Heartbeat frame counter jumped backwards — stream was reset.

MessageReadyForMcu = SdlStatus.MessageReadyForMcu

Sensor has received the message and it is ready for MCU processing.

ValidMessageParsed = SdlStatus.ValidMessageParsed

Sensor parsed a valid SDL message but did not apply the requested change.

ObserverOnly = SdlStatus.ObserverOnly

The client is in observer-only mode and cannot issue SDL commands; nothing was sent. A primary (non-observer) client owns the sensor. Appended out of group order to keep discriminants stable (the values are exposed to Python via eq_int).

class SyncQuality:

Qualitative health of the host↔FPGA clock synchronization, judged by the absolute clock difference |host − FPGA|. Ordered worst-to-best (Unsynced … Excellent), so ordering comparisons (e.g. "at least Good") are meaningful.

Unsynced = SyncQuality.Unsynced

No fresh measurement — link down, or never measured.

Poor = SyncQuality.Poor

Off by more than 1 ms; timestamps are unreliable.

Fair = SyncQuality.Fair

Within 1 ms but worse than 100 µs; usable, but watch for drift.

Good = SyncQuality.Good

Within 100 µs but worse than 1 µs of the host; trustworthy.

Excellent = SyncQuality.Excellent

Within 1 µs of the host — sub-microsecond alignment.

class TimeSyncState:

Host↔FPGA time-sync health: quality verdict plus the measured offset, jitter, and round-trip. Read from CarbonClient.time_sync_state().

round_trip_ns: int

Round-trip of the kept read for the latest measurement, nanoseconds.

age_ms: int

Milliseconds since the last fresh measurement (very large when never measured).

jitter_ns: int

Standard deviation of the offset over the rolling window, nanoseconds.

valid: bool

True when the latest measurement is fresh (within the manager's stale window).

offset_ns: int

Most recent host−FPGA clock difference, nanoseconds (>0 ⇒ the FPGA clock is behind).

quality: SyncQuality
sample_count: int

Number of measurements currently in the rolling window.

class VoyantFrame:

Python-exposed frame containing point cloud data

def xyz_columns() -> list[str]:

Get column names for the xyz() array

def xyzv_columns() -> list[str]:

Get column names for the xyzv() array

def spherical_columns() -> list[str]:

Get column names for the spherical() array

def spherical_v_columns() -> list[str]:

Get column names for the spherical_v() array

def points_columns() -> list[str]:

Get column names for the points() array

def points_extended_columns() -> list[str]:

Get column names for the points_extended() array

def special_test_columns() -> list[str]:

Get column names for the special_test() array

def xyz(self) -> NDArray[numpy.float32]:

Get XYZ coordinates as (N, 3) array

def valid_xyz(self) -> NDArray[numpy.float32]:

Get XYZ coordinates as (N, 3) array, filtering out invalid points If invalid points are already filtered out, this return is identical to: xyz()

def xyzv(self) -> NDArray[numpy.float32]:

Get XYZ + radial velocity as (N, 4) array

def valid_xyzv(self) -> NDArray[numpy.float32]:

Get XYZ + radial velocity as (N, 4) array, filtering out invalid points If invalid points are already filtered out, this return is identical to: xyzv()

def spherical(self, degrees: bool | None = None) -> NDArray[numpy.float32]:

Get spherical coordinates as (N, 3) array [range, azimuth, elevation]

Arguments:
  • degrees (bool): If True, angles in degrees; if False, angles in radians. Default: False (radians)
def valid_spherical(self, degrees: bool | None = None) -> NDArray[numpy.float32]:

Get spherical coordinates as (N, 3) array, filtering out invalid points

Arguments:
  • degrees (bool): If True, angles in degrees; if False, angles in radians. Default: False (radians)
def spherical_v(self, degrees: bool | None = None) -> NDArray[numpy.float32]:

Get spherical coordinates + velocity as (N, 4) array [range, azimuth, elevation, radial_vel]

Arguments:
  • degrees (bool): If True, angles in degrees; if False, angles in radians. Default: False (radians)
def valid_spherical_v(self, degrees: bool | None = None) -> NDArray[numpy.float32]:

Get spherical coordinates + velocity as (N, 4) array, filtering out invalid points

Arguments:
  • degrees (bool): If True, angles in degrees; if False, angles in radians. Default: False (radians)
def points(self) -> NDArray[numpy.float32]:

Get points as a 2D array (N, 7) with columns: [x, y, z, radial_vel, snr_linear, nanosecs_since_frame, drop_reason]

def valid_points(self) -> NDArray[numpy.float32]:

Get all point data as (N, 7) array, filtering out invalid points If invalid points are already filtered out, this return is identical to: points()

def points_extended(self) -> NDArray[numpy.float32]:

Get points as a 2D array (N, 11) with columns: [x, y, z, radial_vel, snr_linear, nanosecs_since_frame, drop_reason, calibrated_reflectance, noise_mean_estimate, min_ramp_snr, point_index]

def valid_points_extended(self) -> NDArray[numpy.float32]:

Get all point data as (N, 11) array, filtering out invalid points If invalid points are already filtered out, this return is identical to: points_extended()

def special_test(self) -> NDArray[numpy.float32]:

Get points as a 2D array (N, 5) with columns: [azimuth_idx, elevation_idx, range, doppler, snr]

This is designed for internal systems test team usage

def valid_special_test(self) -> NDArray[numpy.float32]:

Get special test data as (N, 5) array, filtering out invalid points If invalid points are already filtered out, this return is identical to: special_test()

def valid_mask(self) -> list[bool]:

Get boolean mask of valid points

def describe(self) -> None:

Print detailed frame information (like the Rust print_info)

timestamp: float
device_id: str
n_points: int
n_valid_points: int
frame_index: int
class VoyantPlayback:

Python wrapper for VoyantPlayback Provides simple frame-by-frame playback control

VoyantPlayback( rate: float | None = None, loopback: bool | None = None, filter_points: bool | None = None)

Create a new VoyantPlayback instance

Arguments:
  • rate: Playback rate (1.0 = real-time, None = as fast as possible, default: None)
  • loopback: Whether to loop when reaching the end of file (default: false)
  • filter_points: Whether to filter out invalid points during playback (default: true)
def open(self, file_path: str) -> None:

Open a file for playback

Arguments:
  • file_path: Path to the recording file
def reset(self) -> None:

Reset playback to the beginning of the file

is_open: bool

Check if a file is currently open

frame_index: int

Get the current frame index

rate: float | None

Get playback rate

loopback: bool

Get loopback setting

frame_timestamp: float

Get the current frame timestamp in seconds

class VoyantClient:

Python wrapper for PointsClient Provides simple interface for receiving frames over UDP

VoyantClient( bind_addr: str, group_addr: str, interface_addr: str, filter_points: bool | None = None, use_msg_stamps: bool | None = None, track_timing: bool | None = None, disable_spn: bool | None = None)

Create a new multicast client

Arguments:
  • bind_addr: Local socket address to bind to (e.g., "0.0.0.0:4444")
  • group_addr: Multicast group address (e.g., "224.0.0.0")
  • interface_addr: Interface address (e.g., "127.0.0.1")
  • filter_points: Whether to filter out invalid points (default: true)
  • use_msg_stamps: Whether to use timestamps from received point groups [false = use system time] (default: false)
  • track_timing: Whether to collect statistics on packet timing and latency (default: false)
  • disable_spn: Whether to disable Single Point Noise filtering (default: false)
Warning:

Do NOT set disable_spn=True with Carbon Dev Kit (Meadowlark) units!

def try_receive_frame(self) -> VoyantFrame | None:

Try to get the latest frame if available

Returns:

VoyantFrame if new data is available, None otherwise

class VoyantRecorder:

Python wrapper for VoyantRecorder Provides interface for recording frames to disk with automatic file splitting

VoyantRecorder( output_path: str, timestamp_filename: bool | None = None, frames_per_file: int | None = None, duration_per_file: int | None = None, size_per_file_mb: int | None = None, max_total_frames: int | None = None, max_total_duration: int | None = None, max_total_size_mb: int | None = None, buffer_size_mb: int | None = None)

Create a new VoyantRecorder instance

Arguments:
  • output_path: Base path for output recording file(s)
  • timestamp_filename: Whether to add timestamp to filename (default: true)
  • frames_per_file: Maximum frames per file before splitting (None = no limit)
  • duration_per_file: Maximum seconds per file before splitting (None = no limit)
  • size_per_file_mb: Maximum MB per file before splitting (None = no limit)
  • max_total_frames: Maximum total frames across all files (None = no limit)
  • max_total_duration: Maximum total seconds across all files (None = no limit)
  • max_total_size_mb: Maximum total MB across all files (None = no limit)
  • buffer_size_mb: Frame buffer size in MB for writing (default: 4, advanced users only)
def record_frame(self, frame: VoyantFrame) -> RecordStatus:

Record a frame to disk

Arguments:
  • frame: VoyantFrame to record
Returns:

RecordStatus indicating the result:

def finalize(self) -> None:

Finalize the recording and close all files

This should be called when recording is complete to ensure all data is written and files are properly closed.

split_count: int

Get the number of files created (split count)

frames_recorded: int

Get the total number of frames recorded across all files

is_active: bool

Check if the recorder is still active (not finalized)

class RecordStatus:

Simple result status for recording operations

OK: RecordStatus = RecordStatus.OK

Frame recorded successfully, continue recording

SPLIT: RecordStatus = RecordStatus.SPLIT

Frame recorded, but file was split due to reaching limits

STOP: RecordStatus = RecordStatus.STOP

Recording limit reached, should stop recording

def init_voyant_logging() -> None:

Initialize Voyant logging system from Python

This function initializes the Rust logging system for use in Python applications. It sets up logging to stderr with INFO level as default, but respects the RUST_LOG environment variable for custom configuration.

The logging system supports multiple levels:

  • TRACE: Very detailed diagnostic information for fine-grained debugging
  • DEBUG: Detailed diagnostic information
  • INFO: General informational messages (default)
  • WARN: Warning messages for potentially harmful situations
  • ERROR: Error messages for failures

Note: This function should be called only once, typically during application startup. Calling it multiple times will cause the program to panic.

Environment Variables: Set RUST_LOG environment variable to control logging:

  • RUST_LOG=trace - Enable trace level logging (most verbose)
  • RUST_LOG=debug - Enable debug level logging
  • RUST_LOG=info - Enable info level logging (default)
  • RUST_LOG=warn - Enable warning level logging only
  • RUST_LOG=error - Enable error level logging only
  • RUST_LOG=off - Disable all logging
Raises:
  • RuntimeError: If logging has already been initialized
def create_default_frame() -> VoyantFrame:

Create a default, empty VoyantFrame for Python

This function creates a minimal frame with all default values. Useful for testing serialization/deserialization or as a starting point for manual frame construction.

Returns:

PyVoyantFrame: An empty frame with default header and no points

def create_test_frame( timestamp_seconds: int | None, timestamp_nanoseconds: int | None, frame_index: int | None, elevations: int | None, azimuths: int | None, range: float | None, drop_center: bool | None) -> VoyantFrame:

Create a test VoyantFrame with synthetic point cloud data arranged in a curved wall pattern

This function generates a complete frame with realistic point cloud data for testing purposes. The points are arranged in a curved wall pattern similar to what a real LiDAR sensor would produce when scanning a wall at a fixed distance. Each point includes proper timing information, with timestamps incrementing every 8 points to simulate the sensor's parallel channel acquisition.

Arguments:
  • timestamp_seconds (Optional[int]): Unix timestamp seconds for the frame header. Defaults to 0. Use actual timestamps for time-based testing.
  • timestamp_nanoseconds (Optional[int]): Nanoseconds portion of the timestamp (0-999999999). Defaults to 0. Combined with timestamp_seconds for precise frame timing.
  • frame_index (Optional[int]): Sequential frame number for tracking frame order. Defaults to 0. Should increment for each frame in a sequence.
  • elevations (Optional[int]): Number of elevation scan lines in the point cloud. Defaults to 32. Higher values create denser vertical resolution.
  • azimuths (Optional[int]): Number of azimuth points per elevation line. Defaults to 720. Higher values create denser horizontal resolution.
  • range (Optional[float]): Distance to the wall in meters. Defaults to 10.0. All points will be at approximately this distance.
  • drop_center (Optional[bool]): Whether to mark center region points as invalid. Defaults to False. When True, simulates obstruction or invalid returns in FOV center.
Returns:

PyVoyantFrame: A complete frame with header, config, and point cloud data. The frame will contain elevations * azimuths total points arranged in a grid pattern.

Notes:
  • Point timestamps increment by 16384 nanoseconds every 8 points to simulate the sensor's 8-channel parallel acquisition pattern
  • The generated header includes device ID "SIM-001" to indicate synthetic data
  • All version fields (proto, API, firmware, HDL) are populated with test values
  • Points are arranged in elevation-first scan order (all elevations at first azimuth, then move to next azimuth)
  • When drop_center is True, approximately 25% of the center FOV will have invalid points
class DspFrameStartToggle:

Enum for dsp::frame_start_toggle register

class StreamTransport:

How the sensor delivers its UDP push stream, selected by what a primary client writes to network.push_dest_ip on connect.

Unicast = StreamTransport.Unicast

Sensor sends the stream straight back to this client (reply-to-last-sender). The default: it arrives reliably even when the host has several network connections active at once (e.g. wired plus Wi-Fi).

Multicast = StreamTransport.Multicast

Sensor sends the stream to a shared group so several clients can view one sensor. Opt-in: the host's network must be set up to deliver multicast to the right connection.