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# MAINTENANCE: every #[pyclass] registered in Rust must be re-exported here and
 4# listed in __all__. The generated __init__.pyi declares the whole native module,
 5# so a class missing from this file still type-checks while failing at runtime
 6# (`from voyant_api import X` -> ImportError). Adding a pyclass? Add it below too.
 7
 8# Import the native extension module first (this loads the Rust code)
 9from .voyant_api import VoyantFrame
10from .voyant_api import VoyantPlayback
11from .voyant_api import VoyantRecorder
12from .voyant_api import RecordStatus
13from .voyant_api import init_voyant_logging
14from .voyant_api import create_default_frame
15from .voyant_api import create_test_frame
16from .voyant_api import decode_sensor_state
17from .voyant_api import api_version
18from .voyant_api import interface_contract_version
19from .voyant_api import CarbonClient
20from .voyant_api import CarbonConfig
21from .voyant_api import SdlCommand
22from .voyant_api import SdlState
23from .voyant_api import SdlRampLength
24from .voyant_api import SdlStatus
25from .voyant_api import DspFrameStartToggle
26from .voyant_api import SyncQuality
27from .voyant_api import TimeSyncState
28from .voyant_api import StreamTransport
29from .voyant_api import ProductId
30
31# Per-frame / per-heartbeat state snapshots, reachable from
32# VoyantFrame.sensor_state(), VoyantFrame.host_state(), and
33# CarbonClient.sensor_state(). Exported so they can be named in annotations.
34from .voyant_api import SensorState
35from .voyant_api import DeviceInfo
36from .voyant_api import HealthState
37from .voyant_api import CounterState
38from .voyant_api import CalibrationState
39from .voyant_api import DspHeaderState
40from .voyant_api import SdlDeviceState
41from .voyant_api import HostState
42
43# Utility submodules are not imported here to avoid importing heavier
44# dependencies (such as pandas and pypcd4) unless they are actually needed.
45# Import explicitly as needed:
46#   from voyant_api import pandas_utils
47#   from voyant_api import pcd_utils
48
49__version__ = "1.0.0"
50__all__ = [
51    "CarbonClient",
52    "CarbonConfig",
53    "SdlCommand",
54    "SdlState",
55    "SdlRampLength",
56    "SdlStatus",
57    "SyncQuality",
58    "ProductId",
59    "TimeSyncState",
60    "SensorState",
61    "DeviceInfo",
62    "HealthState",
63    "CounterState",
64    "CalibrationState",
65    "DspHeaderState",
66    "SdlDeviceState",
67    "HostState",
68    "VoyantFrame",
69    "VoyantPlayback",
70    "VoyantRecorder",
71    "RecordStatus",
72    "init_voyant_logging",
73    "create_default_frame",
74    "create_test_frame",
75    "decode_sensor_state",
76    "api_version",
77    "interface_contract_version",
78    "DspFrameStartToggle",
79    "StreamTransport",
80]
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) -> 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 diagnostic captures 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_diagnostic_capture( self, path: str | os.PathLike | pathlib.Path, overwrite: bool | None = None) -> None:

Begin a diagnostic capture for Voyant support: the raw peak stream is written to a .vynt sidecar log while the point cloud keeps running. Record frames alongside it with VoyantRecorder — together they form the support bundle.

Requires diagnostic_mode in the client config (set before start()), so the recorded frames carry the invalid points the sidecar explains. The capture runs until stop_diagnostic_capture(). The writer runs on its own thread, so this returns immediately.

Arguments:
  • path: Output .vynt path, used verbatim. Name it after the frame recording it accompanies, e.g. my_recording_peaks.vynt.
  • overwrite: Allow overwriting an existing sidecar file (default: False — fail instead). Match the recorder's overwrite setting.
Raises:
  • RuntimeError if diagnostic_mode is off, the client is not running, or a capture is already in progress.
def stop_diagnostic_capture(self) -> None:

Request the active diagnostic capture to stop. Returns immediately; the file ends on a whole frame. Poll is_diagnostic_capturing() for completion.

def is_diagnostic_capturing(self) -> bool:

Returns True while a diagnostic capture is running.

def diagnostic_capture_error(self) -> str | None:

Why the last diagnostic capture ended badly, if it did. None for a capture that completed or is still running, so a caller polling is_diagnostic_capturing() can tell a finished capture from a broken one.

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_diagnostic_mode(self, v: bool) -> None:

Diagnostic capture: include invalid points (with their drop reason) in frames. Default: false (valid points only).

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_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 — 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 — 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_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.

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, which centers the mirror regardless.

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.

req_state: SdlState

Requested sensor operating state.

hfov_deg: float

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

ramp_bandwidth_ghz: float

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

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 ProductId:

Hardware and simulation product identifiers.

Sent in the heartbeat product_id field. Values 250–252 are FPGA/RTL simulators; 253 is Isaac Sim; 254 is the basic software simulator; 255 is reserved.

Unknown or future variants are preserved as Unknown so that the rest of the pipeline never panics on an unrecognised ID.

Unknown = ProductId.Unknown
Meadowlark = ProductId.Meadowlark
CarbonBenchtop = ProductId.CarbonBenchtop
Carbon30 = ProductId.Carbon30
VivadoSimulator = ProductId.VivadoSimulator
VerilatorSimulator = ProductId.VerilatorSimulator
XceliumSimulator = ProductId.XceliumSimulator
IsaacSim = ProductId.IsaacSim
SoftwareSimulator = ProductId.SoftwareSimulator

Basic software simulator (lightweight, no physics/FPGA).

class TimeSyncState:

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

quality: SyncQuality
valid: bool

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

round_trip_ns: int

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

sample_count: int

Number of measurements currently in the rolling window.

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.

offset_ns: int

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

class SensorState:

Full sensor state snapshot from the most recent heartbeat, in physical units.

def to_sdl_command(self) -> SdlCommand:

The type of the None singleton.

last_heartbeat_frame: int
calibration: CalibrationState
peaks_per_frame: int
counters: CounterState
health: HealthState
device: DeviceInfo
dsp_header: DspHeaderState
class DeviceInfo:

Device identity and firmware versions.

mcu_version_major: int
mcu_version_patch: int
fpga_version_patch: int
fpga_version_major: int
fpga_version_minor: int
product_id: ProductId
serial_number: int
device_id: str

Formatted device ID string, e.g. "CAR-30-005".

mcu_version_minor: int
fpga_version: str

Formatted FPGA version string, e.g. "v1.2.3".

mcu_version: str

Formatted MCU version string, e.g. "v1.2.3".

class HealthState:

Sensor health in physical units.

adc_temp_c: float
soa_temp_c: float
pic_temp_c: float
carat_board_temp_c: float
error_word: int
power_health: int
ll_fom_smoothed: float
lo_power_mon: int
fpga_temp_c: float
hardware_health: int
clarity_board_temp_c: float
ll_fom: float
class CounterState:

Frame and ramp counters.

total_frame_count: int
total_drops_count: int
total_ramp_count: int
mcu_cycles_counter: int
any_drops_sticky: bool

True if any drops have occurred since sensor boot.

class CalibrationState:

Calibration data in physical units.

encoder_eccentricity_coeffs: list[float]

Encoder eccentricity polynomial coefficients (6 values, constant term first).

reflectance_poly_coeffs: list[float]

Reflectance polynomial coefficients (4 values).

encoder_azimuth_offset: int
datum_delta_y_m: float
reflectance_channel_scales: list[float]

Reflectance channel scale factors for channels 1..=7 relative to channel 0.

datum_delta_z_m: float
chirp_bandwidth_hz: float
doppler_calibration_mirror_bias: float
datum_delta_x_m: float
doppler_calibration_mirror_offset: float
class DspHeaderState:

DSP header fields.

frame_start_toggle: DspFrameStartToggle
timestamp_seconds: int
timestamp_nanoseconds: int

Sub-second nanoseconds remainder (0..999_999_999).

class SdlDeviceState:

SDL configuration confirmed by the sensor, in physical units.

device_state: SdlState
sdl_status: SdlStatus
hfov_deg: float
frame_rate_fps: float

Confirmed line rate in fps. In static-line mode (hfov_deg == 0) this reports the firmware-fixed linescan line rate (900 fps), not the swept-mode frame rate.

hfov_center_deg: float
ramp_bandwidth_ghz: float
ramp_length: SdlRampLength
class HostState:

Host-side context stamped when a frame was published — what the client knew at that moment. Zero fields mean "nothing to report": sources without host context (e.g. recordings from other producers) read as all zeros.

state_age_frames: int

Frames the sensor-state snapshot lags this frame; 0 = fresh, positive = stale snapshot (dropped heartbeats), negative = counter ahead of the rollover rail (e.g. after mid-stream frame loss).

sync_jitter_ns: int

Standard deviation of the offset over the measurement window, nanoseconds; saturates at 4294967295 (TimeSyncState.jitter_ns is the unclamped value).

sync_offset_ns: int

Most recent host − FPGA clock difference, nanoseconds.

dropped_ramps: int

Data-plane ramp drops within this frame.

sync_round_trip_ns: int

Round-trip of the kept read, nanoseconds; saturates at 4294967295.

sync_valid: bool

True when the sync_* fields hold a fresh time-sync measurement; when False they hold the last stale measurement (or zeros if never measured).

class VoyantFrame:

Python-exposed point-cloud frame: sensor state + host state + points.

VoyantFrame()

An empty frame that declares no state, so carries_state is False and sensor_state() / host_state() return None.

def points_columns(cartesian: bool | None = None) -> list[str]:

Column names for points() / valid_points(), one per recorded point field. Pass the same cartesian value you pass to points().

def xyz_columns() -> list[str]:

Column names for xyz() / valid_xyz().

def xyzv_columns() -> list[str]:

Column names for xyzv() / valid_xyzv().

def points(self, cartesian: bool | None = None) -> NDArray[numpy.float64]:

All points as an (N, 12) float64 array, one column per recorded point field: [range_m, azimuth_rad, elevation_rad, doppler_mps, snr, calibrated_reflectance, timestamp_nanosecs, azimuth_idx, elevation_idx, drop_reason, combine_method, user_data].

With cartesian=True the leading geometry columns swap to [x, y, z] in meters, derived from the stored spherical coordinates (equal to xyz()).

  • range_m — meters from the lidar datum; azimuth_rad / elevation_rad — radians, 0 at boresight, positive left and up
  • doppler_mps — meters/second, positive moving away from the sensor
  • snr — linear (not dB) SNR of the combined chirp pair
  • calibrated_reflectance — dB
  • timestamp_nanosecs — nanoseconds since frame start
  • azimuth_idx / elevation_idx — scan grid position; (0, 0) is the frame's top-left, matching camera image conventions
  • drop_reason — numeric code; 1 = valid point (see valid_points()), any other value means the point was dropped
  • combine_method — numeric code; 0 = unknown, non-zero identifies how the point was produced
  • user_data — the point's two user-owned bytes read as one little-endian u16 (0–65535); opaque to the API

float64 represents every recorded field exactly, so the default (spherical) matrix is the recorded data with no precision lost.

Includes invalid returns, which reach a recording from a diagnostic-mode capture or from frames built by hand; with neither, this matches valid_points().

def valid_points(self, cartesian: bool | None = None) -> NDArray[numpy.float64]:

points() restricted to valid returns (drop_reason == 1).

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

Cartesian positions as an (N, 3) float32 array [x, y, z] in meters, derived from the stored spherical coordinates (+x forward, +y left, +z up). Includes invalid returns — see points().

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

xyz() restricted to valid returns.

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

Cartesian positions plus Doppler as an (N, 4) float32 array [x, y, z, doppler_mps]. Includes invalid returns — see points().

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

xyzv() restricted to valid returns.

def with_points( self, matrix: NDArray[numpy.float64], cartesian: bool | None = None) -> VoyantFrame:

A new frame holding matrix as its points; everything else — state snapshots, carries_state, timestamps, frame index, device identity — carries over from this frame unchanged, so the result records exactly like the frame it came from. Works on any frame, live from a client or replayed from a recording. This frame is not modified.

matrix is an (N, 12) float64 array in the points() layout. The default (spherical) geometry is stored bit-exactly — an unedited points() matrix reproduces the frame's points losslessly.

Pass the same cartesian you pass to points(). Cartesian geometry converts back to the stored spherical form (~1e-5), and a basis mismatch is not detectable: a cartesian=True matrix passed without the flag stores x, y, z into the range and angle fields. Invalid returns may have no meaningful x, y, z either — one dropped before a range was measured loses its angles in a cartesian edit, so edit in the default basis if you are keeping those. A cartesian row that names no position — all-zero, or carrying NaN or inf, as a ROS cloud writes an absent return — stores as range 0 rather than propagating.

Float columns store as float32, and a value beyond float32's range saturates to ±inf — except the geometry columns under cartesian=True, which then name no position and store as range 0 by the rule above; the integer-backed columns (timestamp_nanosecs, the indices, drop_reason, combine_method, user_data) must hold exact in-range integers, else ValueError.

From a pandas DataFrame, select columns by name so column order can't drift and a basis mismatch raises KeyError rather than storing wrong coordinates: frame.with_points(df[frame.points_columns()].to_numpy()).

def synthetic( matrix: NDArray[numpy.float64], source: ProductId | None = None, timestamp_seconds: int | None = None, timestamp_nanoseconds: int | None = None, frame_index: int | None = None, cartesian: bool | None = None) -> VoyantFrame:

Build a frame of generated data — an Isaac Sim scene, a scripted test cloud — that records and replays like a real log.

source names the generator (e.g. ProductId.IsaacSim vs ProductId.SoftwareSimulator; None for unattributed) and becomes the frame's device_id: the honest marker for synthetic data. Real sensor ids raise ValueError, as does a timestamp outside the epoch (seconds >= 0, nanoseconds 0..=999_999_999). Give each frame of a recorded sequence its own timestamp_seconds / timestamp_nanoseconds / frame_index — the frame records as an ordinary state-carrying entry, so the replayed log keeps the scene's timeline, identity, and any with_sdl-authored configuration; unauthored state fields hold defaults, not measurements (a zeroed FPGA temperature reads -273.15 °C).

matrix is an (N, 12) float64 array in the points() layout; the column and basis rules are with_points's.

def stateless( matrix: NDArray[numpy.float64], source: ProductId | None = None, serial_number: int | None = None, timestamp_seconds: int | None = None, timestamp_nanoseconds: int | None = None, frame_index: int | None = None, cartesian: bool | None = None) -> VoyantFrame:

Build a frame from points that arrived without the sensor's heartbeat — a ROS bag, a CSV — so it declares no state.

carries_state is False and sensor_state()/host_state() return None, so a reader sees "none was recorded" rather than defaults that look like readings. Otherwise this is synthetic: timestamp_seconds / timestamp_nanoseconds / frame_index record and replay the same way, so give each frame of a sequence its own. source and serial_number may name a real sensor, which synthetic must not — the frame claims no measurements, so naming the sensor the points came from is provenance, not a claim about state. An out-of-epoch timestamp clamps instead of raising.

matrix is an (N, 12) float64 array in the points() layout; the column and basis rules are with_points's.

def with_sdl(self, sdl: SdlCommand) -> VoyantFrame:

A new frame whose SDL state reads as if the sensor had applied sdl (status Applied); everything else — points, identity, state provenance — carries over unchanged, like with_points. A synthetic scene declares the configuration its points were generated under; a recorded frame is equally editable, as its points already are. sdl is not put through send_sdl's send-time checks, so a scene may declare a configuration a real sensor would refuse. Raises ValueError on a frame with no state (carries_state False), or if sdl.req_state is not one the host can request — a fresh SdlCommand() defaults to SdlState.Unknown.

On a frame whose identity is a simulator, the measured chirp bandwidth is also set to what sdl commands: simulation has no measurement to differ from the command, and the field otherwise stays at its power-up 0, which the DSP reads as an invalid bandwidth and drops every point on. A real sensor's measurement is left alone — as is that of a frame built without a source, whose ProductId.Unknown is indistinguishable from an unrecognised real sensor. Name a simulator source on any scene you intend to synthesise peaks for.

def valid_mask(self) -> NDArray[numpy.bool]:

Boolean mask of valid returns, aligned with points() rows — e.g. frame.points()[frame.valid_mask()].

def sensor_state(self) -> SensorState | None:

Sensor state snapshotted at this frame's heartbeat, in physical units, or None when the frame carries none (see carries_state). Derived from the raw snapshot on each call — hoist out of loops.

def state_payload(self) -> bytes | None:

This frame's sensor state as the bytes a log entry stores, or None when the frame carries none (see carries_state). The inverse of decode_sensor_state, so a snapshot moves between logs without a trip through physical units — including out of an authored synthetic frame and into a data-stream log that needs a state entry.

Returns:

bytes | None: One stored state region, zero-padded as written.

def host_state(self) -> HostState | None:

Host-side context stamped when this frame was published (snapshot age, ramp drops, time sync), or None when the frame carries none.

def describe(self) -> str:

A one-screen summary of the frame and its state snapshots.

n_points: int
device_id: str

Formatted device ID, e.g. "CAR-30-005".

n_valid_points: int
timestamp: float

Frame-start time as fractional seconds since the Unix epoch. Convenient, but f64 quantizes a present-day epoch to a few hundred nanoseconds — use timestamp_seconds and timestamp_nanoseconds for the exact recorded value.

timestamp_seconds: int

Whole seconds of the frame-start time since the Unix epoch.

timestamp_nanoseconds: int

Sub-second remainder of the frame-start time, in nanoseconds (0..=999_999_999).

carries_state: bool

Whether this frame carries real sensor and host state. False when no state was recorded for it — most often a recording converted from the pre-v1.0.0 format, which stored none — in which case sensor_state() and host_state() return None rather than zeros that look like readings. Points are real either way; device identity and timestamps come from the recording, so only a frame built from scratch leaves them unset. Synthetic frames are state-carrying; the sim device_id is what marks them.

frame_index: int

Device frame counter.

class VoyantPlayback:

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

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

Create a new VoyantPlayback instance

Arguments:
  • rate: Playback rate (1.0 = real-time, 0.0 or None = as fast as possible, default: None). A rate too small to pace (below 0.001x), or one that is negative or non-finite, also plays as fast as possible and logs a warning.
  • loopback: Whether to loop when reaching the end of file (default: false)
  • keep_invalid_points: Whether frames keep their invalid points (default: false). Invalid points come from a capture made with the client's diagnostic mode enabled, or from frames constructed or edited by hand.
def open(self, file_path: str) -> None:

Open a file for playback

Arguments:
  • file_path: Path to the recording file
Raises:
  • IOError: If the file cannot be read, or is a pre-v1.0.0 recording — the message names the command that converts it. Any file already open is left open.
def reset(self) -> None:

Reset playback to the beginning of the file

rate: float | None

Get playback rate

api_version: str | None

Version of the API that wrote the open recording, e.g. "1.0.0" or "1.0.0-alpha". A converted recording reports the version of the API that produced the original, not the one that converted it. None if no file is open.

frame_timestamp: float

Get the current frame timestamp in seconds

loopback: bool

Get loopback setting

is_open: bool

Check if a file is currently open

frame_index: int

Get the current frame index

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, overwrite: bool | None = None)

Create a new VoyantRecorder instance

Arguments:
  • output_path: Base path for output recording file(s); must end in .vynt
  • 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)
  • overwrite: Allow overwriting an existing output file (default: False — fail instead)
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.

current_file_path: str

Get the path of the file being written, with timestamp and split naming resolved

Construction opens the first file, so this reads immediately — use it to name files that have to pair with the recording. After finalize() the last path stays readable.

is_active: bool

Check if the recorder is still active (not finalized)

frames_recorded: int

Get the total number of frames recorded across all files

After finalize() the final count stays readable.

split_count: int

Get the number of files created (split count)

After finalize() the final count stays readable.

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 or as a starting point for manual frame construction.

Returns:

VoyantFrame: An empty frame with no points. It declares no state, so carries_state is False and sensor_state() / host_state() return None — the frame has no readings to report, rather than zeroed ones.

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]): Frame timestamp, whole seconds. 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 exactly this range.
  • 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:

VoyantFrame: A complete frame with sensor state and point cloud data — the "SIM-SW" device identity marks it as synthetic. The frame will contain elevations * azimuths total points arranged in a grid pattern.

Raises:
  • ValueError: If elevations or azimuths exceeds 65535, the largest index the point fields can address, or if their product exceeds the points one recorded frame can hold.
Notes:
  • Point timestamps increment by 16384 nanoseconds every 8 points to simulate the sensor's 8-channel parallel acquisition pattern
  • The frame reports device ID "SIM-SW" to indicate synthetic data
  • Points are generated in azimuth-fastest scan order (a full azimuth sweep per elevation line), starting at the top-left: index (0, 0) is the topmost elevation line and the leftmost azimuth column
  • When drop_center is True, approximately 25% of the center FOV will have invalid points
def decode_sensor_state(payload: bytes) -> SensorState:

Decode a stored sensor-state payload into a snapshot in physical units.

Arguments:
  • payload (bytes): One stored state region. Trailing zero padding is ignored, so both the padded region and the bare snapshot decode.
Returns:

SensorState: The decoded snapshot — the same type and units as VoyantFrame.sensor_state().

Raises:
  • ValueError: The payload is shorter than one snapshot.
def api_version() -> str:

Version of the voyant-api library backing this package, e.g. "1.0.0" or "1.0.0-dev".

Compiled in from the library's own version, so it reports what is actually installed. VoyantPlayback.api_version is a different fact — the version that wrote an open recording.

def interface_contract_version() -> str:

Version of the FPGA interface contract this package was built against, e.g. "1.5.6".

The contract pins the sensor wire protocol; [api_version] is the library's own version. Report both when describing a build.

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.