TendrlDocs / Contact / sdks/micropython/examples
MicroPython Client Examples
Practical examples for common microcontroller use cases.
Basic Sensor Publishing
from tendrl import Client
client = Client(debug=True)
client.start()
while True:
# Replace with your actual sensor reading
data = {
"temperature": 25.5,
"humidity": 60,
"light": 850
}
client.publish(data, tags=["sensor", "environment"])
time.sleep(30)
DHT Temperature & Humidity Sensor
from tendrl import Client
sensor = dht.DHT22(machine.Pin(4))
client = Client(debug=True, offline_storage=True)
client.start()
while True:
try:
sensor.measure()
client.publish({
"temperature": sensor.temperature(),
"humidity": sensor.humidity()
}, tags=["dht22", "sensor"])
except OSError as e:
print(f"Sensor error: {e}")
time.sleep(30)
Tether Decorator
Automatically publish the return value of a function:
from tendrl import Client
client = Client(debug=True, offline_storage=True)
client.start()
@client.tether(tags=["sensors"], write_offline=True)
def read_sensors():
return {
"temperature": 25.5,
"humidity": 60,
"battery_voltage": 3.7
}
# Each call publishes the result
while True:
read_sensors()
time.sleep(10)
Inbound Message Routing
Route incoming messages by tag or msg_type with @client.on():
from tendrl import Client
client = Client(debug=True)
@client.on(tag="diagnostic")
def run_diagnostics(message):
report = self_test(message.get("data", {}))
client.publish(report, tags=["diagnostic-result"])
@client.on(tag="ai-response")
def handle_ai_reply(message):
print("AI:", message.get("data", {}).get("response"))
@client.on(tags=["alert", "anomaly"])
def handle_alert(message):
print("Alert:", message.get("data"))
@client.on(tags_all=["validated-fail", "sensor"])
def handle_validation_error(message):
client.update_state({"status": "needs_maintenance", "error": message.get("data")})
@client.on_default
def unhandled(message):
print("No route:", message.get("msg_type"), message.get("tags"))
client.start()
Camera as a Sensor (OpenMV)
Watch one region and publish only when it changes: no model, no training data. Nothing is sent while the scene is stable; a ~200-byte message crosses the network when reality changes.
Requires the vision extension:
from tendrl import Client
from tendrl.vision import Camera, Watch, roi_brightness
client = Client(mode="sync", debug=True)
client.start()
time.sleep(5)
cam = Camera()
watch = Watch(
client,
detect=roi_brightness(roi=(120, 80, 60, 90), threshold=0.45,
states=("full", "empty")),
debounce=5, # frames that must agree before a change is believed
cooldown_s=30,
tags=["coffee"], # routes to a flow / Strand workflow
)
while True:
watch.update(cam.snapshot()) # publishes ONLY on a real transition
time.sleep_ms(500)
Published on transition:
{"state": "empty", "prev": "full", "conf": 0.81}
Change the ROI, the threshold, and the tag, and the same file watches something else: an indicator lamp, a tank level, a shelf, a gate. Use roi_level(img, roi) to pick a threshold: print it in a loop, change the thing you care about, and choose a value between the two numbers you see.
Person Detection (OpenMV)
OpenMV ships a YOLO LC person detector in flash. Its weights are read in place rather than copied into the heap, so it runs on boards that could never fit a custom model, but loading the model still allocates a ~275 KB tensor arena on the heap.
from tendrl import Client
from tendrl.vision import Camera, Watch, person
client = Client(mode="sync", debug=True)
client.start()
time.sleep(5)
cam = Camera(framesize="VGA", window=(400, 400))
def on_change(state, prev, img):
if state != "person":
return
cam.lock_exposure(True) # stop auto-exposure pumping mid-clip
try:
client.send_clip_burst(cam.snapshot, duration_s=3, fps=5,
tags=["person-detected"])
finally:
cam.lock_exposure(False)
watch = Watch(
client,
detect=person(min_conf=0.4),
debounce=2,
cooldown_s=30,
tags=["person-detected"],
on_change=on_change,
)
while True:
watch.update(cam.snapshot())
time.sleep_ms(200)
Measured on an OpenMV RT1062: ~275 KB of heap to load the arena (the model reports model.ram ~270 KB), 0 bytes per inference, ~310 ms per frame. Budget that arena plus the camera framebuffer (VGA windowed to 400×400 is another ~280 KB): roughly 550 KB free before you start.
Face and Hand Detection (OpenMV)
The same deal as person detection, with Google's MediaPipe models: also in flash, also nothing to train. face is the small one (~390 KB arena, ~14 ms/frame on an AE3); palm and hand_landmarks are bigger and want a board with an NPU (AE3 or N6).
Unlike the state detectors, the *_boxes forms hand back keypoints, so you can draw what the model saw. Everything is drawn into the pixels, which means the overlay survives JPEG compression into a preview, a stream, or a clip:
from tendrl import preview
from tendrl.vision import Camera, draw_boxes, draw_keypoints, face_boxes
# MediaPipe models want SQUARE input, hence the window.
cam = Camera(framesize="VGA", window=(400, 400))
detect = face_boxes(min_conf=0.4)
while True:
img = cam.snapshot()
for x, y, w, h, score, kps in detect(img):
draw_boxes(img, [(x, y, w, h)], label="face %.2f" % score)
draw_keypoints(img, kps) # eyes, nose, mouth, ears
preview.frame(img) # draw FIRST: this compresses img in place
time.sleep_ms(50)
For the full 21-joint hand skeleton, swap in hand_landmarks and draw_skeleton:
from tendrl.vision import HAND_LINES, draw_skeleton, hand_landmarks
detect = hand_landmarks(min_conf=0.4) # two models: palm, then landmarks per hand
for x, y, w, h, score, kps, label in detect(img):
draw_boxes(img, [(x, y, w, h)], label=label) # "left" / "right"
draw_skeleton(img, kps, HAND_LINES)
To publish instead of preview, face(), palm() and hands() are the state forms; they drop into a Watch exactly like person() above. See On-device vision for the full API, and the Vision playground to tune one by eye and export the file.
JPEG Video Streaming (OpenMV)
Stream camera video to the Contact platform:
from tendrl import Client
async def main():
client = Client(mode="async", debug=True)
client.start()
# Wait for connection
await asyncio.sleep(5)
# Start streaming (auto-configures OpenMV camera)
client.start_streaming(
target_fps=15,
quality=80,
framesize="QVGA" # 320x240
)
# Keep running
while True:
await asyncio.sleep(3600)
asyncio.run(main())
Available frame sizes: QQVGA (160x120), QVGA (320x240), VGA (640x480)
Streaming with Messaging
Combine video streaming with sensor data:
from tendrl import Client
async def main():
client = Client(mode="async", debug=True)
@client.on(tag="ai-response")
def on_ai_response(message):
print(f"Received: {message}")
@client.on(tags=["alert", "anomaly"])
def on_alert(message):
print(f"Alert: {message}")
client.start()
await asyncio.sleep(5)
# Start video stream
client.start_streaming(target_fps=10, quality=70)
# Publish sensor data alongside the stream
while True:
client.publish({
"frame_rate": 10,
"uptime": time.ticks_ms() // 1000
}, tags=["camera", "status"])
await asyncio.sleep(30)
asyncio.run(main())
File Transfer Between Devices
Send a file from one entity to another. Every file is malware-scanned before delivery:
from tendrl import Client
client = Client(debug=True)
client.start()
# Send a file to another entity
client.send_file("/data/reading.csv", dest="gateway-01")
# Send to a fanout (broadcast to all members)
client.send_file("/data/firmware.bin", dest="fleet-sensors")
# Tag-routed: triggers matching Strand automations
client.send_file("/data/report.json", tags=["telemetry", "ingest"])
Receiving Files
Check for and download files addressed to your entity:
from tendrl import Client
client = Client(debug=True)
client.start()
# List clean files in your inbox
files = client.check_files(limit=10)
for f in files:
print(f"File: {f['file_name']} ({f['size']} bytes)")
# Download and save locally
data = client.download_file(f["transfer_id"])
with open(f"/data/{f['file_name']}", "wb") as fp:
fp.write(data)
Motion Clip Capture (OpenMV)
Capture a short video clip when motion is detected and upload it for review. The SDK packs JPEG frames into a zip and Contact transcodes it into a playable MP4.
Requires the vision extension (package-vision.json):
from tendrl import Client
from tendrl.vision import Camera, Watch, motion
client = Client(mode="sync", debug=True)
client.start()
time.sleep(5)
cam = Camera()
def on_change(state, prev, img):
if state != "motion":
return # the scene going still again doesn't need a video
client.send_clip_burst(
cam.snapshot,
duration_s=5,
fps=8,
tags=["motion-alert"],
meta={"zone": "driveway"},
)
watch = Watch(
client,
detect=motion(threshold=0.02),
debounce=2, # two frames must agree; ignore single-frame flickers
cooldown_s=30, # at most one clip every 30s
tags=["motion-alert"],
on_change=on_change,
)
while True:
watch.update(cam.snapshot())
time.sleep_ms(200)
Watch does the work that makes this survivable in the real world: it debounces (a change must hold for debounce frames), publishes once on transition rather than every frame, and enforces a cooldown. A raw detector fires on every shadow and passing cloud.
Clips persist for review in the dashboard (they aren't consumed on first download like regular files). View them under the Clips tab or query with ?kind=clip.
img.difference(other) mutates the image in place and returns itself, so the obvious prev = img loop silently stores the difference as its reference frame. And image statistics are reported on 0–100 for colour (the LAB L channel), not 0–255, so a threshold copied from a grayscale example will never fire. motion() handles both.
File Transfer from In-Memory Data
Send raw bytes without writing to flash first, which is useful for sensor dumps or generated reports:
from tendrl import Client
client = Client(debug=True)
client.start()
# Build a report in memory and send it
report = json.dumps({"readings": [25.5, 26.0, 24.8], "avg": 25.43})
client.send_file(
data=report.encode(),
filename="daily_report.json",
dest="data-collector",
)
Local Database Operations
Store and query data locally on the device:
from tendrl import Client
client = Client(client_db=True, client_db_in_memory=False)
client.start()
# Store a reading
key = client.db_put(
{"temperature": 25.5, "timestamp": "2025-01-15T10:00:00Z"},
tags=["temp"]
)
# Retrieve by key
reading = client.db_get(key)
print(f"Reading: {reading}")
# Query readings
results = client.db_query({"sensor": "temperature"})
print(f"Found {len(results)} readings")
# Delete
client.db_delete(key)
Set client_db_in_memory=False for persistence across reboots (uses flash storage). Flash-backed storage is a best-effort buffer, not a durable store, and writing to it in a tight loop wears the flash - see durability before using it for anything you cannot afford to lose.
Offline Resilience
Messages are automatically stored when the network is down:
from tendrl import Client
client = Client(
offline_storage=True,
debug=True
)
client.start()
# This message is stored locally if WiFi is down,
# and automatically sent when connectivity returns
client.publish(
{"critical_reading": 42.0},
tags=["critical"],
write_offline=True
)
Ethernet Connection
For boards with Ethernet support:
client = Client(
net="eth", # Use Ethernet instead of WiFi
debug=True
)
client.start()
No WiFi credentials needed; it uses DHCP for IP assignment.
Memory Monitoring
Check available memory (useful for debugging on constrained devices):
from tendrl import free
# Get memory and disk stats
stats = free()
print(f"RAM Free: {stats['mem_free']} bytes")
print(f"RAM Total: {stats['mem_total']} bytes")
print(f"Flash Free: {stats['disk_free']} bytes")
Watching memory while a script runs
free() is a one-shot check. To watch usage over time while your code runs, and have it drive the live metrics rail in the Device tab, call emit_metrics() from your loop:
from tendrl import emit_metrics
while True:
# ...your work...
emit_metrics(refresh_disk=False) # heap moves; disk rarely does
time.sleep(1)
The rail shows used/total plus a peak marker, so you can see both what you're using now and the worst spike since connecting. A flat line is healthy; a steady climb across samples is a leak.