otensor

Python library for writing IoT automations as code — react to device events, run schedules, and trigger actions or notifications on the Otensor platform. Runs anywhere (laptop, server), not on the device itself.

PT-BR — Biblioteca Python para escrever automações IoT em código: reagir a eventos de dispositivos, rodar agendamentos e disparar ações ou notificações na plataforma Otensor. Roda em qualquer lugar (laptop, servidor), não no dispositivo. A documentação abaixo está em inglês.

from otensor import connect

bot = connect()  # reads OTENSOR_API_KEY / OTENSOR_API_BASE_URL

bot.when("living-room-sensor", "temperature").above(30).do("fan", "turn_on")
bot.every_day("18:00").do("porch-light", "turn_on")

bot.run()

---

Which package do I need?

Otensor ships two Python packages with different jobs:

	otensor (this one)	otensor-sdk
Runs	Anywhere (laptop, server)	On the hardware (Raspberry Pi, Linux board)
Job	React to events, write automations	Publish telemetry, execute commands
Talks to	REST API	MQTT broker + REST API

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) and its base_url
• An API key (sk-…) generated in the dashboard

Install

pip install otensor

Optional extras: otensor[email] (SMTP alerts), otensor[industrial] (Modbus/OPC-UA/InfluxDB), otensor[integrations] (Google Calendar, Slack), or otensor[all] for everything.

Configuration

connect()/Client() read their config from environment variables (loaded from a .env file automatically, or from whatever's already exported) — no need to pass them as arguments every time:

OTENSOR_API_KEY=sk-...
OTENSOR_API_BASE_URL=http://localhost:8000   # or https://api.your-otensor.example

Pass api_key/base_url explicitly to override the corresponding env var — useful for multi-instance scripts or tests. Neither has a hardcoded literal default: a value baked into the library would silently point at the wrong instance when you switch environments. If a value is missing from both the argument and the environment, Client raises ValueError right away.

Two modes

Simple mode — fluent chaining

No classes, no decorators. Best for straightforward rules.

from otensor import connect

bot = connect()

bot.when("living-room-sensor", "temperature").above(30).do("fan-relay", "turn_on")
bot.when("living-room-sensor", "temperature").below(25).do("fan-relay", "turn_off")

bot.when("garage-door", "status").equals("open") \
   .notify(email="owner@home.com", message="Garage door was opened!")

bot.weekdays("08:00").do("office-gate", "open")
bot.every_day("06:00").do("garden-valve", "open")

bot.run()  # blocks until Ctrl-C

Advanced mode — decorators + AutomationContext

Full control: multiple conditions, custom logic, direct access to the event context.

from otensor import Client, on, schedule, when

client = Client(api_key="sk-...", base_url="http://localhost:8000")


@on(client.device("living-room-sensor").property("temperature").above(30))
def turn_on_fan(ctx):
    ctx.command("fan-relay", action="turn_on")
    ctx.log(f"Fan turned on — temperature: {ctx.value}°C")


@on(client.device("living-room-sensor").property("temperature").below(25))
@when(lambda ctx: ctx.value is not None)
def turn_off_fan(ctx):
    ctx.command("fan-relay", action="turn_off")


@schedule("0 8 * * 1-5", client=client)
def office_opening(ctx):
    ctx.command("office-gate", action="open")
    ctx.notify(channel="whatsapp", message="Office is open.")


client.run()

test_mode() and simulate() let you exercise handlers without a live MQTT connection — useful in unit tests:

bot = connect(api_key="sk-test", base_url="http://localhost:8000").test_mode()
bot.when("sensor-01", "temperature").above(30).do("fan-relay", "turn_on")
bot.simulate("sensor-01", "temperature", 35)

API reference

Simple mode (SimpleBot)

Symbol	Purpose
connect(api_key=None, *, base_url=None, tenant_id=None) -> SimpleBot	Entry point; api_key/base_url fall back to OTENSOR_API_KEY/OTENSOR_API_BASE_URL
.when(device_id, property_name)	Start a condition — chain .above(v) / .below(v) / .equals(v) / .changes()
...trigger.do(target_device, action)	Run an action when the condition matches
...trigger.notify(email=None, *, message="", channel=None)	Send an email and/or a channel notification
.every_day(time) / .weekdays(time) / .weekends(time)	Start a schedule (time is "HH:MM") — chain .do(device_id, action)
.run()	Connect to MQTT and block until Ctrl-C
.test_mode() / .simulate(device_id, property_name, value)	Exercise handlers without a live connection

Advanced mode (Client, decorators, AutomationContext)

Symbol	Purpose
Client(api_key=None, *, base_url=None, tenant_id=None)	Entry point; api_key/base_url fall back to OTENSOR_API_KEY/OTENSOR_API_BASE_URL
client.device(device_id).property(name)	Build a condition — same .above/.below/.equals/.changes() chain
@on(condition)	Register a handler for a device event
@schedule(cron, *, client=None)	Register a handler on a cron schedule
@when(predicate)	Extra filter stacked on top of @on
client.run()	Connect to MQTT, start the scheduler, block until Ctrl-C
ctx.value / .device_id / .property / .unit / .ts	Event data injected into every handler
ctx.command(device_id, *, action, payload=None, slot=None)	Send a tracked command (ack/history)
ctx.send_email(to, *, subject="", body="")	Send an email via the platform
ctx.notify(*, channel="email", message="")	Send a notification via the platform
ctx.log(message, *, level="info")	Structured log
ctx.history(n=10)	Last n readings for the current device/property
ctx.audit(*, action, details=None)	Write an audit entry

Integrations

Symbol	Purpose
Email.configure(*, smtp_host, smtp_port=587, username, password, from_addr)	One-time SMTP setup (otensor[email])
Email.send(to, *, subject="", body="")	Send an email standalone (outside a handler)
Webhook.post(url, *, payload=None, headers=None, timeout=10.0)	POST JSON to any URL, raises on non-2xx

Roadmap

Not in 0.1 — dropped when the public API was cut down to what actually works, tracked for a later release:

• Duration filters (for_duration(minutes=...)) — trigger only after a condition holds for a period, not on the first match.
• Pipeline — chained source → publish → condition → action, for direct hardware reads (e.g. Modbus) without going through a device already published to the platform.
• @debounce — suppress repeated firings within a time window.
• @learn — record handler behavior for the AI Engine to learn from.
• @predictive — trigger on ML-detected anomalies rather than fixed thresholds.
• ctx.ask_user() — pause a handler and wait for human input via notification.

Versioning

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

otensor~=0.1.0

License

MIT