IoT Platform

Architecture

The platform is split into three layers. Only app.py changes between device types. Everything below it is identical across all devices and OTA-deployed as a unit. boot.py is the sole exception — it is flashed once and never overwritten remotely.

Execution order

boot.py runs first and exits before main.py is ever imported. OTA updates are therefore always applied atomically before any application code runs.

# boot sequence (every power-on or reset)
boot.py
  → rotate logs
  → connect WiFi  (or start AP mode if no credentials)
  → POST /iot/boot  (sends MD5 hashes of all .py files on /)
  → if update: write files atomically → reboot
  → if no update: exit → main.py runs

main.py
  → asyncio.gather(
      platform.start(app),   # register device, web server, heartbeat loop
      app.run()              # device sensor / actuator loop
    )

Telemetry flow

The app never writes to the web server or heartbeat directly. It pushes key/value pairs through platform.report(), which forwards to both GET /status and each outgoing heartbeat payload.

# app.py (sensor loop)
platform.report(
    distance_mm   = dist,
    sensor_status = "ok",
    uptime_s      = uptime,
)

# → web_server._telemetry updated
# → merged into GET /status response
# → included in next POST /iot/heartbeat/{device_id}

boot.py — Firmware & OTA Sync

Runs on every boot before main.py. Responsible for WiFi connectivity, OTA file synchronisation via hash comparison, device identity assignment, and clock sync from the server. This file is never included in the OTA directory and must be flashed manually.

Boot sequence

API request / response

Files managed by boot.py

platform.py — Runtime Contract

The public API that app.py is written against. Manages device registration, telemetry forwarding, the heartbeat loop, and HTTP route registration. Identical across all device types — never edited per-device.

What the platform gives app.py

What app.py must export

Minimal app.py skeleton

# app.py — skeleton for a new device type
import asyncio
import platform

DEVICE_TYPE = "pico"
FIRMWARE    = "1.0.0"
META        = {
    "capabilities": ["sensor"],
    "sensor":       "my_sensor",
}

async def get_reading(request):
    value = platform._telemetry.get("value", 0)
    return {"value": value}

platform.register_routes([
    ("GET", "/reading", get_reading),
])

async def run():
    while True:
        value = read_sensor()
        platform.report(value=value, status="ok")
        await asyncio.sleep_ms(100)

NeoPixel / identify support

If the device has a NeoPixel strip, report num_leds and np_pin via platform.report() once at module level. The platform's POST /identify endpoint will then work with no additional code in app.py.

platform.report(num_leds=10, np_pin=0)   # call once at module level

Internal platform tasks

These run concurrently in the asyncio event loop once platform.start(app) is called. The app's own run() coroutine is gathered alongside them by main.py.

web_server.py — HTTP Layer

Microdot-based HTTP server. Owns all platform endpoints and exposes the Microdot app object so platform.register_routes() can attach device-specific handlers. These endpoints are present on every device and cannot be removed or overridden by app.py.

Internal setters (called by platform.py, not app.py)

iot_registry.py — Registry & Heartbeat

Handles device registration with the server-side IoT registry after main.py starts, and sends periodic heartbeats. A heartbeat response may include OTA files, which are applied immediately without waiting for the next boot.

Public functions

API endpoints used