client

Client-based API access.

Client

Client(url=None)

Client to fetch or publish timeseries.

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_url(url)
    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)

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

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,
) -> 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.

    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]

    descriptor = create_descriptor(
        RequestType.Count,
        pattern=pattern,
        data_type=data_type,
        min_rate=min_rate,
        max_rate=max_rate,
        publisher=publisher,
        validator=self._validator,
    )
    count = 0
    with connect(self.initial_url) as client:
        flight_info = get_flight_info(client, descriptor)
        with MultiEndpointStream(flight_info.endpoints, client) as stream:
            for data in stream.unpack():
                count += data["count"]
    return count

describe

describe(channels)

Get channel metadata for channels requested

Parameters:

Name	Type	Description	Default
channels	list[str]	List of channels to request.	required

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]) -> dict[str, Channel]:
    """Get channel metadata for channels requested

    Parameters
    ----------
    channels : list[str]
        List of channels to request.

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

    """
    descriptor = create_descriptor(
        RequestType.Describe, channels=channels, validator=self._validator
    )
    with connect(self.initial_url) as client:
        return {
            channel.name: channel
            for channel in self._stream_channel_metadata(client, descriptor)
        }

fetch

fetch(channels, start, end)

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

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,
) -> 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.

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

    """
    return concatenate_blocks(*self.stream(channels, start, end))

find

find(pattern=constants.DEFAULT_MATCH, data_type=None, min_rate=constants.MIN_SAMPLE_RATE, max_rate=constants.MAX_SAMPLE_RATE, publisher=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

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,
) -> 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.

    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]

    descriptor = create_descriptor(
        RequestType.Find,
        pattern=pattern,
        data_type=data_type,
        min_rate=min_rate,
        max_rate=max_rate,
        publisher=publisher,
        validator=self._validator,
    )
    with connect(self.initial_url) as client:
        yield from self._stream_channel_metadata(client, descriptor)

stream

stream(channels, start=None, end=None)

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

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,
) -> 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.

    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
    metadata: dict[str, Channel] = {}
    schemas: dict[str, pyarrow.Schema] = {}

    with connect(self.initial_url) as client:
        descriptor = create_descriptor(
            RequestType.Stream,
            channels=channels,
            start=start_ns,
            end=end_ns,
            validator=self._validator,
        )
        flight_info = get_flight_info(client, descriptor)
        # use the serialized endpoints as the mux keys
        keys = [e.serialize() for e in flight_info.endpoints]
        mux: Muxer = Muxer(keys=keys)
        with MultiEndpointStream(flight_info.endpoints, client) as stream:
            for chunk, endpoint in stream:
                time = chunk.data.column("time").to_numpy()[0]
                mux.push(time, endpoint.serialize(), chunk.data)
                # FIXME: how do we handle stream drop-outs that result
                # in timeouts in the muxer that result in null data in
                # the mux pull?
                for mux_data in mux.pull():
                    blocks = []
                    # update channel metadata if needed
                    for key, batch in mux_data.items():
                        if (
                            key not in schemas
                            or schemas[key].metadata != batch.schema.metadata
                        ):
                            channel_fields: list[pyarrow.field] = list(
                                batch.schema
                            )[1:]
                            for field in channel_fields:
                                metadata[field.name] = Channel.from_field(field)
                            schemas[key] = batch.schema

                        blocks.append(
                            SeriesBlock.from_column_batch(batch, metadata)
                        )

                    # generate synchronized blocks
                    yield combine_blocks(*blocks)