MQTT-SN + ESP8266 / Хабр

В процессе поисков более легковесного протокола, похожего на полюбившийся мне MQTT для проекта беспроводных датчиков отслеживания положения на базе ESP8266 - оказалось, что существует, но пока не сильно распространена, версия протокола с названием MQTT For Sensor Networks (MQTT-SN).

“MQTT-SN спроектирован как можно более похожим на MQTT, но адаптирован к особенностям беспроводной среды передачи данных, таким как низкая пропускная способность, высокие вероятность сбоя в соединениях, короткая длина сообщения и т.п. Также оптимизирован для дешевых устройствах с аккумуляторным питанием и ограниченными ресурсами по обработке и хранения.”

В интернете довольно мало информации о данном протоколе, ковыряние в которой и стало основой для написание данной заметки.

Основные отличия MQTT-SN от “старшего брата” это уменьшение размера сообщения , в основном, за счет сокращения “служебной” информации, особенно интересна реализация QOS -1 - когда клиент отправляет сообщение без подтверждения о подтверждении доставки, и возможность использование отличного от TCP протокола, можно встретить реализации для UDP, UDP6, ZigBee, LoRaWAN, Bluetooth.

Не буду сильно погружаться в описание - кому интересно, можете ознакомиться со спецификацией MQTT-SN в OASIS. Приведу лишь пару схем из стандарта:

Первое что бросается в глаза - наличие MQTT брокера на схеме, помимо клиентов, шлюзов и форвардер MQTT-SN (не придумал как перевести, написал по аналогии с DNS). А это значит, что для функционирования протокола “сети сенсоров” полноценный MQTT брокер обязателен и необходим.

Если рассмотреть функции каждого участника обмена то получается следующее:

MQTT-SN клиенты (как принимающие так и передающие сообщения) - подключаются к MQTT брокеру, через MQTT-SN шлюзы.
MQTT-SN шлюз - основная функция двусторонняя "синтаксическая" трансляция MQTT-SN ↔ MQTT.
MQTT-SN форвардер - если клиентам недоступен шлюз, они могут посылать и принимать сообщения через него.
MQTT брокер - сервер, своеобразное ядро системы, который тем и занимается что пересылает сообщения.

Здесь на картинке тоже можно видеть полноценный взрослый MQTT-брокер и два режима работы MQTT-SN шлюзов:

В прозрачном режиме для каждого клиента шлюз устанавливает и поддерживает отдельное соединение с MQTT брокером. Это соединение зарезервировано исключительно для сквозного и прозрачного обмена сообщениями между клиентом и брокером. Шлюз выполняет трансляцию между протоколами. Ну и поскольку весь обмен сообщениями осуществляется сквозным образом, все функции и возможности, которые реализуются, могут быть использованы клиентами.
В режиме агрегации, шлюз будет иметь только одно соединение с MQTT брокером. Все сообщения остаются между клиентами и шлюзом, а уже шлюз решает, что отправлять брокеру или какому клиенту принятое сообщения от MQTT брокера передать.

Ну что же - перейдем к реализации, в качестве операционной системы я использовал Ubuntu 20.04 со статическим адресом 10.10.10.10/24.

MQTT

Устанавливаем Eclipse Mosquitto:

sudo apt install mosquitto

Для тестирования нам не понадобятся какие-то настройки в отношении безопасности и т.п. Но я крайне не рекомендую так делать в производстве. Хотя если вы решили использовать MQTT/MQTT-SN на промышленном уровне все необходимые инструменты имеются. После установки давайте проверим как пересылаются сообщения - я использую Python для этого. Установим библиотеку paho-mqtt.

pip install paho-mqtt

Скрипт, передающий в топик “habr” сообщение “Hello Habrahabr!”:

import paho.mqtt.publish as publish

msg = "Hello Habrahabr!" 
publish.single("habr", msg, hostname="10.10.10.10", port=1883)

Скрипт, подписывается на топик “habr” и принимает все сообщения:

import paho.mqtt.client as mqtt

def on_connect(client, userdata, flags, rc):
    client.subscribe("habr/#")

def on_message(client, userdata, msg):
    print(msg.topic + ' ' + str(msg.payload))

client = mqtt.Client()
client.on_connect = on_connect
client.on_message = on_message

client.connect("10.10.10.10", 1883, 60)
client.loop_forever()

Чтобы более подробнее познакомится с MQTT очень рекомендую блог Steve’s Internet Guide, ну и поиск не только по хабру, конечно.

MQTT-SN

Убедившись что наш брокер работает, переходим к следующему этапу. Я буду использовать шлюз из репозитория paho.mqtt-sn.embedded-c, повторим действия для компиляции шлюза в нашей ОС.

Для начала установим необходимые пакеты для сборки:

sudo apt-get install build-essential libssl-dev

Клонируем репозиторий себе в систему, переходим в папку со шлюзом и компилируем:

git clone -b develop https://github.com/eclipse/paho.mqtt-sn.embedded-c
cd paho.mqtt-sn.embedded-c/MQTTSNGateway
make install
make clean

По умолчанию пакет ставится в директорию, куда вы клонировали репозиторий - если вы как и не особо пока заморачивались - то в домашнюю ;) Для первого запуска нужно отредактировать конфигурацию шлюза и запустить его с правами sudo. В дальнейшем можно запускать уже обычно. Наша простая конфигурация (для большего упрощения я убрал закомментировать строки).

Наша простая конфигурация (для большего упрощения я убрал закомментировать строки):

BrokerName=localhost
BrokerPortNo=1883
BrokerSecurePortNo=8883

ClientAuthentication=NO
AggregatingGateway=NO
QoS-1=NO
Forwarder=NO
PredefinedTopic=NO
GatewayID=1
GatewayName=Paho-MQTT-SN-Gateway
KeepAlive=900
# UDP
GatewayPortNo=10000
MulticastIP=225.1.1.1
MulticastPortNo=1885
MulticastTTL=1

Как и писалось выше первый запуск делаем с “sudo” из домашней директории, при этом у нас будет полный вывод всего происходящего в консоли:

sudo ./MQTT-SNGateway

 ***************************************************************************
 * MQTT-SN Gateway
 * Part of Project Paho in Eclipse
 * (http://git.eclipse.org/c/paho/org.eclipse.paho.mqtt-sn.embedded-c.git/)
 *
 * Author : Tomoaki YAMAGUCHI
 * Version: 1.4.0
 ***************************************************************************

20210404 224219.274 Paho-MQTT-SN-Gateway has been started.

 ConfigFile: ./gateway.conf
 SensorN/W:  UDP Multicast 225.1.1.1:1885 Gateway Port 10000 TTL: 1
 Broker:     localhost : 1883, 8883
 RootCApath: (null)
 RootCAfile: (null)
 CertKey:    (null)
 PrivateKey: (null)

Давайте теперь что-нибудь уже отправим нашему шлюзу, который это сообщение передаст MQTT-брокеру, для отправки будем использовать все тот же Python и MQTT-SN client for Python 3 and Micropython. В репозитории есть примеры для отправки и приема сообщений, немного подправив их, мы сможем уже отправлять и принимать сообщения как из MQTT сегмента куда-либо, так и из MQTT-SN сегмента.

mqttsn_publisher.py

from mqttsn.MQTTSNclient import Client
import struct
import time
import sys

class Callback:

    def published(self, MsgId):
        print("Published")

def connect_gateway():
    try:
        while True:
            try:
                aclient.connect()
                print('Connected to gateway...')
                break
            except:
                print('Failed to connect to gateway, reconnecting...')
                time.sleep(1)
    except KeyboardInterrupt:
        print('Exiting...')
        sys.exit()

def register_topic():
    global topic
    topic = aclient.register("habr")
    print("topic registered.")

aclient = Client("mqtt_sn_client", "10.10.10.10", port=10000)
aclient.registerCallback(Callback())
connect_gateway()

topic = None
register_topic()

payload = ‘Hello Habrahabr!’

pub_msgid = aclient.publish(topic, payload, qos=0)

aclient.disconnect()
print("Disconnected from gateway.")

Не буду здесь приводить много простого кода - если до этих пор все у вас получалось - думаю разберетесь и дальше ;)

ESP8266

Теперь пришло время настоящего веселья. Будем применять протокол, по моему мнению, на наиболее подходящих для него микроконтроллерах esp8266.

На самом деле готовых реализаций несколько и ни одна из них у меня корректно не завелась “без доработки напильником». Наиболее логичной реализацией мне показалась у MQTT-SN клиента у некоего Gabriel Nikol в репозитории arduino-mqtt-sn-client на GitHub.

Проблема 1. Тестовый пример, возможно в авторской реализации шлюза и работает (я не пробовал), но с Paho ни в какую не хочет. Ну что же, в запросах на репозитории висят похожие проблемы, будем пробовать решать. Исправляем, как указано здесь в запросе некорректный параметр типов топика и - все получилось! Сообщения отправляются - красота.

Проблема 2. Но при подписывании на топики - мы наблюдаем, что у нас к каждому сообщению добавляется “0x00”, который считается признаком конца строки, но почему-то у нас во всех других способах отправки сообщений ничего подобного нет. Пробежавшись по спецификации протокола, я и правда не нашел, что у нас сообщение обязательно должно заканчиваться так - вырезаем это в отправке сообщений. Еще стали на шаг ближе! Что мы стараемся отправить, то и получаем.

Проблема 3. Для обычной реализации MQTT протокола, я использовал для передачи кватерниона массив байт - так меньше сообщение, 16 (4 числа типа float) вместо 32 (если считать один знак до и 6 после запятой). Оказалось, что в данной реализации используется символьный тип данных char, где каждый байт интерпретируется как ASCII-символ. Давайте добавим и такую возможность - отправлять массив байтов.

Проблема 4. Заметил довольно длинный временной промежуток между подключением к беспроводной сети микроконтроллера и первым соединением со шлюзом. Давайте посмотрим в чем дело. Оказывается - при подключении, наш микроконтроллер спит при первом подключении 10 секунд, на втором 20. Исправим на 50 мс для первой и соответственно 100 для второй попытки - на данном этапе я думаю этого хватит - я проблем не заметил, но на всякий случай увеличил количество попыток до 5 (разумеется для использования в “реальном мире” нужно пересматривать этот таймаут).

Ну больше каких-то таких проблем в использовании я не нашел, и если кто-то что-то найдет - создавайте запросы автору (как то “меня терзают смутные сомнения”, что он продолжает поддерживать свое творение, но за спрос - не бьют в нос).

main.cpp

#include <I2Cdev.h>
#include <MPU9250_9Axis_MotionApps41.h>
#include <ESP8266WiFi.h>
#include <WiFiUdp.h>
#include <WiFiUdpSocket.h>
#include <MqttSnClient.h>
#include <ArduinoOTA.h>

const char* ssid     = "habr";
const char* password = "Hello Habrahabr!";

MPU9250 mpu;

#define SDA 4
#define SCL 5

IPAddress ip(10, 10, 10, 30);
IPAddress gateway(10, 10, 10, 1);
IPAddress subnet(255, 255, 255, 0);
IPAddress gatewayIPAddress(10, 10, 10, 100);
uint16_t localUdpPort = 10000;

WiFiUDP udp;
WiFiUdpSocket wiFiUdpSocket(udp, localUdpPort);
MqttSnClient<WiFiUdpSocket> mqttSnClient(wiFiUdpSocket);

const char* clientId = "thigh_l";
char* subscribeTopicName = "main";
char* publishTopicName = "adam/thigh_l";
String messageMQTT;

uint16_t packetSize;
uint16_t fifoCount;
uint8_t fifoBuffer[48];
bool blinkState = false;
bool sendQuat = false;

Quaternion q;

int8_t qos = 0;

void mqttsn_callback(char *topic, uint8_t *payload, uint16_t length, bool retain) {
  for (uint16_t i = 0; i < length; i++) {
    messageMQTT += (char)payload[i];
  }
  if (messageMQTT == "start1"){
    sendQuat = true;
    messageMQTT = "";
  }
  else if (messageMQTT == "stop") {
    sendQuat = false;
    messageMQTT = "";
  }
}

void setup_wifi() {
  delay(10);
  WiFi.setSleepMode(WIFI_NONE_SLEEP);
  WiFi.mode(WIFI_STA);
  WiFi.config(ip, gateway, subnet);
  WiFi.begin(ssid, password);
  while (WiFi.status() != WL_CONNECTED) {
    delay(50);
  }
}

void convertIPAddressAndPortToDeviceAddress(IPAddress& source, uint16_t port, device_address& target) {
  target.bytes[0] = source[0];
  target.bytes[1] = source[1];
  target.bytes[2] = source[2];
  target.bytes[3] = source[3];
  target.bytes[4] = port >> 8;
  target.bytes[5] = (uint8_t) port ;
}

void setup() {
  Wire.begin(SDA, SCL);
  Wire.setClock(400000);
  Serial.begin(115200);
  setup_wifi();
  mpu.initialize();
  mpu.dmpInitialize();
  mpu.setDMPEnabled(true);
  packetSize = mpu.dmpGetFIFOPacketSize();
  fifoCount = mpu.getFIFOCount();
  ArduinoOTA.onStart([]() {
  });
  ArduinoOTA.onEnd([]() {
  });
  ArduinoOTA.onProgress([](unsigned int progress, unsigned int total) {
  });
  ArduinoOTA.onError([](ota_error_t error) {
    if (error == OTA_AUTH_ERROR) Serial.println("Auth Failed");
    else if (error == OTA_BEGIN_ERROR) Serial.println("Begin Failed");
    else if (error == OTA_CONNECT_ERROR) Serial.println("Connect Failed");
    else if (error == OTA_RECEIVE_ERROR) Serial.println("Receive Failed");
    else if (error == OTA_END_ERROR) Serial.println("End Failed");
  });
  ArduinoOTA.begin();
  pinMode(LED_BUILTIN, OUTPUT);
  mqttSnClient.begin();
  device_address gateway_device_address;
  convertIPAddressAndPortToDeviceAddress(gatewayIPAddress, localUdpPort, gateway_device_address);
  mqttSnClient.connect(&gateway_device_address, clientId, 180);
  mqttSnClient.setCallback(mqttsn_callback);
  mqttSnClient.subscribe(subscribeTopicName, qos);
}

void loop() {
  fifoCount = mpu.getFIFOCount();
    if (fifoCount == 1024) {
      mpu.resetFIFO();
      }
    else if (fifoCount % packetSize != 0) {
      mpu.resetFIFO();
      }
    else if (fifoCount >= packetSize  && sendQuat) {
        mpu.getFIFOBytes(fifoBuffer, packetSize);
        fifoCount -= packetSize;
        mpu.dmpGetQuaternion(&q, fifoBuffer);
        mqttSnClient.publish((uint8_t*)&q, publishTopicName, qos);
        blinkState = !blinkState;
        digitalWrite(LED_BUILTIN, blinkState);
        mpu.resetFIFO();
      }
  ArduinoOTA.handle();
  mqttSnClient.loop();
}

Тестирование

Вот теперь действительно началось самое интересное. Так ли уже хорош MQTT-SN против MQTT, ведь предназначен именно для беспроводного подключения.

У меня есть 15 датчиков с микроконтроллерами, и в своем тестовом проекте по захвату движений я использовал MQTT, в качестве старта передачи данных использовались сообщения в топик “main” и у меня была проверка на изменение кватерниона (т.е. новое сообщение отправлялось, когда предыдущий кватернион отличался от настоящего примерно на 0,5⁰). Не сложно будет изменить прошивки, чтобы для каждого микроконтроллера была своя команда старта передачи + передавать данные с частотой 50 Гц без проверки на отличия предыдущего и настоящего кватернионов.

Для этого напишем пару скриптов. Я себе представляю алгоритм тестирования следующим образом: Передаем сообщение для старта передачи с датчика, считываем среднее значение полученных сообщений в секунду, каждую секунду пишем в файл полученное количество сообщений, до запуска передачи от следующего датчика смотрим в диспетчере задач сколько потребляет трафика процесс “MQTTSN-Gateway”. Просто, быстро и не очень трудоемко - нужно делать, но лень. Для полномасштабного теста подожду уже готовые платки.

Для начала решил проверить, а все ли сообщения доходят, мы то используем в качестве транспорта UDP, который не гарантирует “обеспечение надежности, упорядочивания или целостности данных”. На протяжении 5 минут делал следующее - скриптом захватывал все сообщения и записывал время приема в файл, параллельно другим скриптом захватывал сообщения из последовательного порта и так же записывал время приема в другой файл. Получилось больше 13200 строк и соответствие в 100%, то есть сколько контроллер отправил сообщений, столько и было получено. Диспетчер задач показывал среднюю нагрузку сетевого интерфейса на получение 24-48 Кбит/с и 170-200 Кбит/с на отдачу. При таком же тестировании но с протоколом MQTT нагрузка на сетевой интерфейс составила 48-64 и 200-300 соответственно. Можете мне не верить и проверить все сами:) Как говорится налицо преимущества, но это для одного только датчика.

Кому интересно - ссылка на ~~этот весь говнокод~~ репозиторий. Продолжение следует...