Usage¶

sgn-arrakis provides two SGN pipeline elements for working with Arrakis data streams: a source for reading data, and a sink for writing it.

Both elements integrate with SGN pipelines via the sgn-ts timeseries framework.

ArrakisSource¶

ArrakisSource streams channel data from an Arrakis server into an SGN pipeline. Source pads are named after the channels they stream.

Key parameters¶

channels -- list of channel names to stream (e.g., ["L1:GDS-CALIB_STRAIN"]); source pads are generated per channel
start -- GPS start time in seconds, or current time if None
end or duration -- GPS end time or total duration in seconds; omit both for an endless stream
replay_id or replay_start/replay_end -- stream from a server-registered replay, or from an ad-hoc replay window
in_queue_timeout -- seconds to wait for a block from the server before timing out (default: 60)

Stream from the current time¶

When start is not specified, ArrakisSource begins streaming from the current GPS time:

from sgn_arrakis import ArrakisSource
from sgn import Pipeline, SignalEOS
from sgnts.sinks import NullSeriesSink

channels = ["L1:GDS-CALIB_STRAIN"]

src = ArrakisSource(
    channels=channels,
)
sink = NullSeriesSink(
    sink_pad_names=channels,
    verbose=True,
)

pipeline = Pipeline()
pipeline.connect(src, sink)
with SignalEOS():
    pipeline.run()

Stream a fixed duration¶

Set start and duration (or end) to stream a bounded segment:

src = ArrakisSource(
    channels=["L1:GDS-CALIB_STRAIN"],
    start=1187008882,
    duration=64,
)

Gap handling¶

When Arrakis delivers blocks containing null data, ArrakisSource creates gap buffers with no data payload. Downstream elements can detect these via the buffer's is_gap property.

Resilience¶

ArrakisSource is resilient to server interruptions. On stream errors it reconnects with capped exponential backoff, failing the pipeline once the reconnect budget is spent (roughly 3 minutes of retries by default); bounded streams resume from the last delivered timestamp, and live streams bridge outages with gap buffers so downstream elements keep flowing. The outage gap horizon trails wall clock by each channel's latency contract (max_latency plus stride) from the stream metadata.

The reconnect settings are inherited from sgn-ts's TSResourceSource:

reconnect -- set to False to fail the pipeline on stream errors instead
reconnect_backoff / reconnect_backoff_max -- initial and maximum backoff between attempts, in seconds (defaults: 1, 30)
max_reconnects -- consecutive failed attempts before giving up (default: 10, roughly 3 minutes at the default backoff; None retries forever)
reconnect_gap_lag -- override the outage gap horizon, in seconds

ArrakisSink¶

ArrakisSink publishes channel data from an SGN pipeline to Arrakis. Sink pads are named after the channels they publish.

Key parameters¶

publisher_id -- admin-assigned publisher ID (required)
channels -- list of channel names to publish; must match all channels registered to the publisher
replay_id -- optional replay namespace to publish derived channels under

Basic usage¶

from sgn_arrakis import ArrakisSink
from sgn import Pipeline, SignalEOS
from sgnts.sources import FakeSeriesSource

channels = ["H1:GDS-CALIB_STRAIN"]

source = FakeSeriesSource(
    source_pad_names=channels,
    signals={
        "H1:GDS-CALIB_STRAIN": {
            "signal-type": "const",
            "sample-shape": (1,),
            "sample-rate": 16384,
            "value": 0.0,
        },
    },
)
sink = ArrakisSink(
    publisher_id="my-publisher-id",
    channels=channels,
)

pipeline = Pipeline()
pipeline.connect(source, sink)
with SignalEOS():
    pipeline.run()

Channel validation¶

On startup, ArrakisSink registers with the Arrakis server and verifies that the configured channels exactly match those assigned to the publisher. A ValueError is raised if there is a mismatch.

Background publishing¶

ArrakisSink publishes data in a background thread so that the main pipeline execution is not blocked by network I/O. Data blocks are queued internally and published asynchronously.

CLI¶

sgn-arrakis includes a command-line tool for testing source and sink functionality.

Stream channels¶

Stream channel data from Arrakis and print to the console:

sgn-arrakis source L1:GDS-CALIB_STRAIN H1:GDS-CALIB_STRAIN

Publish test data¶

Publish constant-value test data to all channels registered to a publisher:

sgn-arrakis sink my-publisher-id

Use --include-gaps to include gap buffers in the published stream:

sgn-arrakis sink --include-gaps my-publisher-id

Options¶

--url / -u -- specify the Arrakis server URL
--version -- display the version
LOG_LEVEL environment variable -- set logging verbosity (default: DEBUG)