Streaming Data

Use arrakis.api.stream to receive timeseries data as a sequence of arrakis.block.SeriesBlock objects. This is useful for processing data incrementally or working with live data.

Live Streaming

Omit start and end times to stream live data starting from the current time:

import arrakis

channels = [
    "H1:CAL-DELTAL_EXTERNAL_DQ",
    "H1:LSC-POP_A_LF_OUT_DQ",
]

for block in arrakis.stream(channels):
    print(f"t={block.time:.1f}  duration={block.duration}s")
    for channel, series in block.items():
        print(f"  {channel}: {len(series)} samples")

Live streaming runs indefinitely. Break out of the loop or use Ctrl+C to stop.

Historical Streaming

Provide start and end GPS times to stream historical data:

for block in arrakis.stream(channels, start=1187000000, end=1187001000):
    print(f"t={block.time:.1f}  duration={block.duration}s")

The stream yields blocks sequentially, each covering a fixed time stride. By default the stride is determined by the channel configuration on the server. Historical streams terminate automatically when the end time is reached.

Requesting Larger Blocks

Request larger blocks with the stride parameter, in seconds. It must be an integer multiple of the stream's native stride (for multiple channels, the least common multiple of the channels' strides):

# receive data in 1-second blocks
for block in arrakis.stream(channels, start=1187000000, end=1187001000, stride=1):
    print(f"t={block.time:.1f}  duration={block.duration}s")

Aggregation happens client-side: native blocks are concatenated until the requested stride is covered. A few consequences:

• For live streams, a block is only yielded once its full span has been received, so delivery latency grows with the stride. Live blocks align to absolute GPS multiples of the stride.
• For bounded requests, blocks align to start; if the requested span is not a multiple of the stride, the final block is shorter.
• Gaps within a block appear as masked spans in the data arrays (see Gap Handling); with on_gap='skip', a block is skipped only if its entire span has no data.

Replay Streaming

Stream data from a replay window by providing replay parameters. The server retimestamps archival data to the present and loops cyclically through the window.

Use replay_id to stream from a named replay. This also gives access to derived channels published under that replay context (see Finding Channels):

# stream using a registered replay ID
for block in arrakis.stream(channels, replay_id="O4a"):
    print(f"t={block.time:.1f}  duration={block.duration}s")

For opportunistic replays without a named ID, use explicit window bounds:

# stream using explicit replay window bounds (GPS seconds)
for block in arrakis.stream(
    channels,
    replay_start=1369224018,
    replay_end=1385193618,
):
    print(f"t={block.time:.1f}  duration={block.duration}s")

Replay streaming behaves like live streaming -- data is retimestamped to "now" and loops indefinitely until you break out of the loop. Combine with start and end to bound the replayed range.

How Streaming Works

When you call stream(), arrakis:

1. Queries the server for channel metadata (stride, latency, partitioning).
2. Opens one or more data streams via Arrow Flight or Kafka.
3. Feeds incoming blocks through a multiplexer (arrakis.mux.BlockMuxStream) that synchronizes data across streams and handles gaps.
4. Yields complete, time-aligned SeriesBlock objects.

Gap Handling

If data for a channel is missing or arrives late, the multiplexer inserts gap blocks -- blocks where the data arrays are NumPy masked arrays with all values masked. You can detect gaps on a per-series basis:

for block in arrakis.stream(channels):
    for channel, series in block.items():
        if series.has_gaps:
            print(f"  {channel}: gap detected")
        else:
            print(f"  {channel}: {len(series)} samples")

The timeout before a gap is inserted is determined by the channel's max_latency attribute. For live streams, the multiplexer continuously checks for timeouts; for historical streams, gaps indicate that the data was not available on the server.

Stream vs Fetch

	fetch()	stream()
Returns	Single SeriesBlock	Generator of SeriesBlock
Memory	Loads full range	One block at a time
Live data	No	Yes (omit start/end)
Use case	Analysis of bounded intervals	Continuous processing, large ranges

See Fetching Data for the batch alternative.