XpectraFlow docsPython SDK
The shortest path from a sensor loop to stored telemetry.
pip install xpectra
export XPECTRA_API_KEY=sk_xp_…The client reads XPECTRA_API_KEY from the environment, so in most cases you
never write a header.
from xpectra import Client
client = Client()
for experiment in client.experiments.list():
print(experiment.name)Registering and streaming
from xpectra import Client
client = Client()
experiment = client.experiments.get(experiment_id)
dataset = experiment.datasets.register(
name="cold-flow-014",
channels=[
{"name": "thrust", "unit": "kN", "dataType": "float"},
{"name": "chamber_pressure", "unit": "bar", "dataType": "float"},
],
sample_rate_hz=1000,
)
print(dataset.id)register posts the same body as
POST /api/streams/register, so everything on
that page applies — in particular, channel order fixes the ch_N indices.
From a live generator
from datetime import datetime, timezone
def my_sensor():
while acquiring:
yield datetime.now(timezone.utc), "thrust", read_adc()
accepted = dataset.ingest_live(my_sensor())
print(f"{accepted} points")ingest_live streams without buffering the whole run in memory, which is what
you want for anything that runs longer than it takes to fill RAM.
Configuration
| Environment | Default | |
|---|---|---|
| API key | XPECTRA_API_KEY | — |
| gRPC endpoint | XPECTRA_ENDPOINT | app.xpectraflow.com:50051 |
| HTTP base | XPECTRA_BASE_URL | https://app.xpectraflow.com |
client = Client(api_key="sk_xp_…", endpoint="localhost:50051", use_tls=False)Known rough edges
These are real and worth knowing before you debug them yourself.
When not to use it
The SDK speaks the per-point gRPC path. It does not implement the columnar
TelemetryFrame format the high-rate pipeline uses, so for genuinely high-rate
first-party production you want a producer built against the frame format
directly. Talk to us if you need one.
For everything up to a few kHz, the SDK is the right tool.