arrakis-python 0.1.0

Arrakis Python client library

arrakis-python

Arrakis Python client library

---

Resources

• Documentation
• Source Code
• Issue Tracker

Installation

pip install git+https://git.ligo.org/ngdd/arrakis-python.git

Features

• Query live and historical timeseries data
• Describe channel metadata
• Search for channels matching a set of conditions
• Publish timeseries data

Quickstart

Fetch timeseries

import arrakis

start = 1187000000
end = 1187001000
channels = [
    "H1:CAL-DELTAL_EXTERNAL_DQ",
    "H1:LSC-POP_A_LF_OUT_DQ",
]

block = arrakis.fetch(channels, start, end)
for channel, series in block.items():
    print(channel, series)

where block is a [arrakis.block.SeriesBlock][] and series is a [arrakis.block.Series][].

Stream timeseries

1. Live data

import arrakis

channels = [
    "H1:CAL-DELTAL_EXTERNAL_DQ",
    "H1:LSC-POP_A_LF_OUT_DQ",
]

for block in arrakis.stream(channels):
	print(block)

2. Historical data

import arrakis

start = 1187000000
end = 1187001000
channels = [
    "H1:CAL-DELTAL_EXTERNAL_DQ",
    "H1:LSC-POP_A_LF_OUT_DQ",
]

for block in arrakis.stream(channels, start, end):
    print(block)

Describe metadata

import arrakis

channels = [
    "H1:CAL-DELTAL_EXTERNAL_DQ",
    "H1:LSC-POP_A_LF_OUT_DQ",
]

metadata = arrakis.describe(channels)

where metadata is a dictionary mapping channel names to [arrakis.channel.Channel][].

Find channels

import arrakis

for channel in arrakis.find("H1:LSC-*"):
    print(channel)

where channel is a [arrakis.channel.Channel][].

Count channels

import arrakis

count = arrakis.count("H1:LSC-*")

Publish timeseries

from arrakis import Channel, Publisher, SeriesBlock, Time
import numpy

# define channel metadata
metadata = {
    "H1:FKE-TEST_CHANNEL1": Channel.from_name(
        "H1:FKE-TEST_CHANNEL1",
        data_type=numpy.float64,
        sample_rate=64,
    ),
    "H1:FKE-TEST_CHANNEL2": Channel.from_name(
        "H1:FKE-TEST_CHANNEL2",
        data_type=numpy.int32,
        sample_rate=32,
    ),
}

with Publisher() as publisher:
    # initial registration
    producer_id = "my_producer"  # admin-assigned ID
    publisher.register(producer_id, source="H1")

    # create block to publish
    series = {
        "H1:FKE-TEST_CHANNEL1": numpy.array([0.1, 0.2, 0.3, 0.4], dtype=numpy.float64),
        "H1:FKE-TEST_CHANNEL2": numpy.array([1, 2], dtype=numpy.int32),
    }
    block = SeriesBlock(
        1234567890 * Time.SECONDS,  # time in nanoseconds for first sample
        series,                     # the data to publish
        metadata,                   # the channel metadata
    )

    # publish timeseries
    publisher.publish(block)