Crate streamer 

The streamer binary: streams LabJack T7 analog input samples into NATS JetStream.

The streamer is the first stage of the data path: streamer -> local NATS JetStream -> archiver -> Parquet, with the exporter serving the Parquet files as CSV over NATS. It reads its active configuration from a local KV bucket (avenabox by default), optionally mirrors that entry from a central NATS deployment, opens the LabJack over Ethernet, and publishes one FlatBuffer Scan message per channel per LJM read. Structured configurations publish to <root>.<site>.<box>.<source>.live.chNN (for example avenars.<site>.<box>.<source>.live.chNN); legacy configurations publish to <root>.<asset>.data.chNN (see the subjects module).

A configuration change in KV that alters SampleConfig (channels, rates, identity and subject fields, rotate_secs or labjack_on_off) stops the active LabJack stream and restarts it with the new settings. Calibrations, filters and label fields are not part of it, so changing them leaves the stream running. Setting labjack_on_off to false stops sampling until the configuration changes again.

Configuration

• NATS_SERVERS - Comma-separated local NATS server URLs. Default: nats://127.0.0.1:4222.
• NATS_CREDS_FILE - Credentials file for the local NATS connection. Default: apt.creds.
• JS_DOMAIN - JetStream domain for the local connection. Unset or empty uses the default domain.
• CFG_BUCKET - Local KV bucket that holds the streamer configuration. Also the fallback for CENTRAL_CFG_BUCKET. Default: avenabox.
• CFG_KEY - Key of the configuration entry in that bucket. Also the fallback for CENTRAL_CFG_KEY. Default: unknown-site.macbook.unknown-source.config.
• CENTRAL_NATS_SERVERS - Comma-separated central NATS server URLs. When neither this nor CFG_NATS_SERVERS is set, central mirroring is off.
• CFG_NATS_SERVERS - Fallback for CENTRAL_NATS_SERVERS.
• CENTRAL_NATS_CREDS_FILE - Credentials file for the central connection. Default: the value of NATS_CREDS_FILE, or apt.creds.
• CENTRAL_CFG_BUCKET - Central KV bucket to mirror from. Default: CFG_BUCKET, or avenabox.
• CENTRAL_CFG_KEY - Central KV key to mirror from. Default: CFG_KEY, or unknown-site.macbook.unknown-source.config.
• CENTRAL_JS_DOMAIN - JetStream domain on the central server. Falls back to CFG_JS_DOMAIN; unset uses the default domain.
• STREAM_MAX_BYTES - Byte limit for the JetStream stream, greater than zero. Unset means unlimited (-1).
• STREAMER_MAX_LABJACK_FAILURES - Consecutive sampler failures before the process exits. Values that do not parse or are zero use the default. Default: 5.
• STREAMER_LABJACK_RETRY_DELAY_SECS - Delay before restarting the sampler after a failure, in seconds. Default: 5.
• LABJACK_IP - IPv4 address of the LabJack T7. Required unless LABJACK_IDENTIFIER holds an IPv4 address.
• LABJACK_IDENTIFIER - Fallback for LABJACK_IP, used only when it parses as an IPv4 address.
• LABJACK_SERIAL - Expected serial number. When set, a device with another serial is rejected. ANY or a non-numeric value disables the check.
• LABJACK_NAME - Logical device name. Only logged.
• LJM_PATH - Path to the LJM shared library (dynlink builds only). When unset, no path is passed to ljmrs.

Design

• Timestamps are derived, not read. The LJM stream API returns values without timestamps, so StreamClock numbers the batches and computes each batch’s first-sample time from the actual scan rate. It slews the timeline toward the system clock by at most 1 ms a minute, so drift between the LabJack crystal and the host clock does not grow over runs that last weeks, without chasing read-latency noise (see StreamClock).
• Publishing does not wait for JetStream acknowledgements. The acks for a batch are checked on a spawned task so a slow ack does not delay the next LabJack read, which would risk a device buffer overflow.
• Scans that LJM fills with -9999.0 after a buffer overflow are published as NaN so they are not archived as real readings.
• After STREAMER_MAX_LABJACK_FAILURES consecutive failures the process exits with success, so a systemd unit that restarts only on failure leaves it down instead of looping on a missing device.
• Central configuration is parsed before it is written to local KV, so an invalid central value cannot replace the last working local one.

Modules

labjack 🔒

LabJack T7 connection and handle-inspection helpers for the streamer binary.

ljm_mode 🔒

LabJack LJM library initialization for the selected linking mode.

nats_config 🔒

Shared NATS server-list and JetStream context helpers.

sample_data_generated 🔒

subjects 🔒

NATS subject builders for LabJack live data and archive exports.

Structs

CentralKvSyncConfig 🔒

Connection and key details for mirroring configuration from central NATS.

LabJackGuard 🔒

RAII guard that stops streaming and closes the LabJack handle on drop.

NestedConfig 🔒

Raw top-level configuration document stored in NATS KV.

SampleConfig 🔒

Normalized streamer configuration used by the sampling loop.

SensorSettings 🔒

Raw sensor settings section from the dashboard configuration document.

StreamClock 🔒

Tracks timestamp continuity for successive LabJack stream batches.

Enums

ClockAdjustment 🔒

What the clock decided to do at the end of one drift window.

Constants

CLOCK_CHECK_WINDOW_NS 🔒

Length of each window over which clock drift is measured, in nanoseconds (60 s).

CLOCK_LOG_EVERY_WINDOWS 🔒

Windows between summary log lines about slewing (60, about one hour).

CLOCK_MAX_SLEW_NS 🔒

Largest adjustment made in one window when slewing, in nanoseconds (1 ms).

CLOCK_OFFSET_HISTORY_WINDOWS 🔒

Windows over which the slewing offset is taken as the smallest window offset (15, about a quarter of an hour).

CLOCK_SLEW_THRESHOLD_NS 🔒

Smallest window offset that is acted on, in nanoseconds (5 ms).

CLOCK_STEP_THRESHOLD_NS 🔒

Offset treated as a system clock step rather than drift, in nanoseconds (2 s).

CLOCK_STEP_WINDOWS 🔒

Consecutive windows a step-sized offset must persist before the timeline jumps.

LJM_DUMMY_VALUE 🔒

Value LJM inserts for scans lost to a stream buffer overflow when it auto-recovers (LJM’s LJM_DUMMY_VALUE constant).

Functions

central_kv_sync_config_from_env 🔒

Builds central-to-local KV mirroring settings from environment variables.

connect_nats_with_creds 🔒

Connects to NATS using a credentials file and explicit server list.

decide_adjustment 🔒

Decides how to adjust the timeline from two consecutive window offsets.

deinterleave_channel 🔒

Copies one channel’s samples out of an interleaved LJM batch into out.

derive_sample_interval_ns 🔒

Converts the actual LabJack scan rate into a sample interval in nanoseconds.

ensure_kv_bucket 🔒

Opens a KV bucket, creating it if it does not exist.

ensure_stream_exists 🔒

Creates or reconciles the JetStream stream used for live LabJack samples.

env_nonempty 🔒

Returns a trimmed environment variable value when it is set and non-empty.

labjack_retry_delay_from_env 🔒

Reads the delay between sampler restarts after a failed run.

load_config_from_kv 🔒

Loads and parses the initial streamer configuration from local NATS KV.

local_creds_path_from_env 🔒

Returns the credentials file used for local NATS connections.

main 🔒

Starts the streamer service, configuration watchers, and sampler supervisor.

max_labjack_failures_from_env 🔒

Reads the number of consecutive sampler failures after which the streamer exits.

mirror_central_kv_once 🔒

Performs one central-to-local KV mirror pass during streamer bootstrap.

mirror_remote_kv_entry_to_local 🔒

Copies one remote KV configuration entry into the local KV bucket if needed.

replace_dummy_samples 🔒

Replaces LJM auto-recovery placeholders with NaN and returns how many.

run_central_kv_sync 🔒

Watches central NATS KV and mirrors valid updates into the local bucket.

run_sampler 🔒

Supervises LabJack sampling across config changes and transient failures.

sample_config_from_json 🔒

Parses a NATS KV JSON value into a runtime sampling config.

sample_config_from_nested 🔒

Converts the dashboard JSON shape into the runtime sampling configuration.

sample_with_config 🔒

Runs one LabJack streaming session with a fixed configuration.

stream_max_bytes_from_env 🔒

Reads the optional JetStream stream byte limit from STREAM_MAX_BYTES.

unix_time_now_ns 🔒

Returns the current Unix timestamp in nanoseconds.

watch_kv_config 🔒

Watches local NATS KV for streamer configuration updates.