client

Client-based API access.

Client

Client(url=None)

Retrieve channel information and timeseries data from Arrakis.

Parameters:

Name	Type	Description	Default
url	str	The URL to connect to. If the URL is not set, connect to a default server or one set by ARRAKIS_SERVER.	None

Source code in arrakis/client.py

def __init__(self, url: str | None = None):
    self.initial_url = parse_arrakis_url(url).geturl()
    logger.debug("initial url: %s", self.initial_url)
    self._validator = RequestValidator()

count

count(pattern=constants.DEFAULT_MATCH, data_type=None, min_rate=constants.MIN_SAMPLE_RATE, max_rate=constants.MAX_SAMPLE_RATE, publisher=None, time=None, replay_id=None)

Count channels matching a set of conditions

Parameters:

Name	Type	Description	Default
pattern	str	Channel pattern to match channels with, using regular expressions.	DEFAULT_MATCH
data_type	dtype - like | list[dtype - like]	If set, find all channels with these data types.	None
min_rate	int	The minimum sampling rate for channels.	MIN_SAMPLE_RATE
max_rate	int	The maximum sampling rate for channels.	MAX_SAMPLE_RATE
publisher	str | list[str]	If set, find all channels associated with these publishers.	None
time	float	GPS time in seconds indicating when the metadata query is valid. If None, routes to the live backend (current state).	None
replay_id	str	Server-registered replay identifier.	None

Returns:

Type	Description
int	The number of channels matching query.

Source code in arrakis/client.py

def count(
    self,
    pattern: str = constants.DEFAULT_MATCH,
    data_type: DataTypeLike | None = None,
    min_rate: int | None = constants.MIN_SAMPLE_RATE,
    max_rate: int | None = constants.MAX_SAMPLE_RATE,
    publisher: str | list[str] | None = None,
    time: float | None = None,
    replay_id: str | None = None,
) -> int:
    """Count channels matching a set of conditions

    Parameters
    ----------
    pattern : str, optional
        Channel pattern to match channels with, using regular expressions.
    data_type : numpy.dtype-like | list[numpy.dtype-like], optional
        If set, find all channels with these data types.
    min_rate : int, optional
        The minimum sampling rate for channels.
    max_rate : int, optional
        The maximum sampling rate for channels.
    publisher : str | list[str], optional
        If set, find all channels associated with these publishers.
    time : float, optional
        GPS time in seconds indicating when the metadata query is valid.
        If None, routes to the live backend (current state).
    replay_id : str, optional
        Server-registered replay identifier.

    Returns
    -------
    int
        The number of channels matching query.

    """
    data_type = _parse_data_types(data_type)
    if min_rate is None:
        min_rate = constants.MIN_SAMPLE_RATE
    if max_rate is None:
        max_rate = constants.MAX_SAMPLE_RATE
    if publisher is None:
        publisher = []
    elif isinstance(publisher, str):
        publisher = [publisher]

    time_ns = time_as_ns(time) if time is not None else None
    descriptor = create_descriptor(
        RequestType.Count,
        pattern=pattern,
        data_type=data_type,
        min_rate=min_rate,
        max_rate=max_rate,
        publisher=publisher,
        time=time_ns,
        replay_id=replay_id,
        validator=self._validator,
    )
    count = 0
    with connect(self.initial_url) as client:
        flight_info = get_flight_info(client, descriptor)
        with MultiFlightReader(flight_info.endpoints, client) as stream:
            for data in stream.unpack():
                count += data["count"]
    return count

describe

describe(channels, time=None, replay_id=None)

Get channel metadata for channels requested

Parameters:

Name	Type	Description	Default
channels	list[str]	List of channels to request.	required
time	float	GPS time in seconds indicating when the metadata query is valid. If None, routes to the live backend (current state).	None
replay_id	str	Server-registered replay identifier.	None

Returns:

Type	Description
dict[str, Channel]	Mapping of channel names to channel metadata.

Source code in arrakis/client.py

def describe(
    self,
    channels: list[str],
    time: float | None = None,
    replay_id: str | None = None,
) -> dict[str, Channel]:
    """Get channel metadata for channels requested

    Parameters
    ----------
    channels : list[str]
        List of channels to request.
    time : float, optional
        GPS time in seconds indicating when the metadata query is valid.
        If None, routes to the live backend (current state).
    replay_id : str, optional
        Server-registered replay identifier.

    Returns
    -------
    dict[str, Channel]
        Mapping of channel names to channel metadata.

    """
    time_ns = time_as_ns(time) if time is not None else None
    with connect(self.initial_url) as client:
        return self._describe(
            client,
            channels,
            time_ns=time_ns,
            replay_id=replay_id,
        )

domains

domains()

Get the list of domains available on the server.

Returns:

Type	Description
list[str]	Sorted list of domain names.

Source code in arrakis/client.py

def domains(self) -> list[str]:
    """Get the list of domains available on the server.

    Returns
    -------
    list[str]
        Sorted list of domain names.

    """
    result = self.scope_map()
    domain_set = set()
    for endpoint_info in result["endpoints"].values():
        domain_set.update(endpoint_info["scopes"].keys())
    return sorted(domain_set)

fetch

fetch(channels, start, end, replay_start=None, replay_end=None, replay_id=None, on_gap=ONGAP_DEFAULT)

Fetch timeseries data

Parameters:

Name	Type	Description	Default
channels	list[str]	List of channels to request.	required
start	float	GPS start time, in seconds.	required
end	float	GPS end time, in seconds.	required
replay_start	float	GPS start time of the replay window, in seconds.	None
replay_end	float	GPS end time of the replay window, in seconds.	None
replay_id	str	Server-registered replay identifier. Mutually exclusive with replay_start/replay_end.	None
on_gap	str	Policy for spans with no data: 'fill' (default) returns masked arrays over the gaps (note the memory cost over large holes), 'raise' raises arrakis.mux.GapError. 'skip' is not supported: fetch returns one contiguous block.	ONGAP_DEFAULT

Returns:

Type	Description
SeriesBlock	Dictionary-like object containing all requested channel data.

Source code in arrakis/client.py

def fetch(
    self,
    channels: list[str],
    start: float,
    end: float,
    replay_start: float | None = None,
    replay_end: float | None = None,
    replay_id: str | None = None,
    on_gap: str = ONGAP_DEFAULT,
) -> SeriesBlock:
    """Fetch timeseries data

    Parameters
    ----------
    channels : list[str]
        List of channels to request.
    start : float
        GPS start time, in seconds.
    end : float
        GPS end time, in seconds.
    replay_start : float, optional
        GPS start time of the replay window, in seconds.
    replay_end : float, optional
        GPS end time of the replay window, in seconds.
    replay_id : str, optional
        Server-registered replay identifier. Mutually exclusive with
        replay_start/replay_end.
    on_gap : str, optional
        Policy for spans with no data: 'fill' (default) returns
        masked arrays over the gaps (note the memory cost over
        large holes), 'raise' raises ``arrakis.mux.GapError``.
        'skip' is not supported: fetch returns one contiguous
        block.

    Returns
    -------
    SeriesBlock
        Dictionary-like object containing all requested channel data.

    """
    if on_gap.upper() == "SKIP":
        msg = "fetch returns one contiguous block; on_gap='skip' is not supported"
        raise ValueError(msg)
    return concatenate_blocks(
        *self.stream(
            channels,
            start,
            end,
            replay_start=replay_start,
            replay_end=replay_end,
            replay_id=replay_id,
            on_gap=on_gap,
        )
    )

find

find(pattern=constants.DEFAULT_MATCH, data_type=None, min_rate=constants.MIN_SAMPLE_RATE, max_rate=constants.MAX_SAMPLE_RATE, publisher=None, time=None, replay_id=None)

Find channels matching a set of conditions

Parameters:

Name	Type	Description	Default
pattern	str	Channel pattern to match channels with, using regular expressions.	DEFAULT_MATCH
data_type	dtype - like | list[dtype - like]	If set, find all channels with these data types.	None
min_rate	int	Minimum sampling rate for channels.	MIN_SAMPLE_RATE
max_rate	int	Maximum sampling rate for channels.	MAX_SAMPLE_RATE
publisher	str | list[str]	If set, find all channels associated with these publishers.	None
time	float	GPS time in seconds indicating when the metadata query is valid. If None, routes to the live backend (current state).	None
replay_id	str	Server-registered replay identifier.	None

Yields:

Type	Description
Channel	Channel objects for all channels matching query.

Source code in arrakis/client.py

def find(
    self,
    pattern: str = constants.DEFAULT_MATCH,
    data_type: DataTypeLike | None = None,
    min_rate: int | None = constants.MIN_SAMPLE_RATE,
    max_rate: int | None = constants.MAX_SAMPLE_RATE,
    publisher: str | list[str] | None = None,
    time: float | None = None,
    replay_id: str | None = None,
) -> Generator[Channel, None, None]:
    """Find channels matching a set of conditions

    Parameters
    ----------
    pattern : str, optional
        Channel pattern to match channels with, using regular expressions.
    data_type : numpy.dtype-like | list[numpy.dtype-like], optional
        If set, find all channels with these data types.
    min_rate : int, optional
        Minimum sampling rate for channels.
    max_rate : int, optional
        Maximum sampling rate for channels.
    publisher : str | list[str], optional
        If set, find all channels associated with these publishers.
    time : float, optional
        GPS time in seconds indicating when the metadata query is valid.
        If None, routes to the live backend (current state).
    replay_id : str, optional
        Server-registered replay identifier.

    Yields
    -------
    Channel
        Channel objects for all channels matching query.

    """
    data_type = _parse_data_types(data_type)
    if min_rate is None:
        min_rate = constants.MIN_SAMPLE_RATE
    if max_rate is None:
        max_rate = constants.MAX_SAMPLE_RATE
    if publisher is None:
        publisher = []
    elif isinstance(publisher, str):
        publisher = [publisher]

    time_ns = time_as_ns(time) if time is not None else None
    descriptor = create_descriptor(
        RequestType.Find,
        pattern=pattern,
        data_type=data_type,
        min_rate=min_rate,
        max_rate=max_rate,
        publisher=publisher,
        time=time_ns,
        replay_id=replay_id,
        validator=self._validator,
    )
    with connect(self.initial_url) as client:
        yield from self._stream_channel_metadata(client, descriptor)

replays

replays()

Get available replay windows with start/end times.

Returns:

Type	Description
dict	Mapping of replay IDs to their start/end GPS times in seconds.

Source code in arrakis/client.py

def replays(self) -> dict:
    """Get available replay windows with start/end times.

    Returns
    -------
    dict
        Mapping of replay IDs to their start/end GPS times in seconds.

    """
    return self._do_action("replays")

scope_map

scope_map()

Get the mapping of endpoints to their scopes.

Returns:

Type	Description
dict	Mapping of endpoint information including scopes per domain.

Source code in arrakis/client.py

def scope_map(self) -> dict:
    """Get the mapping of endpoints to their scopes.

    Returns
    -------
    dict
        Mapping of endpoint information including scopes per domain.

    """
    return self._do_action("scope-map")

server_info

server_info()

Get server version and capability metadata.

Returns:

Type	Description
dict	Server metadata including version info, backend type, capabilities, and domains.

Source code in arrakis/client.py

def server_info(self) -> dict:
    """Get server version and capability metadata.

    Returns
    -------
    dict
        Server metadata including version info, backend type,
        capabilities, and domains.

    """
    return self._do_action("server-info")

stream

stream(channels, start=None, end=None, kafka_url=None, replay_start=None, replay_end=None, replay_id=None, on_gap=ONGAP_DEFAULT)

Stream live or offline timeseries data

Parameters:

Name	Type	Description	Default
channels	list[str]	List of channels to request.	required
start	float	GPS start time, in seconds.	None
end	float	GPS end time, in seconds.	None
replay_start	float	GPS start time of the replay window, in seconds.	None
replay_end	float	GPS end time of the replay window, in seconds.	None
replay_id	str	Server-registered replay identifier. Mutually exclusive with replay_start/replay_end.	None
on_gap	str	Policy for spans with no data: 'fill' (default) yields masked gap blocks so the stream is continuous, 'skip' yields nothing for such spans (sparse blocks), 'raise' raises arrakis.mux.GapError.	ONGAP_DEFAULT

Yields:

Type	Description
SeriesBlock	Dictionary-like object containing all requested channel data.
Setting neither start nor end begins a live stream starting
from now.

Source code in arrakis/client.py

def stream(
    self,
    channels: list[str],
    start: float | None = None,
    end: float | None = None,
    kafka_url: str | None = None,
    replay_start: float | None = None,
    replay_end: float | None = None,
    replay_id: str | None = None,
    on_gap: str = ONGAP_DEFAULT,
) -> Generator[SeriesBlock, None, None]:
    """Stream live or offline timeseries data

    Parameters
    ----------
    channels : list[str]
        List of channels to request.
    start : float, optional
        GPS start time, in seconds.
    end : float, optional
        GPS end time, in seconds.
    replay_start : float, optional
        GPS start time of the replay window, in seconds.
    replay_end : float, optional
        GPS end time of the replay window, in seconds.
    replay_id : str, optional
        Server-registered replay identifier. Mutually exclusive with
        replay_start/replay_end.
    on_gap : str, optional
        Policy for spans with no data: 'fill' (default) yields
        masked gap blocks so the stream is continuous, 'skip'
        yields nothing for such spans (sparse blocks), 'raise'
        raises ``arrakis.mux.GapError``.

    Yields
    ------
    SeriesBlock
        Dictionary-like object containing all requested channel data.

    Setting neither start nor end begins a live stream starting
    from now.

    """
    start_ns = time_as_ns(start) if start is not None else None
    end_ns = time_as_ns(end) if end is not None else None
    replay_start_ns = time_as_ns(replay_start) if replay_start is not None else None
    replay_end_ns = time_as_ns(replay_end) if replay_end is not None else None
    metadata: dict[str, Channel] = {}
    reader: StreamReader

    # for opportunistic replay, translate start time into the
    # replay window so describe routes to the archival backend
    describe_time: int | None
    if replay_start_ns is not None and replay_end_ns is not None and not replay_id:
        describe_time = translate_time(start_ns, replay_start_ns, replay_end_ns)
    else:
        describe_time = start_ns

    with connect(self.initial_url) as client:
        # initial describe request to get the full metadata needed
        # to construct the appropriate muxer — use start time so the
        # info server routes to the correct backend for the request
        metadata = self._describe(
            client,
            channels,
            time_ns=describe_time,
            replay_id=replay_id,
        )
        for channel in metadata.values():
            assert channel.stride, "Channels do not include stride info."

        # get the flight info for streams
        descriptor = create_descriptor(
            RequestType.Stream,
            channels=channels,
            start=start_ns,
            end=end_ns,
            replay_start=replay_start_ns,
            replay_end=replay_end_ns,
            replay_id=replay_id,
            validator=self._validator,
        )
        flight_info = get_flight_info(client, descriptor)

    reader = _construct_stream_reader(
        flight_info.endpoints,
        metadata,
        start_ns,
        kafka_url=kafka_url,
        initial_url=self.initial_url,
    )

    # streams served through flight endpoints are relays of a
    # server-side muxer and need extra timeout headroom; direct
    # kafka reads consume the source and keep per-queue defaults
    mux_timeout = 0
    if kafka_url is None and any(
        not _endpoint_is_kafka(ep) for ep in flight_info.endpoints
    ):
        mux_timeout = _relay_mux_timeout(metadata)

    mux: BlockMuxStream = BlockMuxStream(
        reader, start=start_ns, timeout=mux_timeout, on_gap=on_gap
    )

    for block in mux.stream(end_ns):
        yield block

parse_arrakis_url

parse_arrakis_url(url=None)

Parse an Arrakis URL, filling in with default values

If url is None or unspecified, the ARRAKIS_SERVER environment variable will be used. If the port is not specified, 31206 will be used.

Source code in arrakis/client.py

def parse_arrakis_url(url: str | None = None) -> ParseResult:
    """Parse an Arrakis URL, filling in with default values

    If url is None or unspecified, the ARRAKIS_SERVER environment
    variable will be used.  If the port is not specified, 31206 will
    be used.

    """
    if url is None:
        url = os.getenv("ARRAKIS_SERVER", constants.DEFAULT_ARRAKIS_SERVER)
    assert url is not None, "ARRAKIS_SERVER not specified."
    if "://" not in url:
        url = "grpc://" + url
    parsed = urlparse(url)
    if ":" not in parsed.netloc:
        parsed = parsed._replace(netloc=parsed.netloc + ":31206")
    if parsed.netloc[0] == ":":
        parsed = parsed._replace(netloc="localhost" + parsed.netloc)
    return parsed