Python SDK

Why

Most hardware teams already have Python on the device: a Raspberry Pi, a Jetson, a test bench PC. Getting readings off it should be a pip install and a function call, not a daemon to configure.

How it works

Create a client with a source_id, the device's name in Plexus, and call send() with a metric name and a value. Numbers become time series. Strings, dicts and lists become events. The device appears in Plexus on its first point.

The SDK connects to the gateway over a WebSocket and falls back to HTTP by itself if the socket isn't available. Points that can't be sent are retried, then kept in a SQLite buffer on disk and sent with the next successful send, so a reboot or a network drop doesn't lose them. Over the WebSocket the SDK also corrects for a wrong device clock, which matters on boards that boot without NTP.

from plexus import Plexus

px = Plexus(source_id="rover-02")   # reads PLEXUS_API_KEY

px.send("battery.voltage", 12.4)
px.send("motor.rpm", 3450, tags={"motor_id": "A1"})
px.event("state_change", {"from": "IDLE", "to": "RUNNING"})

Stable device identity

A device's name is its source_id, and it should stay the same for the life of the device. Pass it in code, or set it once with the setup script's --name. If you set neither, the SDK makes up an id like source-1a2b3c4d on first run and saves it, so the device keeps that name after a restart.

Give every device its own name. Two devices that use the same name write into the same device in Plexus. Don't use the hostname: every Raspberry Pi boots as raspberrypi, and cloned SD cards would all merge into one device.

curl -sL https://app.plexus.company/setup | bash -s -- --key plx_... --name rover-02

How it works

px.batch() gives you a sender with the same send() call. It queues readings and a background thread sends them as one message every interval_ms. Leaving the with block sends whatever is still queued. Each reading keeps the time it was taken, not the time it was sent.

from plexus import Plexus

px = Plexus(source_id="rig-01")   # reads PLEXUS_API_KEY

with px.batch(interval_ms=50) as b:
    while running:
        b.send("imu.accel_x", imu.x)
        b.send("imu.accel_y", imu.y)
        b.send("motor.rpm", encoder.rpm)

Runs from code

A run is a named window on a device, like one hotfire or one test cycle. You can now open and close it from the script that drives the bench. Runs show up on the Runs page, where you can compare them lined up at T+0.

Add pass criteria and Plexus checks them when the run closes. Every sample of that metric inside the window must meet the limit. A metric with no data in the window fails. A clean exit from the with block closes the run as completed. An exception closes it as aborted and re-raises.

with px.run("hotfire-03", pass_criteria=[
    {"metric": "motor.temp_c", "operator": "<", "value": 85},
]):
    bench.execute()

# Or split across two places:
run = px.start_run("hotfire-04", tags={"build": "a41f"})
...
result = px.end_run(run)       # result["test_result"] holds the verdict

Limits

• interval_ms defaults to 100 and is also the longest a reading waits before it is sent.
• One message holds at most max_points (default 5,000). The gateway's own ceiling is 10,000.
• If the gateway is unreachable for a long time, the queue holds up to max_pending points (default 200,000), then drops the oldest and counts them on b.dropped.
• Pass criteria are checked against stored data, which trails ingest by a few seconds. Give the last points a moment to land before you close a run.