Stream data

A write stream sends values to a dataset as your program produces them, batching them in the background. Use it for live telemetry. For files, see Datasets.

For log messages, see Stream logs.

Connect the client

This guide assumes you have installed the client and setup your API keys.

from nominal.core import NominalClient

client = NominalClient.from_profile("default")

Stream values

get_write_stream() returns a stream. Leaving the with block sends what’s left and closes it.

import time
from datetime import datetime, timezone

from nominal.core import ChannelDataType

dataset = client.create_dataset("Motor test telemetry", prefix_tree_delimiter=".")
dataset.add_channel("motor.temperature", ChannelDataType.DOUBLE, unit="K")
dataset.add_channel("motor.mode", ChannelDataType.STRING)

start = datetime.now(timezone.utc)
with dataset.get_write_stream() as stream:
    for i in range(50):
        now = datetime.now(timezone.utc)
        for motor, baseline in [("left", 295.0), ("right", 297.0)]:
            stream.enqueue("motor.temperature", now, baseline + 0.1 * i, tags={"motor_id": motor})
            stream.enqueue("motor.mode", now, "idle" if i < 10 else "running", tags={"motor_id": motor})
        time.sleep(0.2)
end = datetime.now(timezone.utc)

Registering channels first sets their types. See Channels and data types. Timestamps can be timezone-aware datetimes, ISO 8601 strings or epoch nanoseconds. Use the time each value was measured, not the time you enqueue it.

To write several channels at one timestamp, use enqueue_from_dict. To write many points of one channel, use enqueue_batch. Arrays and structs have enqueue_float_array, enqueue_string_array and enqueue_struct.

Batching and delivery

The stream sends a batch when it reaches batch_size points, 250,000 by default, or after max_wait, 0.25 seconds by default. Smaller values lower latency. Larger values lower request overhead.

enqueue doesn’t confirm delivery, and there’s no file to poll. The two implementations report failures differently:

• Rust, the default where nominal-streaming is installed, can save failed batches to an .avro file with file_fallback for later upload. It also supports num_workers and track_metrics.

• Python logs failed uploads and doesn’t raise them when the stream closes.

Choose one with implementation="rust" or implementation="python". After streaming, check that the channels cover the expected time range.

Add tags to filter data in a run or asset

Tags tell apart points in the same channel, such as the two motors above. A run or asset can select one tag value with series_tags:

run = client.create_run("Left motor test", start=start, end=end)
run.add_dataset("motor", dataset, series_tags={"motor_id": "left"})

Use tags for a small set of values, such as component, vehicle or test configuration. Don’t tag with values that change every point, such as a counter or a UUID per sample, because each tag combination adds query work.

View streamed data

Open the dataset’s asset or run in a workbook and click Go Live to follow new points. For a full walkthrough, see the upload and stream tutorial.

Migrate from connections

Connection and StreamingConnection were removed. Stream to a dataset instead:

dataset = client.get_dataset("<DATASET_RID>")
with dataset.get_write_stream() as stream:
    stream.enqueue("temperature", "2026-01-01T00:00:00Z", 42.0)

Get a migrated connection’s dataset RID from Nominal. A connection RID passed to get_dataset doesn’t resolve. External datasets read from their provider and don’t accept streamed writes. nominal.experimental.stream_v2.create_write_stream() now takes a Dataset as its first argument.