otensor-sdk

Python SDK that runs on your device — publishes telemetry and receives commands from the Otensor IoT platform. Built for makers on Raspberry Pi and other Linux hardware.

PT-BR — SDK Python que roda no dispositivo: publica telemetria e recebe comandos da plataforma Otensor. Feito para makers com Raspberry Pi e outro hardware Linux. A documentação abaixo está em inglês.

from otensor_sdk import OtensorSDK

device = OtensorSDK(api_base_url=..., api_key=..., device_id=...).connect()
sensor = device.slot("main")

sensor.publish_property("temperature", 24.5)

@sensor.on_action("set_pixels")
def light_up(payload: dict) -> None:
    print("platform asked for:", payload)

Which package do I need?

Otensor ships two Python packages with different jobs:

Use both if you want a device that reports data and logic that reacts to it. They are independent — neither requires the other.

Requirements

• Python 3.11+
• An Otensor instance (self-hosted or managed)
• An API key (sk-…) and a device registered in the dashboard

Install

pip install otensor-sdk

Hardware wrappers (otensor_sdk.sensehat, otensor_sdk.gpio) rely on sense-hat and gpiozero, which ship with Raspberry Pi OS. They are not pip dependencies of this package, so installing here never drags in hardware libraries you don't need.

Configuration

API_BASE_URL=https://api.your-otensor.example   # or http://localhost:8000
API_KEY=sk-...
DEVICE_ID=...

Quick start

import os
from otensor_sdk import OtensorSDK

sdk = OtensorSDK(
    api_base_url=os.environ["API_BASE_URL"],
    api_key=os.environ["API_KEY"],
    device_id=os.environ["DEVICE_ID"],
)

device = sdk.connect()          # fetches MQTT credentials and connects
main = device.slot("main")      # a slot is one capability of the device

# Register handlers BEFORE publishing, so a command arriving right after
# connect() is not missed.
@main.on_action("set_pixels")
def handle_set_pixels(payload: dict) -> None:
    print("light up:", payload)

main.publish_property("temperature", 24.5)

Slots

A device declares capabilities, each bound to a named slot — a board can carry a sensor unit on main, a relay on relay1 and a motion sensor on pir1 at the same time. Properties and actions are always scoped to a slot, so the platform validates each reading against the right schema.

device.slot("main").publish_property("humidity", 61.2)
device.slot("relay1").publish_property("state", True)

Publishing on a schedule

TelemetryPublisher groups every source into one message per tick, instead of one message per property:

from otensor_sdk import TelemetryPublisher

publisher = TelemetryPublisher(device, interval=5.0)
publisher.set_source("main", "temperature", read_temperature)
publisher.set_source("main", "humidity", read_humidity)

publisher.run_forever()          # blocks until Ctrl-C
# publisher.publish_once()       # or drive the loop yourself

Complete example — no hardware required

Runs anywhere with simulated sensors, so you can see the full loop before touching a Raspberry Pi:

import os
import random
from otensor_sdk import OtensorSDK, TelemetryPublisher

sdk = OtensorSDK(
    api_base_url=os.environ["API_BASE_URL"],
    api_key=os.environ["API_KEY"],
    device_id=os.environ["DEVICE_ID"],
)

device = sdk.connect()
main = device.slot("main")

@main.on_action("set_pixels")
def set_pixels(payload: dict) -> None:
    print("LED matrix ->", payload.get("pixels"))

@main.on_action("clear_display")
def clear_display(payload: dict) -> None:
    print("LED matrix cleared")

publisher = TelemetryPublisher(device, interval=5.0)
publisher.set_source("main", "temperature", lambda: round(random.uniform(18, 32), 1))
publisher.set_source("main", "humidity", lambda: round(random.uniform(40, 80), 1))

try:
    publisher.run_forever()
except KeyboardInterrupt:
    device.disconnect()

Raspberry Pi hardware

from otensor_sdk.sensehat import SenseHatUnit
from otensor_sdk.gpio import DigitalInput, DigitalOutput

unit = SenseHatUnit()
unit.bind("main", device, publisher)          # sensors + LED matrix, wired in one call

DigitalOutput(pin=17).bind("relay1", device)  # actions: turn_on / turn_off / toggle
DigitalInput(pin=4).bind("pir1", device)      # publishes motion_detected on edge

bind() registers the action handlers and the telemetry sources for that slot, so you don't wire each property by hand.

Reliability

Built for devices that stay online for weeks on flaky networks:

• Reconnect-safe commands — the cmd topic is re-subscribed on every connect, including automatic reconnects. Without this, telemetry keeps flowing while commands go silently mute after the first network blip.
• State republish — device.on_reconnect(fn) fires on first connect and every reconnect. Hardware wrappers use it to re-publish current state, which the dashboard would otherwise show as stale until the next physical change.
• Last Will — the broker marks the device offline if it drops without a clean disconnect.
• Local ack — device.on_ack(fn) fires in-process the moment this device confirms a command, with {commandId, slot, action, success, error?}. Lets local code know its own command finished without a REST round-trip.

API reference

Exceptions: OtensorError (base), OtensorAuthError, OtensorSessionError, OtensorHardwareError.

Versioning

Semantic versioning. While on 0.x, a minor bump may carry breaking changes — pin the minor version in production:

otensor-sdk~=0.2.0

License

MIT