Finding Channels¶

arrakis provides three functions for channel discovery: arrakis.api.find to search, arrakis.api.count to count matches, and arrakis.api.describe to get metadata for specific channels.

Find Channels by Pattern¶

arrakis.api.find yields arrakis.channel.Channel objects matching a regular expression pattern:

import arrakis

for channel in arrakis.find("H1:LSC-.*"):
    print(f"{channel.name}: {channel.sample_rate} Hz, {channel.data_type}")

The default pattern ".*" matches all channels.

Filter by Properties¶

Narrow results by data type, sample rate range, or publisher:

import numpy

# float64 channels between 256 and 4096 Hz
for channel in arrakis.find(
    "H1:.*",
    data_type=numpy.float64,
    min_rate=256,
    max_rate=4096,
):
    print(channel)

# channels from a specific publisher
for channel in arrakis.find(publisher="my_producer"):
    print(channel)

The data_type parameter accepts anything that numpy.dtype() understands: strings like "float64", NumPy types like numpy.float64, or numpy.dtype instances.

Historical Metadata Queries¶

By default, find(), count(), and describe() query the live server state. Pass a time parameter (GPS seconds) to query past metadata at a specific point in time:

# find channels that existed at GPS time 1187008882
for channel in arrakis.find("H1:.*", time=1187008882):
    print(channel)

# describe channels at a historical time
metadata = arrakis.describe(
    ["H1:CAL-DELTAL_EXTERNAL_DQ"],
    time=1187008882,
)

This routes the query to the appropriate backend for that time period, which is useful when channel configurations have changed over time.

Count Matches¶

arrakis.api.count returns the number of matching channels without iterating over them:

count = arrakis.count("H1:LSC-.*")
print(f"Found {count} channels")

It accepts the same filter parameters as find().

Describe Specific Channels¶

arrakis.api.describe retrieves metadata for channels you already know by name:

metadata = arrakis.describe([
    "H1:CAL-DELTAL_EXTERNAL_DQ",
    "H1:LSC-POP_A_LF_OUT_DQ",
])

for name, channel in metadata.items():
    print(f"{name}:")
    print(f"  sample_rate: {channel.sample_rate} Hz")
    print(f"  data_type:   {channel.data_type}")
    print(f"  publisher:   {channel.publisher}")
    print(f"  stride:      {channel.stride} ns")

The Channel Object¶

arrakis.channel.Channel is a frozen dataclass with the following fields:

Field	Type	Description
`name`	`str`	Full channel name (e.g. `H1:CAL-DELTAL_EXTERNAL_DQ`)
`data_type`	`str`	NumPy dtype name (e.g. `float64`, `int32`)
`sample_rate`	`float`	Sampling rate in Hz
`time`	`int` or `None`	Timestamp when metadata became active
`publisher`	`str` or `None`	Publisher ID
`partition_id`	`str` or `None`	Kafka partition ID
`partition_index`	`int` or `None`	Index within the partition
`stride`	`int` or `None`	Block duration in nanoseconds
`max_latency`	`int` or `None`	Max expected latency in seconds

Channels also expose parsed name components:

channel = metadata["H1:CAL-DELTAL_EXTERNAL_DQ"]
print(channel.domain)     # "H1"
print(channel.subsystem)  # "CAL"