Python SDK

Best for Raspberry Pi, NVIDIA Jetson, and any Linux-based board.

1. Setup client

pip install plexus-python

The base package only needs websocket-client — it's the right choice for metrics, events, and commands. Video and thermal streaming require heavier dependencies; only add them when needed:

The [video] extra adds OpenCV headless, Pillow, and NumPy (~100 MB).

Create a client with your API key and a stable source_id for the device. Plexus registers it automatically on first send.

from plexus import Plexus

px = Plexus(
    api_key="YOUR_API_KEY",
    source_id="robot-01",
)

In production, set PLEXUS_API_KEY as an environment variable instead of hardcoding the key — the SDK picks it up automatically:

px = Plexus(source_id="robot-01")  # reads PLEXUS_API_KEY from the environment

Get your API key from API Keys in the dashboard.

2. Send metrics

# Single metric
px.send(metric="battery.voltage", value=12.4)

# With tags — slice by location, firmware version, etc.
px.send(
    metric="sensor.temperature_c",
    value=23.5,
    tags={"location": "warehouse-a", "firmware": "1.4.2"},
)

# Batch — reduces network overhead; recommended on constrained hardware
px.send_batch(
    points=[
        ("battery.voltage",      12.4),
        ("battery.current_ma",   340),
        ("sensor.temperature_c", 23.5),
        ("sensor.humidity_pct",  54.1),
        ("motor.rpm",            1200),
    ],
    tags={"firmware": "1.4.2"},
)

Metric names follow <subsystem>.<metric> — e.g. battery.voltage, motor.rpm, sensor.temperature_c.

By default the SDK timestamps on send. To use a device-side clock, pass timestamp as Unix seconds:

import time

px.send(metric="battery.voltage", value=12.4, timestamp=time.time())

Batch fast readings

Each send() or send_batch() call goes out as its own message, and the gateway's rate limits count messages, not points (see Limits). If readings arrive faster than a few times a second, use px.batch(). It collects readings and sends them together on a background thread:

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

Leaving the with block sends whatever is still queued. Outside a with block, call b.start() first and b.close() when you're done; b.flush() sends the queue right away.

Retries and offline buffering

When a send fails, the SDK retries with exponential backoff (3 retries by default). Points it still can't deliver are saved to a file for that source, ~/.plexus/buffer-<source_id>.db — up to 10,000 points by default — and go out with that source's next successful send. Change the cap with Plexus(max_buffer_size=...).

When a send still fails, send(), send_batch() and event() raise. The points are already saved, so catch the error and keep going:

import time
from plexus import Plexus, PlexusError

px = Plexus(source_id="robot-01")

while True:
    try:
        px.send("battery.voltage", read_voltage())
    except PlexusError as err:
        print("send failed, will retry:", err)
    time.sleep(1)

PlexusError covers timeouts, connection failures and server errors. AuthenticationError (a rejected key) and RateLimitedError (the gateway dropped messages for rate) are kinds of PlexusError.

On the Free plan the gateway refuses the SDK's live WebSocket, and the SDK sends over HTTP instead. You don't need to change any code. See Plans and limits.

3. Send events

Use event() for non-metric data: faults, state changes, and important log lines.

px.event(name="device.fault",       data={"code": "SENSOR_TIMEOUT", "sensor": "imu"})
px.event(name="motor.state_change", data={"from": "IDLE", "to": "RUNNING"})
px.event(name="system.boot",        data="firmware-1.4.2")

Events are stored separately from metrics and queryable via the Events API. A plain string value can be up to 256 characters; a dict or list up to 4,096 bytes of JSON. For forwarding log lines, see Send logs and events.

4. Stream video

Video needs Pro: frames travel over the live WebSocket, which Free doesn't include. Frames are relayed live and not stored. Video is saved only when someone presses Record on the camera in the app — up to 4 hours per recording. Recorded video is stored at $0.50 per GB-month.

Install the video extras first (OpenCV headless + Pillow + NumPy):

pip install "plexus-python[video]"

Use send_video_frame() for frame-by-frame control (e.g. with OpenCV or picamera2), or stream_camera() to pull from a URL via FFmpeg.

Frame by frame

import cv2
from plexus import Plexus

px = Plexus(api_key="YOUR_API_KEY", source_id="robot-01")

cap = cv2.VideoCapture(0)
while True:
    ret, frame = cap.read()
    if ret:
        px.send_video_frame(frame=frame, camera_id="camera:0")

Frames are JPEG-encoded at quality=85 by default. Frames over ~750 KB are adaptively re-encoded; the gateway hard limit is 1 MB.

Stream from a URL

Requires FFmpeg on $PATH. Accepts any FFmpeg-supported source — RTSP, HLS, a video file path.

stop = px.stream_camera(url="rtsp://192.168.1.10/stream", camera_id="camera:0")

# ... do other work ...

stop.set()  # stop streaming

5. Stream thermal camera

Supports MLX90640 (32×24) and MLX90641 (16×12) sensors wired to I2C.

pip install "plexus-python[video]"

Note: I2C sensors (MLX90640, MLX90641) also require a board-specific driver installed separately on the device. Refer to your sensor's documentation.

from plexus import Plexus
from plexus.cameras.thermal import ThermalSource

px = Plexus(source_id="robot-01")

cam = ThermalSource.open("mlx90640")  # or "mlx90641"
while True:
    px.send_thermal_frame(cam.read_frame(), camera_id="thermal")

To test the full pipeline without hardware, pass "sim":

cam = ThermalSource.open("sim")

Raspberry Pi setup

On a Raspberry Pi (or any Linux box), one command installs the SDK and checks the device can reach Plexus. Create a key under API Keys, then run on the Pi:

curl -sL https://app.plexus.company/setup | bash -s -- --key plx_... --name greenhouse-pi-01

The script:

• creates a virtual environment at /opt/plexus/venv (or ~/.plexus/venv without sudo) and installs plexus-python into it
• saves the key and device name to ~/.plexus/config.json
• sends one test reading, setup.check, so the device shows up in the app

It also installs Python if it is missing. Where apt is available and it can use sudo, it installs python3-venv (and libcap-dev on a Pi) if they are missing. With sudo it also adds your user to the i2c group, when that group exists.

The test reading counts as the device's first data. On a new account, that is what the starter dashboard is built from, and the device uses one of the 3 that Free covers.

The device's source_id comes from --name: lowercased, with spaces and other characters turned into -. Pass --slug to set it exactly.

The script doesn't install a service or run anything in the background. You write the script that reads your sensors and calls px.send(). To start it on boot, write your own systemd unit that runs it with /opt/plexus/venv/bin/python (or ~/.plexus/venv/bin/python). The Raspberry Pi fleet observer guide has a full example unit.