RelayX Device SDK for Python

Official Python SDK for connecting IoT devices to the RelayX platform.

View Full Documentation →

Installation

pip install relayx_device_sdk

Quick Start

import asyncio
from relayx_device_sdk import RelayDevice

device = RelayDevice({
    "api_key": "<YOUR_API_KEY>",
    "secret": "<YOUR_SECRET>",
    "mode": "production",  # or "test"
})

async def main():
    await device.connect()

    # Publish telemetry
    await device.telemetry.publish("temperature", 22.5)

    # Emit logs (batched to backend, mirrored to console)
    device.log.info("device booted")
    device.log.warn("disk usage at", 87, "%")

    # Listen for RPC calls
    async def on_reboot(req):
        print("Reboot requested:", req.payload)
        await req.respond({"status": "rebooting"})

    await device.rpc.listen("reboot", on_reboot)

    # Listen for commands
    def on_firmware_update(msg):
        print("Firmware update:", msg.payload)

    await device.command.listen("firmware_update", on_firmware_update)

    # Disconnect when done
    await device.disconnect()

asyncio.run(main())

Configuration

device = RelayDevice({
    "api_key": "<YOUR_API_KEY>",   # JWT issued by RelayX
    "secret": "<YOUR_SECRET>",      # NKEY seed
    "mode": "production",           # "production" | "test"
})

Functionality

Connection

await device.connect()     # returns True on success, False if already connected or failed
await device.disconnect()  # drains NATS connection, cleans up consumers

# Listen for connection status changes
device.connection.listeners(lambda event: print("Status:", event["type"]))
# event["type"]: "connected" | "disconnected" | "reconnecting" | "reconnected" | "auth_failed"

Telemetry

Fire-and-forget sensor data publishing. Readings are validated against the device schema fetched on connect.

await device.telemetry.publish("temperature", 22.5)          # number
await device.telemetry.publish("status", "online")            # string
await device.telemetry.publish("active", True)                # boolean
await device.telemetry.publish("metadata", {"fw": "1.2"})    # json

Each message is published with a server-synced timestamp.

Schema validation: On connect, the SDK fetches the device schema from the server. If a schema exists, publish() will throw a ValidationError if the metric name is not in the schema or the reading type does not match.

Remote Procedure Calls (RPC)

Register handlers for incoming RPC calls.

# Register a handler
async def on_get_status(req):
    print("Payload:", req.payload)

    # Respond with success
    await req.respond({"uptime": 12345})

    # Or respond with error
    # await req.error({"code": "UNAVAILABLE", "message": "Device busy"})

await device.rpc.listen("get_status", on_get_status)

# Unregister
await device.rpc.off("get_status")

Duplicate listeners for the same name throw DuplicateListenerError.

Commands

One-way commands delivered for long running tasks that do not require a status update.

def on_firmware_update(msg):
    print("Command:", msg.payload)

    # Process based on data...

await device.command.listen("firmware_update", on_firmware_update)

await device.command.off("firmware_update")

Config

Get and set device configuration

# Fetch current config
config = await device.config.get()
print(config)

# Update config
success = await device.config.set({"interval": 10000, "name": "sensor-1"})

Events

Fire-and-forget event publishing of events (fault codes, etc).

await device.event.send("door_opened", {"door_id": "front", "timestamp": device.time.now()})

Logs

Structured device logging with three levels (info, warn, error). Each call accepts a variadic argument list — strings, numbers, booleans, plain dicts, lists, datetimes, None, and exception instances are all accepted; callables and other non-serializable values raise ValidationError.

device.log.info("hello world")
device.log.info("a number reading", 42)
device.log.warn("careful — disk usage at 87%")
device.log.error("parse failed", ValueError("bad json"))
device.log.info("an object", {"port": 8080, "retries": 3})

Locally, each call also mirrors to Python's standard logging module (at INFO/WARNING/ERROR) so you keep your normal dev experience.

Batching. Entries are buffered in memory and flushed to the backend either when 15 entries accumulate or after 5 seconds, whichever comes first. disconnect() flushes any pending buffer.

Time

NTP-synchronized clock. Syncs with time.google.com on connect and every 3 hours. Auto-resyncs on reconnect if stale.

# Initialize (called automatically on connect)
await device.time.init()

# Get current server-corrected timestamp (ms)
now = device.time.now()

# Convert SDK timestamp to datetime
dt = device.time.to_date(now)

# Convert datetime to SDK timestamp
ts = device.time.to_timestamp(datetime.now())

# Set timezone for display
device.time.set_timezone("America/New_York")

Error Handling

The SDK exports four error types:

from relayx_device_sdk import (
    NotConnectedError,       # Operation attempted while disconnected
    DuplicateListenerError,  # rpc.listen() or command.listen() called twice for same name
    ValidationError,         # Invalid arguments or schema mismatch
    TimeoutError,            # Request/reply timed out
)

Offline Behavior

• Telemetry & Events (publish): Messages are buffered in memory while disconnected and flushed automatically on reconnect.
• Logs (device.log.*): Entries are buffered and flushed on the same 15-entry / 5-second policy. Any buffer left at disconnect() is flushed before the connection drains.
• RPC & Commands (listen): Throw NotConnectedError if transport is disconnected.
• Config (get/set): Throw NotConnectedError if transport is disconnected.

Testing

pytest

The SDK is designed for full unit testability. All subsystems accept a transport dependency that can be mocked:

from relayx_device_sdk import RelayDevice

mock_transport = ...  # mock methods
device = RelayDevice._create_for_test(
    {"api_key": "test", "secret": "test", "mode": "test"},
    mock_transport,
)

License

Apache-2.0