Sending data to Refract

RefractIO from Python

refract-io-lib is the official client for RefractIO. It is published on PyPI and imported as refract_io. Requires Python 3.10 or newer.

pip install refract-io-lib

Quick start

import time
from refract_io import RefractStream, float32, float64

stream = RefractStream()          # finds Refract on the network

stream.create_table(
    id=1,
    name="sensors",
    columns={"timestamp": float64,
             "temperature": float32,
             "pressure": float64},
)
stream.create_table(
    id=2,
    name="gps",
    columns={"timestamp": float64,
             "latitude": float64,
             "longitude": float64},
)

try:
    t0 = time.time()
    while True:
        t = time.time() - t0
        stream.send_row(1, [t, read_temperature(), read_pressure()])
        stream.send_row(2, [t, lat, lon])
        time.sleep(0.001)
finally:
    stream.close()

In Refract, open a connection with transport RefractIO on port 50051. Traces appear, grouped by table, as soon as the client registers.

API

RefractStream(host=None, port=None)

With both omitted, discovers Refract over Bonjour and falls back to localhost:50051.

create_table(id, name, columns)

Register a schema. name becomes the group in the source browser; column order defines the packing order for every row.

send_row(table_id, values)

Send one row, packed per the declared types.

send_rows(table_id, rows)

Send a batch of rows in one call.

start() / close()

Open and close the connection. The first send starts it automatically; stop() is an alias for close().

discover_refract(timeout=3.0)

Returns (host, port) or None.

RefractStream is also a context manager:

with RefractStream() as stream:
    stream.create_table(id=1, name="run", columns={"t": float64, "v": float64})
    for t, v in samples:
        stream.send_row(1, [t, v])

Sending is asynchronous and thread-safe, so send_row does not block an acquisition loop.

Column types

Import type constants from refract_io: float32, float64, int8, int16, int32, int64, uint8, uint16, uint32, uint64. They are aliases for members of the exported ValueType enum.

Network discovery

Refract advertises its listeners over Bonjour as _refract._tcp.

RefractStream() with no address browses for three seconds, prefers a waiting listener, and falls back to localhost:50051. Discovery uses multicast DNS, so pass the host explicitly across subnets or VPNs.

from refract_io import RefractStream, discover_refract

found = discover_refract(timeout=5.0)
stream = RefractStream(*found) if found else RefractStream("localhost", 50051)

Wire protocol

To write a client in another language, implement this service:

syntax = "proto3";
package refract.kvstream;

enum ValueType {
  FLOAT32 = 0;  FLOAT64 = 1;
  INT8    = 2;  INT16   = 3;  INT32  = 4;  INT64  = 5;
  UINT8   = 6;  UINT16  = 7;  UINT32 = 8;  UINT64 = 9;
}

message ColumnDef   { string name = 1; ValueType value_type = 2; }
message RegisterTable {
  uint32 table_id = 1;
  string name = 2;
  repeated ColumnDef columns = 3;
}
message TableRow {
  uint32 table_id = 1;
  bytes values = 2;   // raw bytes for all columns, in registration order
}

message StreamMessage {
  oneof payload {
    RegisterTable register_table = 1;
    TableRow      table_row      = 2;
  }
}
message StreamResponse { bool ok = 1; string error = 2; }

service KVStream {
  rpc Stream (stream StreamMessage) returns (StreamResponse);
}

Open the Stream RPC, send a RegisterTable per table, then send TableRow messages. A row's values field is the concatenated little-endian bytes of each column in registration order.

Microcontrollers

The simplest path from a board is to print delimited text.

void setup() {
  Serial.begin(115200);
}

void loop() {
  float t  = millis() / 1000.0;
  Serial.print(t);            Serial.print(',');
  Serial.print(readAccelX()); Serial.print(',');
  Serial.println(readAccelY());
  delay(10);
}

In Refract, load the Serial / CSV preset, choose your port and baud rate, and connect. Assign the timestamp column as the x-axis — see X-axis transforms.

For a packed binary struct, use the binary packet builder.