Samples

"""02 - Publish exactly one signed temperature reading.

Adds the two things 01 did not do: build a telemetry body, and sign it
into an LM1 frame before publishing.

Run:
    python 02_publish_temperature.py

Expected result: the reading appears in IoT Manager within a couple of
seconds. There is no per-message ACK on MQTT - the QoS 1 PUBACK only
means the broker took the message, not that the backend accepted it.
Check the dashboard (or sample 07's command round trip) to confirm.
"""

from __future__ import annotations

import json
import sys
import threading
import time
from datetime import UTC, datetime

import paho.mqtt.client as mqtt
from common.lm1 import SequenceCounter, load_config, sign_frame


def main() -> int:
    config = load_config()
    topic = config.topics["telemetry"]
    sequence = SequenceCounter()
    connected = threading.Event()

    def on_connect(
        client: mqtt.Client, userdata: object, flags: object, reason_code: object, properties: object = None
    ) -> None:
        connected.set()

    client = mqtt.Client(mqtt.CallbackAPIVersion.VERSION2, client_id=config.username, clean_session=True)
    client.username_pw_set(config.username, config.password)
    client.tls_set()
    client.on_connect = on_connect
    client.connect(config.host, config.port, keepalive=30)
    client.loop_start()
    try:
        if not connected.wait(timeout=15):
            print("timed out waiting for CONNACK", file=sys.stderr)
            return 1

        body = json.dumps({
            "timestamp": datetime.now(UTC).isoformat().replace("+00:00", "Z"),
            # Strictly increasing per device, seeded from wall-clock time -
            # see SequenceCounter's docstring for why 0/1 is a trap.
            "sequence": sequence.next(),
            "values": {"temperature": {"value": 24.7, "unit": "C", "quality": "valid"}},
        }).encode()

        # Sign the EXACT bytes being published - never re-serialize the body.
        frame = sign_frame(topic, body, config.password)
        info = client.publish(topic, frame, qos=1)
        info.wait_for_publish(timeout=10)
        print(f"published {len(frame)} bytes to {topic} (rc={info.rc})")

        time.sleep(1)  # let paho flush the PUBACK exchange before disconnecting
        return 0
    finally:
        client.loop_stop()
        client.disconnect()


if __name__ == "__main__":
    raise SystemExit(main())

// 02 - Publish exactly one signed temperature reading.
//
// Adds the two things sketch 01 did not do: build a telemetry body, and
// sign it into an LM1 frame before publishing.
//
// Two gotchas this sketch handles explicitly:
//   * PubSubClient's buffer defaults to 256 bytes and it SILENTLY drops
//     anything larger. A signed frame carries an 80-byte prefix on top of
//     the body, so raise it with setBufferSize().
//   * `sequence` is seeded from wall-clock time, never from 0 or 1 - see
//     Lm1Sequence in lm1.h for why.
//
// Libraries: PubSubClient, ArduinoJson (Library Manager).
//
// Expected serial output: "published <n> bytes" and the reading showing
// up in IoT Manager within a couple of seconds. There is no per-message
// ACK on MQTT: the QoS 1 PUBACK only means the broker took the frame.

#include <WiFi.h>
#include <WiFiClientSecure.h>
#include <PubSubClient.h>
#include <ArduinoJson.h>

#include "lm1.h"

WiFiClientSecure tlsClient;
PubSubClient mqttClient(tlsClient);
Lm1Sequence sequence;

bool publishTemperature(float celsius) {
  JsonDocument doc;
  doc["sequence"] = sequence.next();
  JsonObject value = doc["values"]["temperature"].to<JsonObject>();
  value["value"] = celsius;
  value["unit"] = "C";
  value["quality"] = "valid";

  String body;
  serializeJson(doc, body);

  String topic = lm1TelemetryTopic();
  String frame = lm1SignFrame(topic, body, DEVICE_PASSWORD);
  if (frame.length() > LM1_MAX_FRAME_BYTES) {
    Serial.println("frame too large - the backend drops anything over 64 KB");
    return false;
  }

  // retained = false. Retained messages are disabled platform-wide.
  bool ok = mqttClient.publish(topic.c_str(), (const uint8_t *)frame.c_str(), frame.length(), false);
  Serial.print(ok ? "published " : "publish FAILED at ");
  Serial.print(frame.length());
  Serial.println(" bytes");
  return ok;
}

void setup() {
  Serial.begin(115200);
  delay(500);

  WiFi.mode(WIFI_STA);
  WiFi.begin(WIFI_SSID, WIFI_PASSWORD);
  while (WiFi.status() != WL_CONNECTED) {
    delay(250);
  }
  if (!lm1WaitForClock()) {
    Serial.println("NTP sync failed - refusing to sign with a wrong clock");
    return;
  }
  sequence.begin();  // seed AFTER the clock is real

  tlsClient.setInsecure();  // bring-up only; pin a CA certificate in production
  mqttClient.setServer(MQTT_HOST, MQTT_PORT);
  mqttClient.setKeepAlive(30);
  // Default is 256 bytes and oversized publishes fail silently.
  mqttClient.setBufferSize(1024);

  if (!mqttClient.connect(DEVICE_ID, DEVICE_ID, DEVICE_PASSWORD)) {
    Serial.print("mqtt FAILED, state=");
    Serial.println(mqttClient.state());
    return;
  }
  Serial.println("mqtt ok");
  publishTemperature(24.7f);
}

void loop() {
  mqttClient.loop();
  delay(10);
}

"""04 - Every supported value type, in one measurement.

A single telemetry message carries a `values` map with one entry per
signal. Each entry is `{"value": ..., "unit": "...", "quality": "..."}`.

Supported value types (signals/telemetry.py's MeasurementValue):

    float   24.7        a normal analog reading
    int     1013        an integer counter or level
    bool    true        a door, a relay, any on/off state
    str     "heating"   a mode/status string

`unit` is a free-form string (default ""), `quality` is one of
`valid` (default), `stale`, `uncertain`, `error`.

Limits that apply here: at most 50 keys in one `values` map, and a
per-plan ceiling on how many DISTINCT signal keys one device may ever
use (20 on every plan today).

Run:
    python 04_datatypes_multi_datastream.py

Expected result: all six signals update together in IoT Manager - one
message, one timestamp, one sequence number.
"""

from __future__ import annotations

import json
import sys
import threading
import time
from datetime import UTC, datetime

import paho.mqtt.client as mqtt
from common.lm1 import SequenceCounter, load_config, sign_frame


def read_sensors() -> dict[str, object]:
    """Replace with your real reads. Values only - typing is per key below."""
    return {
        "temperature": 24.7,
        "humidity": 61.2,
        "pressure": 1013,
        "door_open": False,
        "mode": "heating",
        "battery": 87,
    }


def main() -> int:
    config = load_config()
    topic = config.topics["telemetry"]
    sequence = SequenceCounter()
    connected = threading.Event()

    def on_connect(
        client: mqtt.Client, userdata: object, flags: object, reason_code: object, properties: object = None
    ) -> None:
        connected.set()

    client = mqtt.Client(mqtt.CallbackAPIVersion.VERSION2, client_id=config.username, clean_session=True)
    client.username_pw_set(config.username, config.password)
    client.tls_set()
    client.on_connect = on_connect
    client.connect(config.host, config.port, keepalive=30)
    client.loop_start()
    try:
        if not connected.wait(timeout=15):
            print("timed out waiting for CONNACK", file=sys.stderr)
            return 1

        readings = read_sensors()
        body = json.dumps({
            # Device-observed time, UTC, ISO-8601. The server records its own
            # received-at separately, so buffered data still lands correctly.
            "timestamp": datetime.now(UTC).isoformat().replace("+00:00", "Z"),
            "sequence": sequence.next(),
            "values": {
                # float + unit
                "temperature": {"value": readings["temperature"], "unit": "C", "quality": "valid"},
                # float, flagged as not fully trustworthy
                "humidity": {"value": readings["humidity"], "unit": "%", "quality": "uncertain"},
                # int + unit
                "pressure": {"value": readings["pressure"], "unit": "hPa"},
                # bool - no unit makes sense
                "door_open": {"value": readings["door_open"]},
                # str - an enum-like mode
                "mode": {"value": readings["mode"]},
                # int as a percentage
                "battery": {"value": readings["battery"], "unit": "%"},
            },
            # Free-form; useful for firmware version, site id, and the like.
            "metadata": {"firmware": "1.4.2"},
        }).encode()

        info = client.publish(topic, sign_frame(topic, body, config.password), qos=1)
        info.wait_for_publish(timeout=10)
        print(f"published 6 signals ({len(body)} body bytes, rc={info.rc})")
        time.sleep(1)
        return 0
    finally:
        client.loop_stop()
        client.disconnect()


if __name__ == "__main__":
    raise SystemExit(main())

// 03 - Several signals, on a fixed interval, in one message.
//
// One telemetry message carries a `values` map with one entry per
// signal, so a multi-sensor board sends ONE frame per tick, not one per
// sensor. Supported value types: float, int, bool and string. `unit` is
// free-form; `quality` is valid / stale / uncertain / error.
//
// Pacing: the backend enforces a per-plan minimum interval between
// accepted measurements (60 s free, 20 s premium1, 10 s premium2 today -
// those numbers live in the plans table and an admin can change them).
// Anything faster is silently dropped. SEND_INTERVAL_MS defaults to 60 s,
// which is safe on every plan.
//
// Note the millis() rollover: the elapsed-time check below is an unsigned
// subtraction, which stays correct across the ~49-day wrap. A
// `millis() > deadline` comparison would not.
//
// Libraries: PubSubClient, ArduinoJson.
//
// Expected serial output: one "sent seq=..." line per interval, and five
// charts moving together in IoT Manager.

#include <WiFi.h>
#include <WiFiClientSecure.h>
#include <PubSubClient.h>
#include <ArduinoJson.h>

#include "lm1.h"

#ifndef SEND_INTERVAL_MS
#define SEND_INTERVAL_MS 60000UL
#endif

WiFiClientSecure tlsClient;
PubSubClient mqttClient(tlsClient);
Lm1Sequence sequence;
unsigned long lastSendMs = 0;

void ensureConnected() {
  if (WiFi.status() != WL_CONNECTED) {
    WiFi.begin(WIFI_SSID, WIFI_PASSWORD);
    while (WiFi.status() != WL_CONNECTED) {
      delay(250);
    }
  }
  if (!mqttClient.connected()) {
    mqttClient.connect(DEVICE_ID, DEVICE_ID, DEVICE_PASSWORD);
  }
}

void sendReadings() {
  JsonDocument doc;
  unsigned long seq = sequence.next();
  doc["sequence"] = seq;
  JsonObject values = doc["values"].to<JsonObject>();

  // float + unit
  values["temperature"]["value"] = 24.7f;
  values["temperature"]["unit"] = "C";
  // float, flagged as not fully trustworthy
  values["humidity"]["value"] = 61.2f;
  values["humidity"]["unit"] = "%";
  values["humidity"]["quality"] = "uncertain";
  // int + unit
  values["pressure"]["value"] = 1013;
  values["pressure"]["unit"] = "hPa";
  // bool - no unit makes sense
  values["door_open"]["value"] = false;
  // string - an enum-like mode
  values["mode"]["value"] = "heating";

  String body;
  serializeJson(doc, body);

  String topic = lm1TelemetryTopic();
  String frame = lm1SignFrame(topic, body, DEVICE_PASSWORD);
  bool ok = mqttClient.publish(topic.c_str(), (const uint8_t *)frame.c_str(), frame.length(), false);
  Serial.print(ok ? "sent seq=" : "publish FAILED seq=");
  Serial.println(seq);
}

void setup() {
  Serial.begin(115200);
  delay(500);

  WiFi.mode(WIFI_STA);
  WiFi.begin(WIFI_SSID, WIFI_PASSWORD);
  while (WiFi.status() != WL_CONNECTED) {
    delay(250);
  }
  if (!lm1WaitForClock()) {
    Serial.println("NTP sync failed - refusing to sign with a wrong clock");
    return;
  }
  sequence.begin();  // seed AFTER the clock is real

  tlsClient.setInsecure();  // bring-up only; pin a CA certificate in production
  mqttClient.setServer(MQTT_HOST, MQTT_PORT);
  mqttClient.setKeepAlive(30);
  mqttClient.setBufferSize(1024);  // default 256 - oversized publishes fail silently
  mqttClient.connect(DEVICE_ID, DEVICE_ID, DEVICE_PASSWORD);
  Serial.println("ready");
}

void loop() {
  ensureConnected();
  mqttClient.loop();

  if (millis() - lastSendMs >= SEND_INTERVAL_MS) {
    lastSendMs = millis();
    sendReadings();
  }
  delay(10);
}

"""06 - Survive a network outage: reconnect, then flush a local buffer.

Two independent mechanisms:

1. Reconnection. paho does it for you once `loop_start()` is running -
   `reconnect_delay_set(min_delay, max_delay)` gives you the exponential
   backoff. Do NOT hand-roll a tight reconnect loop. There is no re-auth
   step to redo: MQTT authenticates in CONNECT itself.

2. Offline buffering. While disconnected, keep readings in a BOUNDED
   local buffer (a ring buffer - memory is finite on a microcontroller,
   and a buffer that grows without limit turns a network outage into a
   crash). On reconnect, publish the backlog as batches.

The backlog is accepted even though it all arrives at once: the backend
paces rate limits on each frame's own signed timestamp, not on arrival
time. Sign each flushed batch as you publish it (so `ts` is current) and
keep each message's own `timestamp` field as the moment it was OBSERVED.

Run:
    python 06_reconnect_offline_buffer.py
    # then pull the network cable / disable Wi-Fi for a minute

Expected result: "buffered ..." lines while offline, one "flushed ..."
line per batch on reconnect, and no gap in the IoT Manager chart.
"""

from __future__ import annotations

import json
import os
import threading
import time
from collections import deque
from datetime import UTC, datetime

import paho.mqtt.client as mqtt
from common.lm1 import SequenceCounter, load_config, sign_frame

SEND_INTERVAL_S = float(os.getenv("SEND_INTERVAL_S", "60"))
BUFFER_CAPACITY = 500  # oldest readings are dropped once full
MAX_MESSAGES_PER_BATCH = 100


class BufferedDevice:
    """Owns the paho client, the buffer and the sequence counter.

    The paho callbacks run on paho's own network thread, so the buffer is
    guarded by a lock - the publish loop below runs on the main thread.
    """

    def __init__(self) -> None:
        self._config = load_config()
        self._topic = self._config.topics["telemetry"]
        self._sequence = SequenceCounter()
        self._buffer: deque[dict[str, object]] = deque(maxlen=BUFFER_CAPACITY)
        self._lock = threading.Lock()
        self._online = threading.Event()
        self._client = self._build_client()

    def _build_client(self) -> mqtt.Client:
        client = mqtt.Client(
            mqtt.CallbackAPIVersion.VERSION2, client_id=self._config.username, clean_session=True
        )
        client.username_pw_set(self._config.username, self._config.password)
        client.tls_set()
        client.reconnect_delay_set(min_delay=2, max_delay=30)
        client.on_connect = self._on_connect
        client.on_disconnect = self._on_disconnect
        return client

    def _on_connect(
        self, client: mqtt.Client, userdata: object, flags: object, reason_code: object, properties: object = None
    ) -> None:
        print(f"connected (reason_code={reason_code})")
        self._online.set()

    def _on_disconnect(
        self, client: mqtt.Client, userdata: object, flags: object, reason_code: object, properties: object = None
    ) -> None:
        print(f"disconnected (reason_code={reason_code}) - paho will retry with backoff")
        self._online.clear()

    def _reading(self) -> dict[str, object]:
        return {
            "timestamp": datetime.now(UTC).isoformat().replace("+00:00", "Z"),
            "sequence": self._sequence.next(),
            "values": {"temperature": {"value": 24.7, "unit": "C"}},
        }

    def _flush(self) -> None:
        """Publish the backlog, oldest first, in batches."""
        while True:
            with self._lock:
                if not self._buffer:
                    return
                batch = [self._buffer.popleft() for _ in range(min(MAX_MESSAGES_PER_BATCH, len(self._buffer)))]
            body = json.dumps({"messages": batch}).encode()
            info = self._client.publish(self._topic, sign_frame(self._topic, body, self._config.password), qos=1)
            info.wait_for_publish(timeout=10)
            print(f"flushed {len(batch)} buffered reading(s)")

    def run(self) -> None:
        self._client.connect(self._config.host, self._config.port, keepalive=30)
        self._client.loop_start()
        try:
            while True:
                reading = self._reading()
                if self._online.is_set():
                    self._flush()
                    body = json.dumps(reading).encode()
                    self._client.publish(self._topic, sign_frame(self._topic, body, self._config.password), qos=1)
                    print(f"{datetime.now():%H:%M:%S} sent seq={reading['sequence']}")
                else:
                    with self._lock:
                        self._buffer.append(reading)
                    print(f"{datetime.now():%H:%M:%S} buffered seq={reading['sequence']} ({len(self._buffer)} held)")
                time.sleep(SEND_INTERVAL_S)
        except KeyboardInterrupt:
            print("\nstopped")
        finally:
            self._client.loop_stop()
            self._client.disconnect()


if __name__ == "__main__":
    BufferedDevice().run()

// 04 - Survive a network outage: backoff reconnect + a local buffer.
//
// PubSubClient does NOT reconnect on its own, so the reconnect policy is
// yours to write. Two rules:
//   * Never a tight reconnect loop. Start at ~2 s and double up to a cap
//     (30 s here). A board that hammers a down broker just burns power
//     and gets throttled.
//   * There is no re-auth step. MQTT authenticates inside CONNECT, so a
//     successful reconnect is fully authenticated again - unlike the old
//     WebSocket transport, there is nothing to resend afterwards.
//
// While offline, readings go into a BOUNDED ring buffer. Unbounded
// buffering turns a network outage into an out-of-memory crash; dropping
// the oldest reading is the right trade on a microcontroller.
//
// The backlog is accepted even though it arrives all at once: the backend
// paces rate limits on each frame's own signed timestamp, not on arrival
// time. Each buffered reading keeps its own device-side observation time,
// and the frame is signed fresh at flush time.
//
// Batch limits when flushing: at most 100 messages per batch, at most 50
// keys per values map, and the whole signed frame must stay under 64 KB.
// Sequence numbers must strictly increase inside the batch too, which
// they do here because they are assigned when the reading is captured.
//
// Libraries: PubSubClient, ArduinoJson.
//
// Expected serial output: "buffered (n held)" while the network is down,
// then "flushed n" on reconnect, and no gap in the IoT Manager chart.

#include <WiFi.h>
#include <WiFiClientSecure.h>
#include <PubSubClient.h>
#include <ArduinoJson.h>

#include "lm1.h"

#ifndef SEND_INTERVAL_MS
#define SEND_INTERVAL_MS 60000UL
#endif

static const size_t BUFFER_CAPACITY = 40;   // ring buffer slots
static const size_t MAX_BATCH_MESSAGES = 100;
static const unsigned long RECONNECT_MIN_MS = 2000;
static const unsigned long RECONNECT_MAX_MS = 30000;

struct Reading {
  unsigned long sequence;
  unsigned long observedAt;  // Unix seconds
  float temperature;
};

WiFiClientSecure tlsClient;
PubSubClient mqttClient(tlsClient);
Lm1Sequence sequence;

Reading buffer[BUFFER_CAPACITY];
size_t bufferHead = 0;   // next slot to write
size_t bufferCount = 0;  // how many slots hold data
unsigned long lastSendMs = 0;
unsigned long reconnectDelayMs = RECONNECT_MIN_MS;
unsigned long nextReconnectMs = 0;

void bufferPush(const Reading &reading) {
  buffer[bufferHead] = reading;
  bufferHead = (bufferHead + 1) % BUFFER_CAPACITY;
  if (bufferCount < BUFFER_CAPACITY) {
    bufferCount++;
  }
}

// Index of the i-th oldest buffered reading.
size_t bufferOldestIndex(size_t i) {
  return (bufferHead + BUFFER_CAPACITY - bufferCount + i) % BUFFER_CAPACITY;
}

String isoTime(unsigned long unixSeconds) {
  time_t t = (time_t)unixSeconds;
  struct tm tmUtc;
  gmtime_r(&t, &tmUtc);
  char out[24];
  strftime(out, sizeof(out), "%Y-%m-%dT%H:%M:%SZ", &tmUtc);
  return String(out);
}

Reading captureReading() {
  Reading reading;
  reading.sequence = sequence.next();
  reading.observedAt = (unsigned long)time(nullptr);
  reading.temperature = 24.7f;  // replace with a real sensor read
  return reading;
}

bool publishBatch(size_t count) {
  JsonDocument doc;
  JsonArray messages = doc["messages"].to<JsonArray>();
  for (size_t i = 0; i < count; i++) {
    const Reading &reading = buffer[bufferOldestIndex(i)];
    JsonObject message = messages.add<JsonObject>();
    message["timestamp"] = isoTime(reading.observedAt);
    message["sequence"] = reading.sequence;
    JsonObject value = message["values"]["temperature"].to<JsonObject>();
    value["value"] = reading.temperature;
    value["unit"] = "C";
  }

  String body;
  serializeJson(doc, body);

  String topic = lm1TelemetryTopic();
  String frame = lm1SignFrame(topic, body, DEVICE_PASSWORD);
  if (frame.length() > LM1_MAX_FRAME_BYTES) {
    Serial.println("batch over 64 KB - flush fewer messages at a time");
    return false;
  }
  return mqttClient.publish(topic.c_str(), (const uint8_t *)frame.c_str(), frame.length(), false);
}

void flushBuffer() {
  while (bufferCount > 0) {
    size_t count = bufferCount < MAX_BATCH_MESSAGES ? bufferCount : MAX_BATCH_MESSAGES;
    if (!publishBatch(count)) {
      return;  // keep the backlog; retry on the next tick
    }
    bufferCount -= count;
    Serial.print("flushed ");
    Serial.println(count);
  }
}

// Non-blocking reconnect with exponential backoff.
void serviceConnection() {
  if (mqttClient.connected()) {
    reconnectDelayMs = RECONNECT_MIN_MS;
    return;
  }
  if (millis() < nextReconnectMs) {
    return;
  }
  if (WiFi.status() != WL_CONNECTED) {
    WiFi.begin(WIFI_SSID, WIFI_PASSWORD);
  }
  if (mqttClient.connect(DEVICE_ID, DEVICE_ID, DEVICE_PASSWORD)) {
    Serial.println("reconnected");
    reconnectDelayMs = RECONNECT_MIN_MS;
    flushBuffer();
    return;
  }
  Serial.print("reconnect failed, state=");
  Serial.print(mqttClient.state());
  Serial.print(" - retrying in ");
  Serial.println(reconnectDelayMs);
  nextReconnectMs = millis() + reconnectDelayMs;
  reconnectDelayMs = reconnectDelayMs * 2 > RECONNECT_MAX_MS ? RECONNECT_MAX_MS : reconnectDelayMs * 2;
}

void setup() {
  Serial.begin(115200);
  delay(500);

  WiFi.mode(WIFI_STA);
  WiFi.begin(WIFI_SSID, WIFI_PASSWORD);
  while (WiFi.status() != WL_CONNECTED) {
    delay(250);
  }
  if (!lm1WaitForClock()) {
    Serial.println("NTP sync failed - refusing to sign with a wrong clock");
    return;
  }
  sequence.begin();

  tlsClient.setInsecure();  // bring-up only; pin a CA certificate in production
  mqttClient.setServer(MQTT_HOST, MQTT_PORT);
  mqttClient.setKeepAlive(30);
  mqttClient.setBufferSize(4096);  // batches are much larger than one reading
  mqttClient.connect(DEVICE_ID, DEVICE_ID, DEVICE_PASSWORD);
  Serial.println("ready");
}

void loop() {
  serviceConnection();
  mqttClient.loop();

  if (millis() - lastSendMs >= SEND_INTERVAL_MS) {
    lastSendMs = millis();
    Reading reading = captureReading();
    bufferPush(reading);
    if (mqttClient.connected()) {
      flushBuffer();
    } else {
      Serial.print("buffered (");
      Serial.print(bufferCount);
      Serial.println(" held)");
    }
  }
  delay(10);
}

"""07 - Receive commands and acknowledge them.

Subscribe to your own `cmd` topic once, at connect time. Every command
the backend publishes there is LM1-signed with your own connection
password, exactly like your outbound frames - verify it before acting on
it, then reply on `cmd/ack` with the same `command_id`.

Command envelope (already verified, JSON-decoded):

    {"type": "COMMAND", "command_id": "...", "command": "relay1",
     "kind": "boolean", "value": true}

* `command` is the target datastream's KEY, never a verb.
* `kind` is derived server-side from that datastream and is one of
  `action`, `boolean`, `number`, `string`, `enum`.
* `kind: "action"` never carries a `value`; every other kind always does.

Ack body:

    {"command_id": "...", "status": "ok", "value": true}
    {"command_id": "...", "status": "error", "error": "actuator_fault"}

`status` must be exactly `"ok"` for success - anything else is read as a
device-reported failure. A command with no ack inside the server-side
timeout is marked timed out.

Run:
    python 07_receive_commands_ack.py
    # then send a command to this device from the IoT Manager dashboard

Expected result: one "COMMAND ..." line per command and one ack per
command. The subscription lasts the whole connection; no re-subscribing.
"""

from __future__ import annotations

import json

import paho.mqtt.client as mqtt
from common.lm1 import SignatureError, load_config, sign_frame, verify_frame

CONFIG = load_config()
TOPICS = CONFIG.topics


def on_connect(
    client: mqtt.Client, userdata: object, flags: object, reason_code: object, properties: object = None
) -> None:
    # Subscribing from on_connect (not after connect()) means an automatic
    # reconnect re-subscribes too - a clean session keeps no subscriptions.
    client.subscribe(TOPICS["cmd"], qos=1)
    print(f"connected, subscribed to {TOPICS['cmd']}")


def on_message(client: mqtt.Client, userdata: object, message: mqtt.MQTTMessage) -> None:
    # paho swallows exceptions raised inside a callback, so an unhandled
    # bug in here would just look like "nothing happens". Catch broadly.
    try:
        handle_command(client, message.topic, message.payload)
    except Exception as exc:  # a device must not die on one malformed frame
        print(f"on_message failed: {exc!r}")


def handle_command(client: mqtt.Client, topic: str, payload: bytes) -> None:
    try:
        body = verify_frame(topic, payload, CONFIG.password)
    except SignatureError as exc:
        print(f"rejected an unverifiable frame on {topic}: {exc}")
        return

    command = json.loads(body)
    if not isinstance(command, dict) or command.get("type") != "COMMAND":
        return

    command_id = command.get("command_id", "")
    key = command.get("command", "")
    kind = command.get("kind", "")
    value = command.get("value")
    print(f"COMMAND {key} kind={kind} value={value!r} (id={command_id})")

    if key == "relay1" and kind == "boolean":
        # Drive the real hardware here; report what actually happened.
        result: dict[str, object] = {"command_id": command_id, "status": "ok", "value": value}
    elif key == "reboot" and kind == "action":
        result = {"command_id": command_id, "status": "ok"}
    else:
        result = {"command_id": command_id, "status": "error", "error": "unsupported_command"}

    ack_topic = TOPICS["cmdAck"]
    ack_body = json.dumps(result).encode()
    client.publish(ack_topic, sign_frame(ack_topic, ack_body, CONFIG.password), qos=1)
    print(f"acked {command_id} -> {result['status']}")


def main() -> int:
    client = mqtt.Client(mqtt.CallbackAPIVersion.VERSION2, client_id=CONFIG.username, clean_session=True)
    client.username_pw_set(CONFIG.username, CONFIG.password)
    client.tls_set()
    client.reconnect_delay_set(min_delay=2, max_delay=30)
    client.on_connect = on_connect
    client.on_message = on_message
    client.connect(CONFIG.host, CONFIG.port, keepalive=30)
    try:
        client.loop_forever()
    except KeyboardInterrupt:
        print("\nstopped")
    finally:
        client.disconnect()
    return 0


if __name__ == "__main__":
    raise SystemExit(main())

// 05 - Drive a relay from the dashboard: commands in, state reported back.
//
// Subscribe to your own `cmd` topic from the moment you connect. Every
// command the backend publishes there is LM1-signed with your own
// connection password, exactly like your outbound frames - verify it
// before touching hardware.
//
// Command envelope (after verification):
//   {"type":"COMMAND","command_id":"...","command":"relay1",
//    "kind":"boolean","value":true}
//   * `command` is the target datastream's KEY, never a verb.
//   * `kind` is derived server-side: action / boolean / number / string /
//     enum. `kind:"action"` never carries a value; every other kind does.
//
// Ack on `cmd/ack`, signed, with the same command_id:
//   {"command_id":"...","status":"ok","value":true}
//   {"command_id":"...","status":"error","error":"actuator_fault"}
// `status` must be exactly "ok" for success. No ack inside the server
// timeout means the command is marked timed out.
//
// Acking is not reporting: the dashboard's *reported* state only moves
// when the device publishes the new value as telemetry, so this sketch
// publishes `relay1` right after applying a command, and again
// periodically so a dashboard opened later still sees the truth.
//
// Prerequisites in IoT Manager: a datastream keyed `relay1`, kind
// actuator, data type boolean, writable. Sending device commands
// requires a premium plan.
//
// Libraries: PubSubClient, ArduinoJson.

#include <WiFi.h>
#include <WiFiClientSecure.h>
#include <PubSubClient.h>
#include <ArduinoJson.h>

#include "lm1.h"

#ifndef RELAY_PIN
#define RELAY_PIN 26
#endif
#ifndef SEND_INTERVAL_MS
#define SEND_INTERVAL_MS 60000UL
#endif

static const char *RELAY_SIGNAL = "relay1";

WiFiClientSecure tlsClient;
PubSubClient mqttClient(tlsClient);
Lm1Sequence sequence;
bool relayOn = false;
unsigned long lastReportMs = 0;

void publishAck(const String &commandId, const char *status, bool value, const char *error) {
  JsonDocument doc;
  doc["command_id"] = commandId;
  doc["status"] = status;
  if (strcmp(status, "ok") == 0) {
    doc["value"] = value;
  } else {
    doc["error"] = error;
  }

  String body;
  serializeJson(doc, body);
  String topic = lm1CommandAckTopic();
  String frame = lm1SignFrame(topic, body, DEVICE_PASSWORD);
  mqttClient.publish(topic.c_str(), (const uint8_t *)frame.c_str(), frame.length(), false);
}

void reportRelayState() {
  JsonDocument doc;
  doc["sequence"] = sequence.next();
  doc["values"][RELAY_SIGNAL]["value"] = relayOn;

  String body;
  serializeJson(doc, body);
  String topic = lm1TelemetryTopic();
  String frame = lm1SignFrame(topic, body, DEVICE_PASSWORD);
  mqttClient.publish(topic.c_str(), (const uint8_t *)frame.c_str(), frame.length(), false);
}

void onCommand(char *topic, byte *payload, unsigned int length) {
  String frame;
  frame.reserve(length);
  for (unsigned int i = 0; i < length; i++) {
    frame += (char)payload[i];
  }

  String body;
  if (!lm1VerifyFrame(String(topic), frame, DEVICE_PASSWORD, body)) {
    Serial.println("rejected an unverifiable frame");
    return;
  }

  JsonDocument doc;
  if (deserializeJson(doc, body)) {
    return;
  }
  if (String(doc["type"] | "") != "COMMAND") {
    return;
  }

  String commandId = doc["command_id"] | "";
  if (String(doc["command"] | "") != RELAY_SIGNAL || String(doc["kind"] | "") != "boolean") {
    publishAck(commandId, "error", false, "unsupported_command");
    return;
  }

  relayOn = doc["value"] | false;
  digitalWrite(RELAY_PIN, relayOn ? HIGH : LOW);
  Serial.print("relay -> ");
  Serial.println(relayOn ? "ON" : "OFF");

  publishAck(commandId, "ok", relayOn, "");
  reportRelayState();  // an ack resolves the command; telemetry moves reported state
}

bool connectMqtt() {
  if (!mqttClient.connect(DEVICE_ID, DEVICE_ID, DEVICE_PASSWORD)) {
    return false;
  }
  // Re-subscribe on every connect: a clean session keeps no subscriptions
  // across a reconnect.
  mqttClient.subscribe(lm1CommandTopic().c_str(), 1);
  Serial.println("connected and subscribed");
  return true;
}

void setup() {
  Serial.begin(115200);
  delay(500);
  pinMode(RELAY_PIN, OUTPUT);
  digitalWrite(RELAY_PIN, LOW);

  WiFi.mode(WIFI_STA);
  WiFi.begin(WIFI_SSID, WIFI_PASSWORD);
  while (WiFi.status() != WL_CONNECTED) {
    delay(250);
  }
  if (!lm1WaitForClock()) {
    Serial.println("NTP sync failed - refusing to sign with a wrong clock");
    return;
  }
  sequence.begin();

  tlsClient.setInsecure();  // bring-up only; pin a CA certificate in production
  mqttClient.setServer(MQTT_HOST, MQTT_PORT);
  mqttClient.setKeepAlive(30);
  mqttClient.setBufferSize(1024);
  mqttClient.setCallback(onCommand);
  connectMqtt();
}

void loop() {
  if (!mqttClient.connected()) {
    delay(2000);
    connectMqtt();
    return;
  }
  mqttClient.loop();

  if (millis() - lastReportMs >= SEND_INTERVAL_MS) {
    lastReportMs = millis();
    reportRelayState();
  }
  delay(10);
}

"""09 - A complete device: every piece of the protocol in one program.

Combines samples 01-08:

* TLS CONNECT with username == clientid, clean session, keepalive 30 s
* automatic reconnection with backoff (paho's own), re-subscribing from
  on_connect
* periodic multi-signal telemetry, sent as one signed batch per tick
* a bounded offline buffer, flushed as batches after a reconnect
* an inbound command channel with signature verification, acks, and a
  reported-state publish afterwards
* wall-clock-seeded, strictly increasing sequence numbers everywhere

This is the sample to copy for a real deployment; the earlier ones exist
to isolate one idea each.

Run:
    SEND_INTERVAL_S=60 python 09_full_bidirectional_device.py

Expected result: a telemetry line per interval, the temperature and relay
charts moving in IoT Manager, and dashboard relay toggles landing here
and being reported back.
"""

from __future__ import annotations

import json
import os
import random
import threading
import time
from collections import deque
from datetime import UTC, datetime

import paho.mqtt.client as mqtt
from common.lm1 import SequenceCounter, SignatureError, load_config, sign_frame, verify_frame

SEND_INTERVAL_S = float(os.getenv("SEND_INTERVAL_S", "60"))
RELAY_SIGNAL = "relay1"
BUFFER_CAPACITY = 500
MAX_MESSAGES_PER_BATCH = 100
MAX_FRAME_BYTES = 65536


class Device:
    def __init__(self) -> None:
        self._config = load_config()
        self._topics = self._config.topics
        self._sequence = SequenceCounter()
        self._buffer: deque[dict[str, object]] = deque(maxlen=BUFFER_CAPACITY)
        self._lock = threading.Lock()
        self._online = threading.Event()
        self._temperature = 22.0
        self._relay_on = False
        self._client = self._build_client()

    # -- setup --

    def _build_client(self) -> mqtt.Client:
        client = mqtt.Client(
            mqtt.CallbackAPIVersion.VERSION2, client_id=self._config.username, clean_session=True
        )
        client.username_pw_set(self._config.username, self._config.password)
        client.tls_set()
        client.reconnect_delay_set(min_delay=2, max_delay=30)
        client.on_connect = self._on_connect
        client.on_disconnect = self._on_disconnect
        client.on_message = self._on_message
        return client

    # -- connection lifecycle --

    def _on_connect(
        self, client: mqtt.Client, userdata: object, flags: object, reason_code: object, properties: object = None
    ) -> None:
        client.subscribe(self._topics["cmd"], qos=1)
        self._online.set()
        print(f"connected (reason_code={reason_code})")

    def _on_disconnect(
        self, client: mqtt.Client, userdata: object, flags: object, reason_code: object, properties: object = None
    ) -> None:
        self._online.clear()
        print(f"disconnected (reason_code={reason_code}) - retrying with backoff")

    # -- inbound commands --

    def _on_message(self, client: mqtt.Client, userdata: object, message: mqtt.MQTTMessage) -> None:
        try:
            self._handle_command(message.topic, message.payload)
        except Exception as exc:  # never let one bad frame kill the loop
            print(f"on_message failed: {exc!r}")

    def _handle_command(self, topic: str, payload: bytes) -> None:
        try:
            body = verify_frame(topic, payload, self._config.password)
        except SignatureError as exc:
            print(f"rejected an unverifiable frame: {exc}")
            return

        command = json.loads(body)
        if not isinstance(command, dict) or command.get("type") != "COMMAND":
            return

        command_id = command.get("command_id", "")
        if command.get("command") == RELAY_SIGNAL and command.get("kind") == "boolean":
            with self._lock:
                self._relay_on = bool(command.get("value"))
                state = self._relay_on
            print(f"relay -> {'ON' if state else 'OFF'} (command {command_id})")
            self._ack({"command_id": command_id, "status": "ok", "value": state})
            self._publish([self._measurement({RELAY_SIGNAL: {"value": state}})])
        else:
            self._ack({"command_id": command_id, "status": "error", "error": "unsupported_command"})

    def _ack(self, result: dict[str, object]) -> None:
        topic = self._topics["cmdAck"]
        body = json.dumps(result).encode()
        self._client.publish(topic, sign_frame(topic, body, self._config.password), qos=1)

    # -- outbound telemetry --

    def _measurement(self, values: dict[str, object]) -> dict[str, object]:
        return {
            "timestamp": datetime.now(UTC).isoformat().replace("+00:00", "Z"),
            "sequence": self._sequence.next(),
            "values": values,
        }

    def _read_sensors(self) -> dict[str, object]:
        self._temperature = max(15.0, min(35.0, self._temperature + random.uniform(-0.4, 0.4)))
        with self._lock:
            relay = self._relay_on
        return {
            "temperature": {"value": round(self._temperature, 1), "unit": "C", "quality": "valid"},
            RELAY_SIGNAL: {"value": relay},
        }

    def _publish(self, messages: list[dict[str, object]]) -> bool:
        """Publish one frame. A single message goes bare; more than one is
        wrapped in the batch shape. Returns False if it could not be sent."""
        if not messages:
            return True
        topic = self._topics["telemetry"]
        body = json.dumps(messages[0] if len(messages) == 1 else {"messages": messages}).encode()
        frame = sign_frame(topic, body, self._config.password)
        if len(frame) > MAX_FRAME_BYTES:
            print(f"dropping oversized frame ({len(frame)} bytes > {MAX_FRAME_BYTES})")
            return True
        info = self._client.publish(topic, frame, qos=1)
        return info.rc == mqtt.MQTT_ERR_SUCCESS

    def _flush_buffer(self) -> None:
        while True:
            with self._lock:
                if not self._buffer:
                    return
                count = min(MAX_MESSAGES_PER_BATCH, len(self._buffer))
                batch = [self._buffer.popleft() for _ in range(count)]
            if not self._publish(batch):
                with self._lock:  # put it back, oldest-first, and try later
                    self._buffer.extendleft(reversed(batch))
                return
            print(f"flushed {len(batch)} buffered reading(s)")

    # -- main loop --

    def run(self) -> None:
        self._client.connect(self._config.host, self._config.port, keepalive=30)
        self._client.loop_start()
        try:
            while True:
                reading = self._measurement(self._read_sensors())
                if self._online.is_set():
                    self._flush_buffer()
                    self._publish([reading])
                    print(f"{datetime.now():%H:%M:%S} sent seq={reading['sequence']}")
                else:
                    with self._lock:
                        self._buffer.append(reading)
                    print(f"{datetime.now():%H:%M:%S} buffered ({len(self._buffer)} held)")
                time.sleep(SEND_INTERVAL_S)
        except KeyboardInterrupt:
            print("\nstopped")
        finally:
            self._client.loop_stop()
            self._client.disconnect()


if __name__ == "__main__":
    Device().run()

// 06 - A complete ESP32 device: everything from sketches 01-05 together.
//
//   * TLS CONNECT with clientid == username, keepalive 30 s
//   * non-blocking reconnect with exponential backoff, re-subscribing on
//     every connect
//   * periodic multi-signal telemetry
//   * a bounded ring buffer flushed as batches after an outage
//   * an inbound command channel with signature verification, acks, and a
//     reported-state publish afterwards
//   * wall-clock-seeded, strictly increasing sequence numbers everywhere
//
// This is the sketch to copy for a real deployment; 01-05 exist to
// isolate one idea each.
//
// Libraries: PubSubClient, ArduinoJson.
//
// Expected serial output: a "sent"/"buffered" line per interval, relay
// toggles from the dashboard landing here, and no gap in the charts
// across a Wi-Fi outage.

#include <WiFi.h>
#include <WiFiClientSecure.h>
#include <PubSubClient.h>
#include <ArduinoJson.h>

#include "lm1.h"

#ifndef SEND_INTERVAL_MS
#define SEND_INTERVAL_MS 60000UL
#endif
#ifndef RELAY_PIN
#define RELAY_PIN 26
#endif

static const char *RELAY_SIGNAL = "relay1";
static const size_t BUFFER_CAPACITY = 40;
static const size_t MAX_BATCH_MESSAGES = 100;
static const unsigned long RECONNECT_MIN_MS = 2000;
static const unsigned long RECONNECT_MAX_MS = 30000;

struct Reading {
  unsigned long sequence;
  unsigned long observedAt;
  float temperature;
  bool relay;
};

WiFiClientSecure tlsClient;
PubSubClient mqttClient(tlsClient);
Lm1Sequence sequence;

Reading buffer[BUFFER_CAPACITY];
size_t bufferHead = 0;
size_t bufferCount = 0;
bool relayOn = false;
unsigned long lastSendMs = 0;
unsigned long reconnectDelayMs = RECONNECT_MIN_MS;
unsigned long nextReconnectMs = 0;

// ---------------------------------------------------------------- utils

String isoTime(unsigned long unixSeconds) {
  time_t t = (time_t)unixSeconds;
  struct tm tmUtc;
  gmtime_r(&t, &tmUtc);
  char out[24];
  strftime(out, sizeof(out), "%Y-%m-%dT%H:%M:%SZ", &tmUtc);
  return String(out);
}

void bufferPush(const Reading &reading) {
  buffer[bufferHead] = reading;
  bufferHead = (bufferHead + 1) % BUFFER_CAPACITY;
  if (bufferCount < BUFFER_CAPACITY) {
    bufferCount++;  // full buffer: the oldest reading is overwritten
  }
}

size_t bufferOldestIndex(size_t i) {
  return (bufferHead + BUFFER_CAPACITY - bufferCount + i) % BUFFER_CAPACITY;
}

void appendValues(JsonObject message, const Reading &reading) {
  JsonObject values = message["values"].to<JsonObject>();
  values["temperature"]["value"] = reading.temperature;
  values["temperature"]["unit"] = "C";
  values[RELAY_SIGNAL]["value"] = reading.relay;
}

bool publishFrame(const String &topic, const String &body) {
  String frame = lm1SignFrame(topic, body, DEVICE_PASSWORD);
  if (frame.length() > LM1_MAX_FRAME_BYTES) {
    Serial.println("frame over 64 KB - the backend would drop it; sending fewer messages");
    return false;
  }
  return mqttClient.publish(topic.c_str(), (const uint8_t *)frame.c_str(), frame.length(), false);
}

// ------------------------------------------------------------ telemetry

Reading captureReading() {
  Reading reading;
  reading.sequence = sequence.next();
  reading.observedAt = (unsigned long)time(nullptr);
  reading.temperature = 24.7f;  // replace with a real sensor read
  reading.relay = relayOn;
  return reading;
}

bool publishSingle(const Reading &reading) {
  JsonDocument doc;
  doc["timestamp"] = isoTime(reading.observedAt);
  doc["sequence"] = reading.sequence;
  appendValues(doc.as<JsonObject>(), reading);
  String body;
  serializeJson(doc, body);
  return publishFrame(lm1TelemetryTopic(), body);
}

bool publishBatch(size_t count) {
  JsonDocument doc;
  JsonArray messages = doc["messages"].to<JsonArray>();
  for (size_t i = 0; i < count; i++) {
    const Reading &reading = buffer[bufferOldestIndex(i)];
    JsonObject message = messages.add<JsonObject>();
    message["timestamp"] = isoTime(reading.observedAt);
    message["sequence"] = reading.sequence;  // strictly increasing inside the batch
    appendValues(message, reading);
  }
  String body;
  serializeJson(doc, body);
  return publishFrame(lm1TelemetryTopic(), body);
}

void flushBuffer() {
  while (bufferCount > 0) {
    size_t count = bufferCount < MAX_BATCH_MESSAGES ? bufferCount : MAX_BATCH_MESSAGES;
    if (!publishBatch(count)) {
      return;  // keep the backlog and retry on the next tick
    }
    bufferCount -= count;
    Serial.print("flushed ");
    Serial.println(count);
  }
}

// -------------------------------------------------------------- commands

void publishAck(const String &commandId, const char *status, bool value, const char *error) {
  JsonDocument doc;
  doc["command_id"] = commandId;
  doc["status"] = status;
  if (strcmp(status, "ok") == 0) {
    doc["value"] = value;
  } else {
    doc["error"] = error;
  }
  String body;
  serializeJson(doc, body);
  publishFrame(lm1CommandAckTopic(), body);
}

void onCommand(char *topic, byte *payload, unsigned int length) {
  String frame;
  frame.reserve(length);
  for (unsigned int i = 0; i < length; i++) {
    frame += (char)payload[i];
  }

  String body;
  if (!lm1VerifyFrame(String(topic), frame, DEVICE_PASSWORD, body)) {
    Serial.println("rejected an unverifiable frame");
    return;
  }

  JsonDocument doc;
  if (deserializeJson(doc, body)) {
    return;
  }
  if (String(doc["type"] | "") != "COMMAND") {
    return;
  }

  String commandId = doc["command_id"] | "";
  if (String(doc["command"] | "") != RELAY_SIGNAL || String(doc["kind"] | "") != "boolean") {
    publishAck(commandId, "error", false, "unsupported_command");
    return;
  }

  relayOn = doc["value"] | false;
  digitalWrite(RELAY_PIN, relayOn ? HIGH : LOW);
  Serial.print("relay -> ");
  Serial.println(relayOn ? "ON" : "OFF");
  publishAck(commandId, "ok", relayOn, "");
  publishSingle(captureReading());  // move the dashboard's reported state
}

// ------------------------------------------------------------ connection

bool connectMqtt() {
  if (!mqttClient.connect(DEVICE_ID, DEVICE_ID, DEVICE_PASSWORD)) {
    return false;
  }
  mqttClient.subscribe(lm1CommandTopic().c_str(), 1);
  return true;
}

void serviceConnection() {
  if (mqttClient.connected()) {
    reconnectDelayMs = RECONNECT_MIN_MS;
    return;
  }
  if (millis() < nextReconnectMs) {
    return;
  }
  if (WiFi.status() != WL_CONNECTED) {
    WiFi.begin(WIFI_SSID, WIFI_PASSWORD);
  }
  if (connectMqtt()) {
    Serial.println("connected and subscribed");
    reconnectDelayMs = RECONNECT_MIN_MS;
    flushBuffer();
    return;
  }
  Serial.print("reconnect failed, state=");
  Serial.println(mqttClient.state());
  nextReconnectMs = millis() + reconnectDelayMs;
  reconnectDelayMs = reconnectDelayMs * 2 > RECONNECT_MAX_MS ? RECONNECT_MAX_MS : reconnectDelayMs * 2;
}

// ----------------------------------------------------------------- main

void setup() {
  Serial.begin(115200);
  delay(500);
  pinMode(RELAY_PIN, OUTPUT);
  digitalWrite(RELAY_PIN, LOW);

  WiFi.mode(WIFI_STA);
  WiFi.begin(WIFI_SSID, WIFI_PASSWORD);
  while (WiFi.status() != WL_CONNECTED) {
    delay(250);
  }
  if (!lm1WaitForClock()) {
    Serial.println("NTP sync failed - refusing to sign with a wrong clock");
    return;
  }
  sequence.begin();  // seed AFTER the clock is real

  tlsClient.setInsecure();  // bring-up only; pin a CA certificate in production
  mqttClient.setServer(MQTT_HOST, MQTT_PORT);
  mqttClient.setKeepAlive(30);
  mqttClient.setBufferSize(4096);  // default 256; batches need far more
  mqttClient.setCallback(onCommand);
  connectMqtt();
  Serial.println("ready");
}

void loop() {
  serviceConnection();
  mqttClient.loop();

  if (millis() - lastSendMs >= SEND_INTERVAL_MS) {
    lastSendMs = millis();
    Reading reading = captureReading();
    if (mqttClient.connected() && bufferCount == 0) {
      publishSingle(reading);
      Serial.print("sent seq=");
      Serial.println(reading.sequence);
    } else {
      bufferPush(reading);
      Serial.print("buffered (");
      Serial.print(bufferCount);
      Serial.println(" held)");
      if (mqttClient.connected()) {
        flushBuffer();
      }
    }
  }
  delay(10);
}