Development of Serial/MQTT connectivity for Robustel producst

02.10.19 11:47 AM Comment(s) By peterkelecsenyi

In the last topic written by me - "Robustel SDK", we have covered basics of the development for Robustel products.

Here we will experiment with specific product M1200 and implement serial and later MQTT connectivity for the device. Which describes the most widely used and basic IoT map system.

Sample Serial App

Directories:

------> files ------> sdk.sh

------> uci.xml

------> usi.xml

------> src ------> main.c

------> Makefile

First of all, we start with uci.xml. File which defines parameter for serial connection, which can be configured over graphical user interface over ROS.

<sample_serial desc="configuration of sample serial" group="Services" group_weight="40" menu="Sample Serial" menu_weight="901" oid="1101">

<port type="enum" desc="Port" range="COM1 COM2" default="COM1" range_desc="COM1 COM2" row_help="The number and type of COM ports for different devices are not the same."

oid="2" />

<baud_rate type="enum" desc="Baud Rate" default="115200" range="300 600 1200 2400 4800 9600 19200 38400 57600 115200"

range_desc="300 600 1200 2400 4800 9600 19200 38400 57600 115200" col_width="80" oid="3" />

<data_bits type="enum" desc="Data Bits" default="8" range="7 8" range_desc="7 8" oid="4" />

<stop_bits type="enum" desc="Stop Bits" default="1" range="1 2" range_desc="1 2" oid="5" />

<flow_ctrl type="enum" desc="Flow Control" default="none" range="none hardware software" range_desc="None Hardware Software" oid="7" />

</sample_serial>

Here we have written interface component which defines serial connectivity parameters.

They can be later modified in the ROS interface of our application (located under the "Services" tab in GUI) as you can see in the picture below.

Next one is usi.xml, file which defines system parameters, that can be accessed and modified through all applications:

<sample_serial cli_help="status of sample serial" group="Services" group_weight="40" menu="Sample Serial" menu_weight="901" oid="1100" >

<tx_bytes desc="TX" col_width="100" oid="2" />

<rx_bytes desc="RX" col_width="100" oid="3" />

<data_payload desc="Payload" tab_weight="20" oid="4" />

</sample_serial>

Above we can see defined parameters, which can be accessible over the system.

Most important parameter is "data_payload", since we will access it from the MQTT application and push it to the cloud. In fact, "data_payload" is data received from a serial connection.

The next and core part of our application is obviously the main code for establishing serial connectivity main.c.

#include "librouter.h"

#include "event.h"

#include <termios.h>

#define MAX_FRAME_LENGTH 3000

uint64_t rx_bytes = 0;

uint64_t tx_bytes = 0;

void sig_handler_serial_demo(int signo)

{

switch (signo)

{

case SIGTERM:

log_notice("sdk sample_serial service terminated");

exit(0);

break;

default:

break;

}

static void serial_data_proc(int sock, short ev, void *arg)

{

int32_t wr_len = 0;

static uint8_t recv_Frame[MAX_FRAME_LENGTH] = {0};

static int32_t frame_length = 0;

int32_t rd_length = 0;

int *com_fd = (int *) arg;

if(EV_TIMEOUT & ev)

{

if(frame_length)

{

wr_len = safe_write(*com_fd, recv_Frame,strlen(recv_Frame));

tx_bytes += wr_len;

usi_printf("sample_serial.tx_bytes", "%lldB", tx_bytes);

memset(recv_Frame, 0, MAX_FRAME_LENGTH);

frame_length = 0;

}

else if(EV_READ & ev)

{

rd_length = read(*com_fd, &recv_Frame[frame_length], MAX_FRAME_LENGTH-frame_length);

if(rd_length <= 0)

{

log_warn("read serial port error");

log_err_location();

}

else

{

frame_length += rd_length;

rx_bytes += rd_length;

usi_printf("sample_serial.data_payload","200,tankGas,percents,%s,%%",recv_Frame);

usi_printf("sample_serial.rx_bytes", "%lldB", rx_bytes);

}

int main(int argc, char **argv)

{

uint32_t baud_rate = 0;

uint32_t data_bits = 0;

uint32_t stop_bits = 0;

uint32_t parity = 0;

uint32_t flow_ctrl = 0;

char tty_device[32];

struct timeval tv;

struct event serial_event;

int32_t com_fd = -1;

daemon(0, 0);

log_open("sample_serial");

usi_printf("sample_serial.tx_bytes", "0B");

usi_printf("sample_serial.rx_bytes", "0B");

usi_printf("sample_serial.data_payload","empty");

if(uci_match("sample_serial.enable", "false"))

{

return;

}

if(daemon_check_running("/var/run/sample_serial.pid"))

{

log_warn("sdk sample_serial already running!");

exit(1);

}

log_notice("sdk sample_serial service started");

signal(SIGTERM, sig_handler_serial_demo);

baud_rate = atol(uci_get("sample_serial.baud_rate"));

data_bits = atol(uci_get("sample_serial.data_bits"));

stop_bits = atol(uci_get("sample_serial.stop_bits"));

if(string_matched(uci_get("sample_serial.parity"), "none"))

{

parity = 0;

}

else if(string_matched(uci_get("sample_serial.parity"), "odd"))

{

parity = 1;

}

else if(string_matched(uci_get("sample_serial.parity"), "even"))

{

parity = 2;

}

memset(tty_device, 0, sizeof(tty_device));

if(string_matched(uci_get("sample_serial.port"), "COM1"))

{

snprintf(tty_device, sizeof(tty_device), "/dev/ttyCOM1");

usi_update("sample_serial.port","COM1");

}

else if(string_matched(uci_get("sample_serial.port"), "COM2"))

{

snprintf(tty_device, sizeof(tty_device), "/dev/ttyCOM2");

usi_update("sample_serial.port","COM2");

}

com_fd = serial_port_open_full(tty_device, baud_rate, data_bits, stop_bits, parity, flow_ctrl);

if(com_fd < 0)

{

log_error("open serial port error!\n");

log_err_location();

exit(-1);

}

log_debug("open port:%s\n", tty_device);

tv.tv_sec = 0;

tv.tv_usec = 100000;

event_init();

event_set(&serial_event, com_fd, EV_READ|EV_PERSIST|EV_TIMEOUT, serial_data_proc, &com_fd);

if(event_add(&serial_event, &tv) == -1)

{

exit(1);

}

event_dispatch();

return 0;

}

As we can see the code pretty similar to the general C implementation. The difference is work with our parameters. And the SW location of the serial connection:

uci_get() - to get parameter.

usi_get() - to get parameter.

usi_printf() - to set parameter.

Also, accessing parameters done simply by the name of the application + the name of parameter. As a result, we have:

uci_get("sample_serial.parity")

Worth to mention that the that the input of the serial port is either /dev/ttyCOM2 or /dev/ttyCOM1. Depending on the family of the Robustel product.

One more addition modification, is changing final payload, to make it similar to MQTT payload format.

As you can see, I'm modifying it to "Gas level" parameter, to be able later to add it to the payload of MQTT packet.

Sample MQTT App

Directory structure is identical to previous example. Except we don't have usi.xml. Since there is no system parameters required.

It means, we start with uci.xml

<sample_mqtt desc="configuration of sample mqtt" group="Services" group_weight="40" menu="Sample MQTT" menu_weight="125" export_disable="sample_mqtt.enable==false" oid="1103" >

<enable type="bool" default="false"

desc="Enable" tab="MQTT" tab_weight="0" area="MQTT Settings" oid="1" />

<interval type="int" range="1..3600" default="60" desc="Interval" row_disable="sample_mqtt.enable==false"

row_help="Unit:Second" oid="9" />

</sample_mqtt>

Here we can see the parameters required for the MQTT connectivity within the ranges if needed.

Those will be configured over the ROS interface for each specific cloud/connection.

As in the example below MQTT connection for "Cumulocity" platform

The initial payload, which you can see above is used to register the device in the cloud, if it is already registered nothing will happen. Later on, this payload will be modified by getting data_payload from Sample Serial app every 30 seconds (defined by interval parameter seen above).

And the core of the application main.c:

#include "librouter.h"

#include <mosquitto.h>

#include "mqtt.h"

bool_t connected = true;

int mqtt_publish(struct mosquitto *mosq, const char *topic, int32_t payloadlen,

const char *payload, int qos)

{

int32_t rc;

const char *error_string = NULL;

if (NULL == mosq

|| NULL == topic

|| NULL == payload)

{

log_debug("MQTT Publish error, NULL params\n");

return ERROR;

}

payload = usi_get("sample_serial.data_payload");

log_debug("publish %s\n", payload);

rc = mosquitto_publish(mosq, NULL, topic, strlen(payload), payload, qos, true);

if(MOSQ_ERR_SUCCESS != rc)

{

error_string = mosquitto_strerror(rc);

log_debug("publish failed:%s\n", error_string);

return ERROR;

}

return OK;

}

void mqtt_connect_callback(struct mosquitto *mosq, void *obj, int result)

{

log_debug("MQTT connected successful\n");

connected = true;

}

void mqtt_disconnect_callback(struct mosquitto *mosq, void *obj, int rc)

{

int rec_rc;

connected = false;

if (rc == 0)

{

connected = false;

mosquitto_disconnect (mosq);

mosquitto_loop_stop(mosq, false);

mosquitto_destroy (mosq);

mosquitto_lib_cleanup();

log_debug("Disconnected between client and broker\n");

}

else

{

log_debug("Unexpectedly disconnected, try reconnect\n");

rec_rc = mosquitto_reconnect(mosq);

if (MOSQ_ERR_SUCCESS != rec_rc)

{

log_debug("reconnect error:%s\n", mosquitto_strerror(rec_rc));

connected = false;

}

void mqtt_log_cb(struct mosquitto *mosq, void *obj, int leve_log, const char *string)

{

log_debug("log:%s\n", string);

}

void mqtt_publish_callback(struct mosquitto *mosq, void *obj, int mid)

{

log_debug("publish a message successful\n");

}

void keep_connected(int sock, short ev, void *arg)

{

int rc;

int rc_reconnect;

struct mosquitto *mosq = (struct mosquitto *)arg;

rc = mosquitto_loop(mosq, -1, 1);

if (MOSQ_ERR_SUCCESS != rc)

{

sleep(10);

log_debug("%s, try reconnected\n", mosquitto_strerror(rc));

if (rc == MOSQ_ERR_TLS)

{

mosquitto_loop_write(mosq, 1);

}

mosquitto_reconnect(mosq);

}

void mqtt_callback_set(struct mosquitto *mosq)

{

mosquitto_connect_callback_set(mosq, mqtt_connect_callback);

mosquitto_publish_callback_set(mosq, mqtt_publish_callback);

mosquitto_disconnect_callback_set(mosq, mqtt_disconnect_callback);

mosquitto_log_callback_set(mosq, mqtt_log_cb);

}

struct mosquitto *connect_to_broker(struct mosquitto *mosq)

{

int32_t rc = 0;

uint32_t port;

uint32_t keepalive = 0;

const char *addr = NULL;

const char *sn = NULL;

const char *usr = NULL;

const char *passwd = NULL;

sn = uci_get("devcfg.serial_number");

mosquitto_lib_init();

mosq = mosquitto_new(sn, true, NULL);

if (!mosq)

{

log_debug("Can't initialize Mosquitto library\n");

}

usr = uci_get("sample_mqtt.user");

passwd = uci_get("sample_mqtt.password");

mosquitto_username_pw_set(mosq, usr, passwd);

mqtt_callback_set(mosq); // Connecting to the MQTT server.

addr = uci_get("sample_mqtt.addr");

port = uci_get_int("sample_mqtt.port");

keepalive = uci_get_int("sample_mqtt.keepalive");

log_debug("usr:%s, passwd:%s, addr:%s, port:%d, keepalive:%d",

usr, passwd, addr, port, keepalive);

rc = mosquitto_connect(mosq, addr, port, keepalive);

while (MOSQ_ERR_SUCCESS != rc)

{

log_debug("%s, try reconnected\n", mosquitto_strerror(rc));

rc = mosquitto_reconnect(mosq);

sleep(1);

}

log_debug("MQTT connected");

return mosq;

}

int main(int argc, char **argv)

{

struct mosquitto *mosq = NULL;

const char *payload = NULL;

const char *topic = NULL;

int32_t interval = 0;

int rc;

if (!uci_match("sample_mqtt.enable", "true"))

{

return -1;

}

interval = uci_get_int("sample_mqtt.interval");

payload = "117,15";

topic = uci_get("sample_mqtt.topic");

log_debug("interval:%d, payload:%s, topic:%s", interval,payload, topic);

mosq = connect_to_broker(mosq);

if (NULL != mosq)

{

}

else

{

log_debug("connected to server failed\n");

goto END;

}

log_debug("start to loop");

do{

mosq="connect_to_broker(mosq);"

mqtt_publish(mosq, topic, strlen(payload), payload, 1);

sleep(interval);

}while(0!=1);

END:

mosquitto_destroy(mosq);

mosquitto_lib_cleanup();

return 0;

}

The code is based on the mosquitto library for MQTT connectivity.

Shortly, the code gets parameters from the ROS interface, gets payload from Sample_serial, and pushes it the MQTT server/cloud.

As a result we have two running applications.

First, scanning serial port and stores data to data_payload parameter of sample_serial app.

Second, periodically gets data_payload parameter from sample_serial app.

Builds a MQTT packet and pushes it to the cloud.

More information!

Kern IoT megoldások és rendszerek. 4G LTE NB-IoT LoRa GPS Bluetooth