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. Copying API keys around by hand is also where keys end up in the wrong place.

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"})

Sign in from the terminal

Installing the package also installs a plexus command. plexus init opens your browser, you sign in, and the CLI saves a new API key to ~/.plexus/config.json. The SDK reads it from there, so your script doesn't need a key in it.

pip install plexus-python
plexus init        # opens the browser, saves a key
plexus whoami      # checks the saved key still works
plexus logout      # removes it from this machine

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

The prompt is on the API Keys page. It points the agent at plexus.company/agent.md, a setup guide written for agents. You approve two things along the way: the list of metrics and the key.

For the key, the agent calls POST /api/auth/claim and gets a link like app.plexus.company/claim/ABCD-EFGH. You open it while signed in and click Approve or Deny. Only workspace admins and editors can approve. The agent polls until you decide, then receives the key exactly once. It can only write telemetry to your workspace, and the request expires after 15 minutes.

What the agent does

• Installs plexus-python for Python projects. Other languages post JSON over HTTP, so no SDK is needed.
• Reads the code and proposes 5 to 10 metrics and events, with file, line and why, then waits for your OK.
• Requests the key and saves it as PLEXUS_API_KEY in your gitignored env file. It never hardcodes or commits it.
• Adds px.send() and px.event() calls under one source id, runs the code so a real point arrives, and gives you a link to its dashboard.

Using it

Copy the prompt from API Keys in the app. It starts like this:

Set up Plexus telemetry in this project.

1. Fetch https://plexus.company/agent.md and follow it as your reference.
2. Python projects: pip install plexus-python. Any other language: no SDK — POST JSON to the gateway /ingest endpoint over plain HTTP (x-api-key header).
3. Read the codebase and propose 5-10 metrics/events worth tracking ... wait for my approval.
...

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.