source

Arrakis source element.

ArrakisSource dataclass

ArrakisSource(*, channels, replay_id=None, replay_start=None, replay_end=None, fast_forward=True, summary_interval=900.0)

Bases: TSResourceSource

Source element that streams channel data from Arrakis.

Streaming is resilient to server interruptions via the TSResourceSource resource hook API: on stream errors the worker reconnects with capped exponential backoff (see the reconnect settings on TSResourceSource), live streams bridge outages with gap buffers, and bounded streams resume from the last delivered timestamp. The outage gap horizon is derived per channel from the stream metadata's latency contract (max_latency + stride + a small grace).

Parameters:

Name	Type	Description	Default
channels	list[str]	Channels to stream from Arrakis. Source pads are automatically generated for each channel, with the channel name as the pad name.	required
start	float | None	Start time of stream in GPS seconds, or "now" if None.	required
duration	float | None	Duration of stream in seconds, or endless stream if None.	required
end	float | None	End time of stream in GPS seconds. Cannot be combined with duration.	required
replay_id	str or None	Server-registered replay identifier. Mutually exclusive with replay_start/replay_end.	None
replay_start	float or None	GPS start time of the replay window, in seconds. Mutually exclusive with replay_id.	None
replay_end	float or None	GPS end time of the replay window, in seconds. Mutually exclusive with replay_id.	None
fast_forward	bool	When True, a live stream read directly from Kafka that has fallen further behind real time than its latency contract allows -- for long enough that it clearly cannot catch up -- skips ahead to current data, surfacing the skipped span as an ordinary gap instead of degrading permanently. The skip and recovery are logged by the arrakis client. Ignored for bounded requests and server-relayed streams.	True
in_queue_timeout	int	How long to wait for a block from the Arrakis server before timing out with an error.	60
summary_interval	float	Seconds between INFO-level activity summaries reporting blocks received and how far the newest block lags real time; its steady presence doubles as a liveness heartbeat. 0 disables.	900

Notes

Logging: At INFO the source logs the first block received and the periodic activity summary; per-block detail is DEBUG. The stream request itself is logged by the arrakis client when it opens, and stream interruptions, reconnects, and recovery by the TSResourceSource reconnect machinery.

Thread safety: Marked thread_safe = True, so with Pipeline.run(threaded=...) this element's pad callbacks dispatch onto the thread pool. Everything this element adds runs through the TSResourceSource resource hook API, which executes on a bare worker-side instance in the worker thread -- never on graph threads (_channel_meta, _declared_channels, and the logging counters exist only on that worker-side instance). Per-pad safety of the graph-thread machinery (internal/new) is guaranteed and documented by TSResourceSource; marking matters because internal() can block waiting on the worker, and unmarked that wait would stall the pipeline's whole graph wave.

**Future editors MUST preserve thread safety**: keep new logic
inside the resource hooks. Overriding graph-thread methods
(``new``, ``internal``, ``set_data``, ``output_prototype``) or
adding element-side state they read requires re-auditing per
the TSResourceSource thread-safety contract.

static_source_pads property

static_source_pads

Define source pads from channels.

resource_gap_lag

resource_gap_lag(pad_name)

Outage-gap lag from the channel's latency contract.

Data for time T may legitimately arrive as late as T + max_latency (+ stride of headroom), so outage gaps are only emitted for spans older than that plus a small grace.

Source code in sgn_arrakis/source.py

def resource_gap_lag(self, pad_name: str) -> float | None:
    """Outage-gap lag from the channel's latency contract.

    Data for time T may legitimately arrive as late as
    T + max_latency (+ stride of headroom), so outage gaps are only
    emitted for spans older than that plus a small grace.
    """
    channel = self._channel_meta.get(pad_name)
    if channel is None or channel.max_latency is None:
        return None
    lag = channel.max_latency / Time.SECONDS + GAP_LAG_GRACE
    if channel.stride:
        lag += channel.stride / Time.SECONDS
    return lag

resource_open

resource_open(start, end, *, channels, replay_id=None, replay_start=None, replay_end=None, fast_forward=True)

Open the Arrakis stream (see TSResourceSource.resource_open).

Source code in sgn_arrakis/source.py

def resource_open(
    self,
    start: float | None,
    end: float | None,
    *,
    channels: list[str],
    replay_id: str | None = None,
    replay_start: float | None = None,
    replay_end: float | None = None,
    fast_forward: bool = True,
) -> Iterator[SeriesBlock]:
    """Open the Arrakis stream (see TSResourceSource.resource_open)."""
    return arrakis.stream(
        channels,
        start,
        end,
        replay_id=replay_id,
        replay_start=replay_start,
        replay_end=replay_end,
        fast_forward=fast_forward,
    )

resource_read

resource_read(stream, *, summary_interval=900.0)

Read one block, converting each channel's series to buffers.

Source code in sgn_arrakis/source.py

def resource_read(
    self,
    stream: Iterator[SeriesBlock],
    *,
    summary_interval: float = 900.0,
) -> dict[str, list[SeriesBuffer]]:
    """Read one block, converting each channel's series to buffers."""
    self._maybe_log_summary(summary_interval)

    block = next(stream)
    deliveries: dict[str, list[SeriesBuffer]] = {}
    block_end_ns: int | None = None
    for name, series in block.items():
        channel = series.channel
        series_end_ns = series.time_ns + series.duration_ns
        if block_end_ns is None or series_end_ns > block_end_ns:
            block_end_ns = series_end_ns

        # FIXME: should we do this for every block?
        assert channel.sample_rate in Offset.ALLOWED_RATES, (
            f"channel {name} has an invalid sample rate: {channel.sample_rate}"
        )
        # FIXME: should we do other checks?

        self._channel_meta[name] = channel
        buffers = []

        # Declare the channel's dtype ahead of its first buffers with
        # a zero-length data buffer (series.array keeps the dtype even
        # when the block is fully masked). If the first block carries
        # no unmasked samples, this declaration is the only thing that
        # stops the pad's spec from finalizing to the float64 default
        # and rejecting the real data when it arrives (see
        # TSResourceSource.output_prototype).
        if name not in self._declared_channels:
            buffers.append(
                SeriesBuffer(
                    offset=Offset.fromns(series.time_ns),
                    sample_rate=int(series.sample_rate),
                    data=series.array[:0],
                )
            )
            self._declared_channels.add(name)

        buffers.extend(series_to_buffers(series))
        deliveries[name] = buffers

    self._blocks_read += 1
    self._summary_blocks += 1
    if block_end_ns is not None:
        self._last_block_end_ns = block_end_ns

    # The first block proves data is flowing end to end; after
    # that, per-block lines are DEBUG detail
    if self._blocks_read == 1:
        self.logger.info(
            "Received first block, ending at GPS %s",
            _gps_str(block_end_ns),
        )
    else:
        self.logger.debug("Received block ending at GPS %s", _gps_str(block_end_ns))
    return deliveries

resource_setup

resource_setup()

Initialize worker-side channel bookkeeping.

Source code in sgn_arrakis/source.py

def resource_setup(self) -> None:
    """Initialize worker-side channel bookkeeping."""
    self._channel_meta: dict[str, Channel] = {}
    self._declared_channels: set[str] = set()

    # Logging state: first-block transition and the per-cadence
    # activity summary
    self._blocks_read: int = 0
    self._last_summary: float = time.monotonic()
    self._summary_blocks: int = 0
    self._last_block_end_ns: int | None = None

series_to_buffers

series_to_buffers(series)

Convert an arrakis Series to a list of SeriesBuffers.

If the Series data is masked, it is split into contiguous valid/gap runs, each returned as a separate SeriesBuffer. Unmasked data produces a single buffer.

Source code in sgn_arrakis/source.py

def series_to_buffers(series: Series) -> list[SeriesBuffer]:
    """Convert an arrakis Series to a list of SeriesBuffers.

    If the Series data is masked, it is split into contiguous valid/gap
    runs, each returned as a separate SeriesBuffer. Unmasked data
    produces a single buffer.
    """
    sample_rate = int(series.sample_rate)

    if not series.has_gaps:
        return [
            SeriesBuffer(
                offset=Offset.fromns(series.time_ns),
                sample_rate=sample_rate,
                data=series.array,
            )
        ]

    buffers = []
    offset = Offset.fromns(series.time_ns)
    masked_runs = [(s, True) for s in np.ma.clump_masked(series.data)]
    unmasked_runs = [(s, False) for s in np.ma.clump_unmasked(series.data)]
    for slc, is_gap in sorted(itertools.chain(masked_runs, unmasked_runs)):
        n = int(slc.stop - slc.start)
        buf = SeriesBuffer(
            offset=offset,
            sample_rate=sample_rate,
            data=None if is_gap else series.array[slc],
            shape=(n,),
        )
        buffers.append(buf)
        offset = buf.end_offset
    return buffers