m5/M5Stack_MicroPython
0
0
Fork 0
mirror of https://github.com/m5stack/M5Stack_MicroPython.git synced 2026-05-20 10:14:44 -07:00
Code Issues Packages Projects Releases Wiki Activity
Files
master
M5Stack_MicroPython/MicroPython_BUILD/components/micropython/esp32/modmqtt.c
T
sakabin 070e06db00 update micropython 
add m5stack.py
add m5ui.py
2018-12-04 14:06:15 +08:00

904 lines
35 KiB
C
Raw Permalink Blame History

/*
* This file is part of the MicroPython ESP32 project, https://github.com/loboris/MicroPython_ESP32_psRAM_LoBo
*
* The MIT License (MIT)
*
* Copyright (c) 2018 LoBo (https://github.com/loboris)
*
* Permission is hereby granted, free of charge, to any person obtaining a copy
* of this software and associated documentation files (the "Software"), to deal
* in the Software without restriction, including without limitation the rights
* to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
* copies of the Software, and to permit persons to whom the Software is
* furnished to do so, subject to the following conditions:
*
* The above copyright notice and this permission notice shall be included in
* all copies or substantial portions of the Software.
*
* THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
* IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
* FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
* AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
* LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
* OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN
* THE SOFTWARE.
*/
/*
* Mqtt Module using MQTT task.
* Based on ESP32 MQTT Library by Tuan PM, https://github.com/tuanpmt/espmqtt
* Adapted for MicroPython by Boris Lovosevic, https://github.com/loboris
*
*/
#include "sdkconfig.h"
#define CONFIG_MICROPY_USE_MQTT
#ifdef CONFIG_MICROPY_USE_MQTT
#include <stdint.h>
#include <stdio.h>
#include <string.h>
#include "mqtt_client.h"
#include "http_parser.h"
#include "py/nlr.h"
#include "py/runtime.h"
#include "modmachine.h"
#include "mphalport.h"
#include "extmod/vfs_native.h"
#define CONFIG_MQTT_MAX_TASKNAME_LEN 16
typedef struct _mqtt_obj_t {
mp_obj_base_t base;
esp_mqtt_client_handle_t client;
esp_mqtt_client_config_t mqtt_cfg;
char name[CONFIG_MQTT_MAX_TASKNAME_LEN];
void *mpy_connected_cb;
void *mpy_disconnected_cb;
void *mpy_subscribed_cb;
void *mpy_unsubscribed_cb;
void *mpy_published_cb;
void *mpy_data_cb;
uint8_t *msgbuf;
uint8_t *topicbuf;
char *certbuf;
uint8_t subs_flag;
uint8_t unsubs_flag;
uint8_t publish_flag;
} mqtt_obj_t;
const mp_obj_type_t mqtt_type;
//------------------------------------------
STATIC int checkClient(mqtt_obj_t *mqtt_obj)
{
if (mqtt_obj->client == NULL) {
nlr_raise(mp_obj_new_exception_msg(&mp_type_OSError, "Mqtt client destroyed"));
}
return mqtt_obj->client->state;
}
//----------------------------------------
STATIC void connected_cb(mqtt_obj_t *self)
{
if (self->mpy_connected_cb) {
mp_sched_carg_t *carg = make_cargs(MP_SCHED_CTYPE_SINGLE);
if (!carg) return;
if (!make_carg_entry(carg, 0, MP_SCHED_ENTRY_TYPE_STR, strlen(self->name), (const uint8_t *)self->name, NULL)) return;
mp_sched_schedule(self->mpy_connected_cb, mp_const_none, carg);
}
}
//-------------------------------------------
STATIC void disconnected_cb(mqtt_obj_t *self)
{
if (self->mpy_disconnected_cb) {
mp_sched_carg_t *carg = make_cargs(MP_SCHED_CTYPE_SINGLE);
if (!carg) return;
if (!make_carg_entry(carg, 0, MP_SCHED_ENTRY_TYPE_STR, strlen(self->name), (const uint8_t *)self->name, NULL)) return;
mp_sched_schedule(self->mpy_disconnected_cb, mp_const_none, carg);
}
}
//------------------------------------------------------------
STATIC void subscribed_cb(mqtt_obj_t *self, const char *topic)
{
if (self->mpy_subscribed_cb) {
mp_sched_carg_t *carg = make_cargs(MP_SCHED_CTYPE_TUPLE);
if (carg == NULL) return;
if (!make_carg_entry(carg, 0, MP_SCHED_ENTRY_TYPE_STR, strlen(self->name), (const uint8_t *)self->name, NULL)) return;
if (topic) {
if (!make_carg_entry(carg, 1, MP_SCHED_ENTRY_TYPE_STR, strlen(topic), (const uint8_t *)topic, NULL)) return;
}
else {
if (!make_carg_entry(carg, 1, MP_SCHED_ENTRY_TYPE_STR, 1, (const uint8_t *)"?", NULL)) return;
}
mp_sched_schedule(self->mpy_subscribed_cb, mp_const_none, carg);
}
}
//--------------------------------------------------------------
STATIC void unsubscribed_cb(mqtt_obj_t *self, const char *topic)
{
if (self->mpy_unsubscribed_cb) {
mp_sched_carg_t *carg = make_cargs(MP_SCHED_CTYPE_TUPLE);
if (carg == NULL) return;
if (!make_carg_entry(carg, 0, MP_SCHED_ENTRY_TYPE_STR, strlen(self->name), (const uint8_t *)self->name, NULL)) return;
if (topic) {
if (!make_carg_entry(carg, 1, MP_SCHED_ENTRY_TYPE_STR, strlen(topic), (const uint8_t *)topic, NULL)) return;
}
else {
if (!make_carg_entry(carg, 1, MP_SCHED_ENTRY_TYPE_STR, 1, (const uint8_t *)"?", NULL)) return;
}
mp_sched_schedule(self->mpy_unsubscribed_cb, mp_const_none, carg);
}
}
//---------------------------------------------------------------------
STATIC void published_cb(mqtt_obj_t *self, const char *topic, int type)
{
if (self->mpy_published_cb) {
mp_sched_carg_t *carg = make_cargs(MP_SCHED_CTYPE_TUPLE);
if (carg == NULL) return;
if (!make_carg_entry(carg, 0, MP_SCHED_ENTRY_TYPE_STR, strlen(self->name), (const uint8_t *)self->name, NULL)) return;
if (topic) {
if (!make_carg_entry(carg, 1, MP_SCHED_ENTRY_TYPE_STR, strlen(topic), (const uint8_t *)topic, NULL)) return;
}
else {
if (!make_carg_entry(carg, 1, MP_SCHED_ENTRY_TYPE_STR, 1, (const uint8_t *)"?", NULL)) return;
}
if (!make_carg_entry(carg, 2, MP_SCHED_ENTRY_TYPE_INT, type, NULL, NULL)) return;
mp_sched_schedule(self->mpy_published_cb, mp_const_none, carg);
}
}
//-------------------------------------------------
STATIC void data_cb(mqtt_obj_t *self, void *params)
{
if (!self->mpy_data_cb) return;
esp_mqtt_event_handle_t event = (esp_mqtt_event_handle_t)params;
if (event->current_data_offset == 0) {
// *** First block of data
if (self->msgbuf != NULL) free(self->msgbuf);
if (self->topicbuf != NULL) free(self->topicbuf);
self->msgbuf = NULL;
self->topicbuf = NULL;
if (event->data_len < event->total_data_len) {
// === more data will follow, allocate the data buffer and copy the first part ===
self->topicbuf = malloc(event->topic_len + 1);
if (self->topicbuf) {
memcpy(self->topicbuf, event->topic, event->topic_len);
self->topicbuf[event->topic_len] = 0;
int buf_len = event->total_data_len + 1;
self->msgbuf = malloc(buf_len + 1);
if (self->msgbuf) {
memcpy(self->msgbuf, event->data, event->data_len);
self->msgbuf[event->data_len] = 0;
}
else {
free(self->topicbuf);
self->msgbuf = NULL;
self->topicbuf = NULL;
}
}
}
else {
// === all data received, we can schedule the callback function now ===
mp_sched_carg_t *carg = make_cargs(MP_SCHED_CTYPE_TUPLE);
if (!carg) return;
if (!make_carg_entry(carg, 0, MP_SCHED_ENTRY_TYPE_STR, strlen(self->name), (const uint8_t *)self->name, NULL)) return;
if (!make_carg_entry(carg, 1, MP_SCHED_ENTRY_TYPE_STR, event->topic_len, (const uint8_t *)event->topic, NULL)) return;
if (!make_carg_entry(carg, 2, MP_SCHED_ENTRY_TYPE_STR, event->data_len, (const uint8_t *)event->data, NULL)) return;
mp_sched_schedule(self->mpy_data_cb, mp_const_none, carg);
}
}
else {
if ((self->topicbuf) && (self->msgbuf)) {
// === more payload data arrived, add to buffer ===
int new_len = event->current_data_offset + event->data_len;
memcpy(self->msgbuf + event->current_data_offset, event->data, event->data_len);
self->msgbuf[new_len] = 0;
if (new_len >= event->total_data_len) {
// === all data received, we can schedule the callback function now ===
mp_sched_carg_t *carg = make_cargs(MP_SCHED_CTYPE_TUPLE);
if (!carg) goto freebufs;
if (!make_carg_entry(carg, 0, MP_SCHED_ENTRY_TYPE_STR, strlen(self->name), (const uint8_t *)self->name, NULL)) goto freebufs;
if (!make_carg_entry(carg, 1, MP_SCHED_ENTRY_TYPE_STR, strlen((const char *)self->topicbuf), self->topicbuf, NULL)) goto freebufs;
if (!make_carg_entry(carg, 2, MP_SCHED_ENTRY_TYPE_STR, event->total_data_len, self->msgbuf, NULL)) goto freebufs;
mp_sched_schedule(self->mpy_data_cb, mp_const_none, carg);
freebufs:
// Free the buffers
free(self->msgbuf);
free(self->topicbuf);
self->msgbuf = NULL;
self->topicbuf = NULL;
}
}
else {
// more payload data arrived, but there is no data buffers allocated (!?)
if (self->msgbuf != NULL) free(self->msgbuf);
if (self->topicbuf != NULL) free(self->topicbuf);
self->msgbuf = NULL;
self->topicbuf = NULL;
}
}
}
//----------------------------------------------------------------
static esp_err_t mqtt_event_handler(esp_mqtt_event_handle_t event)
{
esp_mqtt_client_handle_t client = event->client;
mqtt_obj_t *mpy_client = (mqtt_obj_t *)client->mpy_mqtt_obj;
const char *topic = NULL;
// your_context_t *context = event->context;
switch (event->event_id) {
case MQTT_EVENT_CONNECTED:
ESP_LOGI(MQTT_TAG, "Connected");
connected_cb(mpy_client);
break;
case MQTT_EVENT_DISCONNECTED:
ESP_LOGI(MQTT_TAG, "Disconnected");
disconnected_cb(mpy_client);
break;
case MQTT_EVENT_SUBSCRIBED:
if (client->config->user_context) {
topic = (const char *)client->config->user_context;
ESP_LOGI(MQTT_TAG, "Subscribed to '%s'", topic);
}
else {
ESP_LOGI(MQTT_TAG, "Subscribed");
}
subscribed_cb(mpy_client, topic);
mpy_client->subs_flag = 1;
break;
case MQTT_EVENT_UNSUBSCRIBED:
if (client->config->user_context) {
topic = (const char *)client->config->user_context;
ESP_LOGI(MQTT_TAG, "Unsubscribed from '%s'", topic);
}
else {
ESP_LOGI(MQTT_TAG, "Unsubscribed");
}
unsubscribed_cb(mpy_client, topic);
mpy_client->unsubs_flag = 1;
break;
case MQTT_EVENT_PUBLISHED:
if (client->config->user_context) {
topic = (const char *)client->config->user_context;
ESP_LOGI(MQTT_TAG, "Published to '%s'", topic);
}
else {
ESP_LOGI(MQTT_TAG, "Published");
}
published_cb(mpy_client, topic, event->type);
mpy_client->publish_flag = 1;
break;
case MQTT_EVENT_DATA:
if (mpy_client->mpy_data_cb == NULL) {
ESP_LOGI(MQTT_TAG, "TOPIC: %.*s\r\n", event->topic_len, event->topic);
ESP_LOGI(MQTT_TAG, " DATA: %.*s\r\n", event->data_len, event->data);
}
else {
ESP_LOGI(MQTT_TAG, "Data received");
}
data_cb(mpy_client, event);
break;
case MQTT_EVENT_ERROR:
ESP_LOGI(MQTT_TAG, "Mqtt Error");
break;
}
return ESP_OK;
}
//-------------------------------------------------------------------------------------
STATIC void mqtt_print(const mp_print_t *print, mp_obj_t self_in, mp_print_kind_t kind)
{
mqtt_obj_t *self = self_in;
if (self->client == NULL) {
mp_printf(print, "Mqtt[%s]( Destroyed )\n", self->name);
return;
}
char sstat[16];
if (self->client->state == MQTT_STATE_CONNECTED) sprintf(sstat, "Connected");
else if (self->client->state == MQTT_STATE_INIT) sprintf(sstat, "Initialized");
else if (self->client->state == MQTT_STATE_WAIT_TIMEOUT) sprintf(sstat, "Wait timeout");
else if (self->client->state == MQTT_STATE_UNKNOWN) sprintf(sstat, "Unknown");
else sprintf(sstat, "Error");
const char *server_uri = "Unknown";
if (self->client->config->uri) server_uri = self->client->config->uri;
else if (self->client->config->host) server_uri = self->client->config->host;
mp_printf(print, "Mqtt[%s]\n (Server: %s:%u, Status: %s\n", self->name, server_uri, self->client->config->port, sstat);
if ((self->client->state == MQTT_STATE_CONNECTED) || (self->client->state == MQTT_STATE_INIT)) {
mp_printf(print, " Client ID: %s, Clean session=%s, Keepalive=%ds\n LWT(",
self->client->connect_info.client_id, (self->client->connect_info.clean_session ? "True" : "False"), self->client->connect_info.keepalive);
if (self->client->connect_info.will_topic) {
mp_printf(print, "QoS=%d, Retain=%s, Topic='%s', Msg='%s')\n",
self->client->connect_info.will_qos, (self->client->connect_info.will_retain ? "True" : "False"), self->client->connect_info.will_topic, self->client->connect_info.will_message);
}
else mp_printf(print, "not set)\n");
}
/*
if ((self->client->settings->xMqttTask) && (self->client->settings->xMqttSendingTask)) {
mp_printf(print, " Used stack: %u/%u + %u/%u\n",
self->client->settings->xMqttTask_stacksize - uxTaskGetStackHighWaterMark(self->client->settings->xMqttTask), self->client->settings->xMqttTask_stacksize,
self->client->settings->xMqttSendingTask_stacksize - uxTaskGetStackHighWaterMark(self->client->settings->xMqttSendingTask), self->client->settings->xMqttSendingTask_stacksize);
}
*/
mp_printf(print, " )\n");
}
//------------------------------------------------------------------------------------------------------------
STATIC mp_obj_t mqtt_make_new(const mp_obj_type_t *type, size_t n_args, size_t n_kw, const mp_obj_t *all_args)
{
enum { ARG_name, ARG_server, ARG_user, ARG_pass, ARG_port, ARG_reconnect, ARG_clientid, ARG_cleansess, ARG_keepalive, ARG_cert,
ARG_lwt_topic, ARG_lwt_msg, ARG_lwt_qos, ARG_lwt_retain, ARG_datacb, ARG_connected, ARG_disconnected, ARG_subscribed, ARG_unsubscribed, ARG_published };
const mp_arg_t mqtt_init_allowed_args[] = {
{ MP_QSTR_name, MP_ARG_REQUIRED | MP_ARG_OBJ, {.u_obj = mp_const_none} },
{ MP_QSTR_server, MP_ARG_REQUIRED | MP_ARG_OBJ, {.u_obj = mp_const_none} },
{ MP_QSTR_user, MP_ARG_KW_ONLY | MP_ARG_OBJ, {.u_obj = mp_const_none} },
{ MP_QSTR_password, MP_ARG_KW_ONLY | MP_ARG_OBJ, {.u_obj = mp_const_none} },
{ MP_QSTR_port, MP_ARG_KW_ONLY | MP_ARG_INT, {.u_int = -1} },
{ MP_QSTR_autoreconnect, MP_ARG_KW_ONLY | MP_ARG_INT, {.u_int = 0} },
{ MP_QSTR_clientid, MP_ARG_KW_ONLY | MP_ARG_OBJ, {.u_obj = mp_const_none} },
{ MP_QSTR_cleansession, MP_ARG_KW_ONLY | MP_ARG_BOOL, {.u_bool = false} },
{ MP_QSTR_keepalive, MP_ARG_KW_ONLY | MP_ARG_INT, {.u_int = MQTT_KEEPALIVE_TICK} },
{ MP_QSTR_cert, MP_ARG_KW_ONLY | MP_ARG_OBJ, {.u_obj = mp_const_none} },
{ MP_QSTR_lwt_topic, MP_ARG_KW_ONLY | MP_ARG_OBJ, {.u_obj = mp_const_none} },
{ MP_QSTR_lwt_msg, MP_ARG_KW_ONLY | MP_ARG_OBJ, {.u_obj = mp_const_none} },
{ MP_QSTR_lwt_qos, MP_ARG_KW_ONLY | MP_ARG_INT, {.u_int = 0} },
{ MP_QSTR_lwt_retain, MP_ARG_KW_ONLY | MP_ARG_INT, {.u_int = 0} },
{ MP_QSTR_data_cb, MP_ARG_KW_ONLY | MP_ARG_OBJ, {.u_obj = mp_const_none} },
{ MP_QSTR_connected_cb, MP_ARG_KW_ONLY | MP_ARG_OBJ, {.u_obj = mp_const_none} },
{ MP_QSTR_disconnected_cb, MP_ARG_KW_ONLY | MP_ARG_OBJ, {.u_obj = mp_const_none} },
{ MP_QSTR_subscribed_cb, MP_ARG_KW_ONLY | MP_ARG_OBJ, {.u_obj = mp_const_none} },
{ MP_QSTR_unsubscribed_cb, MP_ARG_KW_ONLY | MP_ARG_OBJ, {.u_obj = mp_const_none} },
{ MP_QSTR_published_cb, MP_ARG_KW_ONLY | MP_ARG_OBJ, {.u_obj = mp_const_none} },
};
mp_arg_val_t args[MP_ARRAY_SIZE(mqtt_init_allowed_args)];
mp_arg_parse_all_kw_array(n_args, n_kw, all_args, MP_ARRAY_SIZE(mqtt_init_allowed_args), mqtt_init_allowed_args, args);
const char *tstr = NULL;
// Create the mqtt object
mqtt_obj_t *self = m_new_obj(mqtt_obj_t );
memset(self, 0 , sizeof(mqtt_obj_t));
// Populate settings
esp_mqtt_client_config_t mqtt_cfg = {0};
// Event handle
mqtt_cfg.event_handle = mqtt_event_handler;
// Object name
tstr = mp_obj_str_get_str(args[ARG_name].u_obj);
if (strlen(tstr) >= CONFIG_MQTT_MAX_TASKNAME_LEN) {
mp_raise_ValueError("Name too long");
}
sprintf(self->name, "%s", tstr);
// Port
if (args[ARG_port].u_int > 0) mqtt_cfg.port = args[ARG_port].u_int;
// Get URI or server domain
tstr = mp_obj_str_get_str(args[ARG_server].u_obj);
if (strlen(tstr) >= MQTT_MAX_HOST_LEN) {
mp_raise_ValueError("URI too long");
}
struct http_parser_url puri;
http_parser_url_init(&puri);
int parser_status = http_parser_parse_url(tstr, strlen(tstr), 0, &puri);
if (parser_status != 0) {
sprintf(mqtt_cfg.host, "%s", tstr);
}
else sprintf(mqtt_cfg.uri, "%s", tstr);
// User name
if (MP_OBJ_IS_STR(args[ARG_user].u_obj)) {
tstr = mp_obj_str_get_str(args[ARG_user].u_obj);
if (strlen(tstr) >= MQTT_MAX_USERNAME_LEN) {
mp_raise_ValueError("User name too long");
}
sprintf(mqtt_cfg.username, "%s", tstr);
}
// Password
if (MP_OBJ_IS_STR(args[ARG_pass].u_obj)) {
tstr = mp_obj_str_get_str(args[ARG_pass].u_obj);
if (strlen(tstr) >= MQTT_MAX_PASSWORD_LEN) {
mp_raise_ValueError("Password too long");
}
sprintf(mqtt_cfg.password, "%s", tstr);
}
// Client ID
if (MP_OBJ_IS_STR(args[ARG_clientid].u_obj)) {
tstr = mp_obj_str_get_str(args[ARG_clientid].u_obj);
if (strlen(tstr) >= MQTT_MAX_CLIENT_LEN) {
mp_raise_ValueError("Client ID too long");
}
sprintf(mqtt_cfg.client_id, "%s", tstr);
}
else sprintf(mqtt_cfg.client_id, "mpy_mqtt_client");
mqtt_cfg.disable_auto_reconnect = args[ARG_reconnect].u_int ? false : true;
mqtt_cfg.keepalive = args[ARG_keepalive].u_int;
mqtt_cfg.disable_clean_session = args[ARG_cleansess].u_int ? false : true;
// LWT options
if (MP_OBJ_IS_STR(args[ARG_lwt_topic].u_obj)) {
tstr = mp_obj_str_get_str(args[ARG_lwt_topic].u_obj);
if (strlen(tstr) >= MQTT_MAX_LWT_TOPIC) {
mp_raise_ValueError("LWT topic too long");
}
sprintf(mqtt_cfg.lwt_topic, "%s", tstr);
if (MP_OBJ_IS_STR(args[ARG_lwt_msg].u_obj)) {
tstr = mp_obj_str_get_str(args[ARG_lwt_msg].u_obj);
if (strlen(tstr) >= MQTT_MAX_LWT_MSG) {
mp_raise_ValueError("LWT message too long");
}
sprintf(mqtt_cfg.lwt_msg, "%s", tstr);
}
else sprintf(mqtt_cfg.lwt_msg, "offline");
mqtt_cfg.lwt_qos = args[ARG_lwt_qos].u_int;
mqtt_cfg.lwt_retain = args[ARG_lwt_retain].u_int;
}
// Get Certificate from file
if (MP_OBJ_IS_STR(args[ARG_cert].u_obj)) {
char *fname = NULL;
char fullname[128] = {'\0'};
fname = (char *)mp_obj_str_get_str(args[ARG_cert].u_obj);
int res = physicalPath(fname, fullname);
if ((res != 0) || (strlen(fullname) == 0)) {
mp_raise_ValueError("Certificate file not found");
}
struct stat sb;
if (stat(fullname, &sb) != 0) {
mp_raise_ValueError("Error opening certificate file");
}
int size = sb.st_size;
FILE *fhndl = fopen(fullname, "rb");
if (fhndl != NULL) {
if (self->certbuf) {
free(self->certbuf);
self->certbuf = NULL;
}
self->certbuf = malloc(size);
if (self->certbuf == NULL) {
fclose(fhndl);
mp_raise_ValueError("Error allocating certification buffer");
}
int len = fread(self->certbuf, 1, size, fhndl); // read cert file
fclose(fhndl);
if (len != size) {
free(self->certbuf);
self->certbuf = NULL;
mp_raise_ValueError("Error reading certificate file");
}
mqtt_cfg.cert_pem = (const char *)self->certbuf;
}
else {
mp_raise_ValueError("Error opening certificate file");
}
}
// Set callbacks
if ((MP_OBJ_IS_FUN(args[ARG_datacb].u_obj)) || (MP_OBJ_IS_METH(args[ARG_datacb].u_obj))) {
self->mpy_data_cb = args[ARG_datacb].u_obj;
}
if ((MP_OBJ_IS_FUN(args[ARG_connected].u_obj)) || (MP_OBJ_IS_METH(args[ARG_connected].u_obj))) {
self->mpy_connected_cb = args[ARG_connected].u_obj;
}
if ((MP_OBJ_IS_FUN(args[ARG_disconnected].u_obj)) || (MP_OBJ_IS_METH(args[ARG_disconnected].u_obj))) {
self->mpy_disconnected_cb = args[ARG_disconnected].u_obj;
}
if ((MP_OBJ_IS_FUN(args[ARG_subscribed].u_obj)) || (MP_OBJ_IS_METH(args[ARG_subscribed].u_obj))) {
self->mpy_subscribed_cb = args[ARG_subscribed].u_obj;
}
if ((MP_OBJ_IS_FUN(args[ARG_unsubscribed].u_obj)) || (MP_OBJ_IS_METH(args[ARG_unsubscribed].u_obj))) {
self->mpy_unsubscribed_cb = args[ARG_unsubscribed].u_obj;
}
if ((MP_OBJ_IS_FUN(args[ARG_published].u_obj)) || (MP_OBJ_IS_METH(args[ARG_published].u_obj))) {
self->mpy_published_cb = args[ARG_published].u_obj;
}
self->base.type = &mqtt_type;
self->client = esp_mqtt_client_init(&mqtt_cfg);
if (self->client == NULL) {
mp_raise_ValueError("Error initializing mqtt client");
}
self->client->mpy_mqtt_obj = self;
//esp_mqtt_client_start(self->client);
return MP_OBJ_FROM_PTR(self);
}
//------------------------------------------------------------------------------------------
STATIC mp_obj_t mqtt_op_config(mp_uint_t n_args, const mp_obj_t *pos_args, mp_map_t *kw_args)
{
enum { ARG_server, ARG_user, ARG_pass, ARG_port, ARG_reconnect, ARG_clientid, ARG_cleansess, ARG_keepalive, ARG_lwt_topic, ARG_lwt_msg,
ARG_lwt_qos, ARG_lwt_retain, ARG_datacb, ARG_connected, ARG_disconnected, ARG_subscribed, ARG_unsubscribed, ARG_published };
const mp_arg_t mqtt_config_allowed_args[] = {
{ MP_QSTR_server, MP_ARG_KW_ONLY | MP_ARG_OBJ, {.u_obj = mp_const_none} },
{ MP_QSTR_user, MP_ARG_KW_ONLY | MP_ARG_OBJ, {.u_obj = mp_const_none} },
{ MP_QSTR_password, MP_ARG_KW_ONLY | MP_ARG_OBJ, {.u_obj = mp_const_none} },
{ MP_QSTR_port, MP_ARG_KW_ONLY | MP_ARG_INT, {.u_int = -1} },
{ MP_QSTR_autoreconnect, MP_ARG_KW_ONLY | MP_ARG_INT, {.u_int = -1} },
{ MP_QSTR_clientid, MP_ARG_KW_ONLY | MP_ARG_OBJ, {.u_obj = mp_const_none} },
{ MP_QSTR_cleansession, MP_ARG_KW_ONLY | MP_ARG_INT, {.u_int = -1} },
{ MP_QSTR_keepalive, MP_ARG_KW_ONLY | MP_ARG_INT, {.u_int = -1} },
{ MP_QSTR_lwt_topic, MP_ARG_KW_ONLY | MP_ARG_OBJ, {.u_obj = mp_const_none} },
{ MP_QSTR_lwt_msg, MP_ARG_KW_ONLY | MP_ARG_OBJ, {.u_obj = mp_const_none} },
{ MP_QSTR_lwt_qos, MP_ARG_KW_ONLY | MP_ARG_INT, {.u_int = 0} },
{ MP_QSTR_lwt_retain, MP_ARG_KW_ONLY | MP_ARG_INT, {.u_int = 0} },
{ MP_QSTR_data_cb, MP_ARG_KW_ONLY | MP_ARG_OBJ, {.u_obj = mp_const_none} },
{ MP_QSTR_connected_cb, MP_ARG_KW_ONLY | MP_ARG_OBJ, {.u_obj = mp_const_none} },
{ MP_QSTR_disconnected_cb, MP_ARG_KW_ONLY | MP_ARG_OBJ, {.u_obj = mp_const_none} },
{ MP_QSTR_subscribed_cb, MP_ARG_KW_ONLY | MP_ARG_OBJ, {.u_obj = mp_const_none} },
{ MP_QSTR_unsubscribed_cb, MP_ARG_KW_ONLY | MP_ARG_OBJ, {.u_obj = mp_const_none} },
{ MP_QSTR_published_cb, MP_ARG_KW_ONLY | MP_ARG_OBJ, {.u_obj = mp_const_none} },
};
mqtt_obj_t *self = pos_args[0];
if (!self->client) {
mp_raise_ValueError("Destroyed");
}
if (self->client->state < MQTT_STATE_INIT) {
mp_raise_ValueError("Not initialized");
}
mp_arg_val_t args[MP_ARRAY_SIZE(mqtt_config_allowed_args)];
mp_arg_parse_all(n_args-1, pos_args+1, kw_args, MP_ARRAY_SIZE(mqtt_config_allowed_args), mqtt_config_allowed_args, args);
const char *tstr = NULL;
bool has_conn_arg = false;
if ((MP_OBJ_IS_STR(args[ARG_server].u_obj)) || (args[ARG_port].u_int > 0) || (MP_OBJ_IS_STR(args[ARG_user].u_obj)) ||
(MP_OBJ_IS_STR(args[ARG_pass].u_obj)) || (MP_OBJ_IS_STR(args[ARG_clientid].u_obj)) ||
(args[ARG_reconnect].u_int >= 0) || (args[ARG_keepalive].u_int > 0) || (args[ARG_cleansess].u_int >= 0) ||
(MP_OBJ_IS_STR(args[ARG_lwt_topic].u_obj)) ) has_conn_arg = true;
if ((has_conn_arg) && (self->client->state < MQTT_STATE_CONNECTED)) {
// not started, we can change all parameters
// URI
if (MP_OBJ_IS_STR(args[ARG_server].u_obj)) {
tstr = mp_obj_str_get_str(args[ARG_server].u_obj);
if (strlen(tstr) >= MQTT_MAX_HOST_LEN) {
mp_raise_ValueError("URI too long");
}
sprintf(self->client->config->uri, "%s", tstr);
}
// Port
if (args[ARG_port].u_int > 0) self->client->config->port = args[ARG_port].u_int;
// User name
if (MP_OBJ_IS_STR(args[ARG_user].u_obj)) {
tstr = mp_obj_str_get_str(args[ARG_user].u_obj);
if (strlen(tstr) >= MQTT_MAX_USERNAME_LEN) {
mp_raise_ValueError("User name too long");
}
sprintf(self->client->connect_info.username, "%s", tstr);
}
// Password
if (MP_OBJ_IS_STR(args[ARG_pass].u_obj)) {
tstr = mp_obj_str_get_str(args[ARG_pass].u_obj);
if (strlen(tstr) >= MQTT_MAX_PASSWORD_LEN) {
mp_raise_ValueError("Password too long");
}
sprintf(self->client->connect_info.password, "%s", tstr);
}
// Client ID
if (MP_OBJ_IS_STR(args[ARG_clientid].u_obj)) {
tstr = mp_obj_str_get_str(args[ARG_clientid].u_obj);
if (strlen(tstr) >= MQTT_MAX_CLIENT_LEN) {
mp_raise_ValueError("Client ID too long");
}
sprintf(self->client->connect_info.client_id, "%s", tstr);
}
else sprintf(self->client->connect_info.client_id, "mpy_mqtt_client");
if (args[ARG_reconnect].u_int >= 0) self->client->config->auto_reconnect = args[ARG_reconnect].u_int ? true : false;
if (args[ARG_keepalive].u_int > 0) self->client->connect_info.keepalive = args[ARG_keepalive].u_int;
if (args[ARG_cleansess].u_int >= 0) self->client->connect_info.clean_session = args[ARG_cleansess].u_int ? true : false;
// LWT options
if (MP_OBJ_IS_STR(args[ARG_lwt_topic].u_obj)) {
tstr = mp_obj_str_get_str(args[ARG_lwt_topic].u_obj);
if (strlen(tstr) >= MQTT_MAX_LWT_TOPIC) {
mp_raise_ValueError("LWT topic too long");
}
sprintf(self->client->connect_info.will_topic, "%s", tstr);
if (MP_OBJ_IS_STR(args[ARG_lwt_msg].u_obj)) {
tstr = mp_obj_str_get_str(args[ARG_lwt_msg].u_obj);
if (strlen(tstr) >= MQTT_MAX_LWT_MSG) {
mp_raise_ValueError("LWT message too long");
}
sprintf(self->client->connect_info.will_topic, "%s", tstr);
}
if (args[ARG_lwt_qos].u_int >= 0) self->client->connect_info.will_qos = args[ARG_lwt_qos].u_int;
if (args[ARG_lwt_retain].u_int >= 0) self->client->connect_info.will_retain = args[ARG_lwt_retain].u_int;
}
}
// Callbacks
if ((MP_OBJ_IS_FUN(args[ARG_datacb].u_obj)) || (MP_OBJ_IS_METH(args[ARG_datacb].u_obj))) {
self->mpy_data_cb = NULL;
self->mpy_data_cb = args[ARG_datacb].u_obj;
}
else if (args[ARG_datacb].u_obj == mp_const_false) self->mpy_data_cb = NULL;
if ((MP_OBJ_IS_FUN(args[ARG_connected].u_obj)) || (MP_OBJ_IS_METH(args[ARG_connected].u_obj))) {
self->mpy_connected_cb = NULL;
self->mpy_connected_cb = args[ARG_connected].u_obj;
}
else if (args[ARG_connected].u_obj == mp_const_false) self->mpy_connected_cb = NULL;
if ((MP_OBJ_IS_FUN(args[ARG_disconnected].u_obj)) || (MP_OBJ_IS_METH(args[ARG_disconnected].u_obj))) {
self->mpy_disconnected_cb = NULL;
self->mpy_disconnected_cb = args[ARG_disconnected].u_obj;
}
else if (args[ARG_disconnected].u_obj == mp_const_false) self->mpy_disconnected_cb = NULL;
if ((MP_OBJ_IS_FUN(args[ARG_subscribed].u_obj)) || (MP_OBJ_IS_METH(args[ARG_subscribed].u_obj))) {
self->mpy_subscribed_cb = NULL;
self->mpy_subscribed_cb = args[ARG_subscribed].u_obj;
}
else if (args[ARG_subscribed].u_obj == mp_const_false) self->mpy_subscribed_cb = NULL;
if ((MP_OBJ_IS_FUN(args[ARG_unsubscribed].u_obj)) || (MP_OBJ_IS_METH(args[ARG_unsubscribed].u_obj))) {
self->mpy_unsubscribed_cb = NULL;
self->mpy_unsubscribed_cb = args[ARG_unsubscribed].u_obj;
}
else if (args[ARG_unsubscribed].u_obj == mp_const_false) self->mpy_unsubscribed_cb = NULL;
if ((MP_OBJ_IS_FUN(args[ARG_published].u_obj)) || (MP_OBJ_IS_METH(args[ARG_published].u_obj))) {
self->mpy_published_cb = NULL;
self->mpy_published_cb = args[ARG_published].u_obj;
}
else if (args[ARG_published].u_obj == mp_const_false) self->mpy_published_cb = NULL;
return mp_const_none;
}
STATIC MP_DEFINE_CONST_FUN_OBJ_KW(mqtt_config_obj, 1, mqtt_op_config);
//-----------------------------------------------------------------------
STATIC mp_obj_t mqtt_op_subscribe(mp_uint_t n_args, const mp_obj_t *args)
{
mqtt_obj_t *self = args[0];
if (checkClient(self) != MQTT_STATE_CONNECTED) return mp_const_false;
const char *topic = mp_obj_str_get_str(args[1]);
int wait = 2000;
int qos = 0;
if (n_args == 3) {
qos = mp_obj_get_int(args[2]);
if ((qos < 0) || (qos > 2)) {
mp_raise_ValueError("Wrong QoS value");
}
}
self->subs_flag = 0;
self->client->config->user_context = (void *)topic;
int res = esp_mqtt_client_subscribe(self->client, topic, qos);
if (res < 0) {
self->client->config->user_context = NULL;
return mp_const_false;
}
while ((wait > 0) && (self->subs_flag == 0)) {
vTaskDelay(10 / portTICK_PERIOD_MS);
wait -= 10;
}
self->client->config->user_context = NULL;
if (wait) return mp_const_true;
else return mp_const_false;
}
MP_DEFINE_CONST_FUN_OBJ_VAR_BETWEEN(mqtt_subscribe_obj, 2, 3, mqtt_op_subscribe);
//----------------------------------------------------------------------
STATIC mp_obj_t mqtt_op_unsubscribe(mp_obj_t self_in, mp_obj_t topic_in)
{
mqtt_obj_t *self = self_in;
if (checkClient(self) != MQTT_STATE_CONNECTED) return mp_const_false;
const char *topic = mp_obj_str_get_str(topic_in);
int wait = 2000;
self->unsubs_flag = 0;
self->client->config->user_context = (void *)topic;
int res = esp_mqtt_client_unsubscribe(self->client, topic);
if (res < 0) {
self->client->config->user_context = NULL;
return mp_const_false;
}
while ((wait > 0) && (self->unsubs_flag == 0)) {
vTaskDelay(10 / portTICK_PERIOD_MS);
wait -= 10;
}
self->client->config->user_context = NULL;
if (wait) return mp_const_true;
else return mp_const_false;
}
MP_DEFINE_CONST_FUN_OBJ_2(mqtt_unsubscribe_obj, mqtt_op_unsubscribe);
//---------------------------------------------------------------------
STATIC mp_obj_t mqtt_op_publish(mp_uint_t n_args, const mp_obj_t *args)
{
mqtt_obj_t *self = args[0];
if (checkClient(self) != MQTT_STATE_CONNECTED) return mp_const_false;
size_t len;
const char *topic = mp_obj_str_get_str(args[1]);
const char *msg = mp_obj_str_get_data(args[2], &len);
int wait = 2000;
int qos = 0;
if (n_args == 4) {
qos = mp_obj_get_int(args[3]);
if ((qos < 0) || (qos > 2)) {
mp_raise_ValueError("Wrong QoS value");
}
}
if (qos == 0) wait = 0;
int retain = 0;
if (n_args == 5) retain = mp_obj_is_true(args[4]);
self->publish_flag = 0;
self->client->config->user_context = (void *)topic;
int res = esp_mqtt_client_publish(self->client, topic, msg, len, qos, retain);
if (res < 0) {
self->client->config->user_context = NULL;
return mp_const_false;
}
while ((wait > 0) && (self->publish_flag == 0)) {
vTaskDelay(10 / portTICK_PERIOD_MS);
wait -= 10;
}
self->client->config->user_context = NULL;
return mp_const_true;
}
MP_DEFINE_CONST_FUN_OBJ_VAR_BETWEEN(mqtt_publish_obj, 3, 5, mqtt_op_publish);
//----------------------------------------------
STATIC mp_obj_t mqtt_op_status(mp_obj_t self_in)
{
mqtt_obj_t *self = self_in;
char sstat[16];
mp_obj_t tuple[2];
if (self->client == NULL) {
tuple[0] = mp_obj_new_int(-1);
sprintf(sstat, "Destroyed");
}
else {
tuple[0] = mp_obj_new_int(self->client->state);
if (self->client->state == MQTT_STATE_CONNECTED) sprintf(sstat, "Connected");
else if (self->client->state == MQTT_STATE_INIT) sprintf(sstat, "Initialized");
else if (self->client->state == MQTT_STATE_WAIT_TIMEOUT) sprintf(sstat, "Wait timeout");
else if (self->client->state == MQTT_STATE_UNKNOWN) sprintf(sstat, "Unknown");
else sprintf(sstat, "Error");
}
tuple[1] = mp_obj_new_str(sstat, strlen(sstat));
return mp_obj_new_tuple(2, tuple);
}
MP_DEFINE_CONST_FUN_OBJ_1(mqtt_status_obj, mqtt_op_status);
//--------------------------------------------
STATIC mp_obj_t mqtt_op_stop(mp_obj_t self_in)
{
mqtt_obj_t *self = self_in;
if ((self->client) && (self->client->state >= MQTT_STATE_INIT)) {
esp_mqtt_client_stop(self->client);
int status = 0;
while ((status < 20) && ((xEventGroupGetBits(self->client->status_bits) & 1) == 0)) {
vTaskDelay(100 / portTICK_RATE_MS);
}
}
return mp_const_none;
}
MP_DEFINE_CONST_FUN_OBJ_1(mqtt_stop_obj, mqtt_op_stop);
//--------------------------------------------
STATIC mp_obj_t mqtt_op_start(mp_obj_t self_in)
{
mqtt_obj_t *self = self_in;
if ((self->client) && (self->client->state < MQTT_STATE_INIT)) {
int res = esp_mqtt_client_start(self->client);
if (res != ESP_OK) {
nlr_raise(mp_obj_new_exception_msg(&mp_type_TypeError, "Error starting client"));
}
}
return mp_const_none;
}
MP_DEFINE_CONST_FUN_OBJ_1(mqtt_start_obj, mqtt_op_start);
//--------------------------------------------
STATIC mp_obj_t mqtt_op_free(mp_obj_t self_in)
{
mqtt_obj_t *self = self_in;
if (self->client) {
self->mpy_data_cb = NULL;
self->mpy_connected_cb = NULL;
self->mpy_disconnected_cb = NULL;
self->mpy_subscribed_cb = NULL;
self->mpy_unsubscribed_cb = NULL;
self->mpy_published_cb = NULL;
esp_mqtt_client_destroy(self->client);
self->client = NULL;
if (self->msgbuf) {
free(self->msgbuf);
self->msgbuf = NULL;
}
if (self->topicbuf) {
free(self->topicbuf);
self->topicbuf = NULL;
}
if (self->certbuf) {
free(self->certbuf);
self->certbuf = NULL;
}
return mp_const_true;
}
return mp_const_false;
}
MP_DEFINE_CONST_FUN_OBJ_1(mqtt_free_obj, mqtt_op_free);
//=========================================================
STATIC const mp_rom_map_elem_t mqtt_locals_dict_table[] = {
{ MP_ROM_QSTR(MP_QSTR_config), (mp_obj_t)&mqtt_config_obj },
{ MP_ROM_QSTR(MP_QSTR_subscribe), (mp_obj_t)&mqtt_subscribe_obj },
{ MP_ROM_QSTR(MP_QSTR_unsubscribe), (mp_obj_t)&mqtt_unsubscribe_obj },
{ MP_ROM_QSTR(MP_QSTR_publish), (mp_obj_t)&mqtt_publish_obj },
{ MP_ROM_QSTR(MP_QSTR_status), (mp_obj_t)&mqtt_status_obj },
{ MP_ROM_QSTR(MP_QSTR_stop), (mp_obj_t)&mqtt_stop_obj },
{ MP_ROM_QSTR(MP_QSTR_start), (mp_obj_t)&mqtt_start_obj },
{ MP_ROM_QSTR(MP_QSTR_free), (mp_obj_t)&mqtt_free_obj },
};
STATIC MP_DEFINE_CONST_DICT(mqtt_locals_dict, mqtt_locals_dict_table);
//===============================
const mp_obj_type_t mqtt_type = {
{ &mp_type_type },
.name = MP_QSTR_Mqtt,
.print = mqtt_print,
.make_new = mqtt_make_new,
.locals_dict = (mp_obj_dict_t*)&mqtt_locals_dict,
};
#endif
Reference in New Issue View Git Blame Copy Permalink