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:

What you need	Install
Metrics, events, commands	`pip install plexus-python`
Video / thermal streaming	`pip install "plexus-python[video]"`

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())

3. Send events

Use event() for non-metric data: faults, state changes, structured logs.

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 Logs API.

4. Stream video

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