Sending Data

The Plexus client sends telemetry from your device to the Plexus platform over HTTP. You authenticate with an API key, call send() or send_batch(), and your data appears in the dashboard in real time.

Authentication

You can provide your API key in three ways (checked in this order, highest priority first):

Constructor argument:
```
px = Plexus(api_key="plx_xxxxx")
```
Environment variable (recommended):
```
export PLEXUS_API_KEY=plx_xxxxx
```
Config file at ~/.plexus/config.json:
```
{ "api_key": "plx_xxxxx" }
```

Sending a Single Value

Use px.send() to send one data point at a time. Pass the metric name, value, and optional timestamp and tags.

from plexus import Plexus
 
px = Plexus()
 
# Number (most common)
px.send("temperature", 72.5)
 
# With timestamp and tags
px.send("temperature", 72.5, timestamp=1700000000, tags={"location": "lab"})
 
# String
px.send("status", "running")
 
# Boolean
px.send("motor.enabled", True)
 
# Dict (structured data)
px.send("position", {"x": 1.2, "y": 3.4, "z": 0.0})
 
# List (waveform, joint angles)
px.send("joint_angles", [0.5, 1.2, -0.3, 0.0, 1.1, -0.8])

`send()` Parameters

Parameter	Type	Required	Description
`metric`	`str`	Yes	Metric name. Use dot notation for hierarchy (`engine.rpm`).
`value`	`number \| str \| bool \| dict \| list`	Yes	The value to send.
`timestamp`	`float`	No	Unix timestamp. Defaults to `time.time()`.
`tags`	`dict`	No	Key-value labels for filtering.

Sending Batches

Use px.send_batch() when you have multiple metrics to send at the same moment. This is more efficient than multiple send() calls because it sends a single HTTP request.

# List of (metric, value) tuples
px.send_batch([
    ("temperature", 72.5),
    ("humidity", 45.2),
    ("pressure", 1013.25),
])
 
# With tags applied to all points
px.send_batch([
    ("motor.rpm", 3450),
    ("motor.current", 2.1),
    ("motor.temp", 65.0),
], tags={"motor_id": "A"})

`send_batch()` Parameters

Parameter	Type	Required	Description
`points`	`list[tuple]`	Yes	List of `(metric, value)` tuples.
`timestamp`	`float`	No	Unix timestamp applied to all points. Defaults to `time.time()`.
`tags`	`dict`	No	Key-value labels applied to all points in the batch.

Return Values and Exceptions

send() and send_batch() return bool — True on success. On failure, they may raise:

AuthenticationError — invalid or missing API key
PlexusError — network error or server error (data stays in buffer for retry)

Context Manager and Cleanup

The client supports use as a context manager and has a close() method for cleanup:

# Context manager — auto-closes on exit
with Plexus() as px:
    px.send("temperature", 72.5)
 
# Or close manually
px = Plexus()
px.send("temperature", 72.5)
px.close()

Sessions

Sessions group related data points for analysis and playback. Use a context manager to start and stop a session automatically:

with px.session("thermal-cycle-001"):
    for i in range(100):
        px.send("temperature", read_temp())
        px.send("fan_speed", read_fan())
        time.sleep(1)
 
# With optional tags applied to the session
with px.session("thermal-cycle-002", tags={"operator": "jchen", "rig": "A"}):
    px.send("temperature", read_temp())

Everything sent inside the with block is tagged with the session ID. You can replay, compare, and export sessions from the dashboard.

Buffering

The client buffers data locally and sends it in batches. If the network is unavailable, data is retained and retried automatically.

In-Memory Buffer (Default)

By default, the client holds up to 10,000 points in memory. If the buffer fills up, the oldest points are dropped.

px = Plexus()
px.buffer_size()     # Returns number of points in the buffer
px.flush_buffer()    # Force-send all buffered data now

Persistent SQLite Buffer

For devices that lose power or connectivity, enable the persistent buffer. Data is written to a local SQLite database and survives restarts.

px = Plexus(persistent_buffer=True)
 
# Buffer survives crashes and power loss
px.send("temperature", 72.5)  # Written to SQLite immediately
 
px.buffer_size()      # Points waiting to be sent
px.flush_buffer()     # Force-send now

Retry Behavior

Failed sends are retried with exponential backoff (jitter applied to each delay):

Retry	Delay
1	~1 second
2	~2 seconds
3	~4 seconds

The client makes up to 3 retries (4 total attempts, max delay 30s). After all retries are exhausted, the data stays in the buffer and retries on the next send() call.

Full Example

from plexus import Plexus
import time
 
px = Plexus(persistent_buffer=True)
 
with px.session("endurance-test-001"):
    while True:
        px.send_batch([
            ("motor.rpm", read_rpm()),
            ("motor.temperature", read_motor_temp()),
            ("coolant.flow_rate", read_flow()),
        ], tags={"rig": "A", "operator": "jchen"})
        time.sleep(0.1)

Next Steps

CLI Reference — Run the agent from the command line
Commands — Accept remote commands from the dashboard
Adapters — Connect to CAN bus, MQTT, MAVLink, and more

Send from Any Language CLI Reference