Introduction to MQTT

SIM900 enables GPRS for embedded applications. We can implement MQTT (Message Queue Telemetry Transport) Client protocol using SIM900 TCP function AT Commands.

MQTT is a lightweight publish-subscribe-based messaging protocol.

• It is quicker (faster) than other request-response based APIs like HTTP.
• It is developed on the base of the TCP/IP protocol.
• It allows remote location devices to connect, subscribe, publish, etc. to a specific topic on the server with the help of a message broker.
• MQTT Broker/Message broker is a module in between the sender and the receiver. It is an element for message validation, transformation, and routing.
• The broker is responsible for distributing messages to the interested clients (subscribed clients) of their interested topic.

• For example, if the temperature sensor publishes the temperature data (message) on the topic “temperature” then interested clients who have subscribed to the “temperature” topic get that published temperature data as shown in the above figure.

MQTT is widely used in IoT (Internet of Things) embedded applications, where every sensor is connected to a server and we have access to control them over the internet.

To know about SIM900 GSM/GPRS Module refer to SIM900

Connection Diagram of SIM900 GSM Module with PIC18F4550

MQTT Client over GPRS

Let’s program PIC18F4550 to configure SIM900A as MQTT Client and Subscribe/Publish data from/to Server using GPRS.

Here we are using the Adafruit server for MQTT Client demo purpose.

In the IOT platform, Adafruit IO Dashboard allows us to visualize and provides control over the connected device to the internet. Anyone can visualize and analyze live data from their sensor devices. To learn more and start with Adafruit IO Dashboard refer link https://learn.adafruit.com/adafruit-io-basics-dashboards/creating-a-dashboard

Just sign up and create a dashboard. After the successful creating of the dashboard, we will get the AIO key which is later used to access feed data.

Example

Now let’s program PIC18F4550 to control LED brightness and monitor POT status on a remote location from the Adafruit dashboard.

Once we created a dashboard on Adafruit we can add various blocks that can be used to control devices as well as monitor the status of devices. To see more about blocks, refer link https://learn.adafruit.com/adafruit-io-basics-dashboards/adding-blocks

In the below program of MQTT Client, do the following

For MQTT Client Subscribe Demo

#define SUBSRCIBE_DEMO				/* Define SUBSRCIBE demo */
//#define PUBLISH_DEMO				/* Define PUBLISH demo */

For MQTT Client Publish Demo

//#define SUBSRCIBE_DEMO			/* Define SUBSRCIBE demo */
#define PUBLISH_DEMO				/* Define PUBLISH demo */

Edit Fields below with respective data

/* Define Required fields shown below */
#define AIO_SERVER		"io.adafruit.com"	/* Adafruit server */
#define AIO_SERVER_PORT		"1883"			/* Server port */
#define AIO_BASE_URL		"/api/v2"		/* Base URL for api */
#define AIO_USERNAME		"Enter Username"	/* Enter username here */
#define AIO_KEY			"Enter AIO key"		/* Enter AIO key here */
#define AIO_FEED		"Enter Feed Key"	/* Enter feed key */

#define APN			"internet"		/* APN of n/w service provider */
#define USERNAME		""
#define PASSWORD		""

In the below program, we are using response-based functions to get the better status if things deviate from normal.

MQTT Packet Formation

MQTT uses many packet formats that used to connect to the server and subscribe or publish to the topic on the server.

Refer below link for MQTT OASIS standard. It will help to understand MQTT packet formations.

http://docs.oasis-open.org/mqtt/mqtt/v3.1.1/os/mqtt-v3.1.1-os.html#_Toc398718027

MQTT Client Code for PIC18F4550

/*
* PIC_GPRS_MQTTClient
* http://www.electronicwings.com
*
*/


#include "SIM900TCPClient.h"		/* Include TCP Client header file */
#include "math.h"

#define MQTT_PROTOCOL_LEVEL		4

#define MQTT_CTRL_CONNECT		0x1
#define MQTT_CTRL_CONNECTACK		0x2
#define MQTT_CTRL_PUBLISH		0x3
#define MQTT_CTRL_PUBACK		0x4
#define MQTT_CTRL_PUBREC		0x5
#define MQTT_CTRL_PUBREL		0x6
#define MQTT_CTRL_PUBCOMP		0x7
#define MQTT_CTRL_SUBSCRIBE		0x8
#define MQTT_CTRL_SUBACK		0x9
#define MQTT_CTRL_UNSUBSCRIBE		0xA
#define MQTT_CTRL_UNSUBACK		0xB
#define MQTT_CTRL_PINGREQ		0xC
#define MQTT_CTRL_PINGRESP		0xD
#define MQTT_CTRL_DISCONNECT		0xE

#define MQTT_QOS_1			0x1
#define MQTT_QOS_0			0x0

/* Adjust as necessary, in seconds */
#define MQTT_CONN_KEEPALIVE		60

#define MQTT_CONN_USERNAMEFLAG		0x80
#define MQTT_CONN_PASSWORDFLAG		0x40
#define MQTT_CONN_WILLRETAIN		0x20
#define MQTT_CONN_WILLQOS_1		0x08
#define MQTT_CONN_WILLQOS_2		0x18
#define MQTT_CONN_WILLFLAG		0x04
#define MQTT_CONN_CLEANSESSION		0x02

#define DEFAULT_BUFFER_SIZE		200
#define DEFAULT_TIMEOUT			10000
#define DEFAULT_CRLF_COUNT		2

#define MINTHR              8000
#define RESOLUTION          488

#define InternalOsc_8MHz    8000000
#define InternalOsc_4MHz    4000000
#define InternalOsc_2MHz    2000000
#define InternalOsc_1MHz    1000000
#define InternalOsc_500KHz  500000
#define InternalOsc_250KHz  250000
#define InternalOsc_125KHz  125000
#define InternalOsc_31KHz   31000

#define Timer2Prescale_1    1
#define Timer2Prescale_4    4
#define Timer2Prescale_16   16

/* Select Demo */
#define SUBSRCIBE_DEMO				/* Define SUBSRCIBE demo */
//#define PUBLISH_DEMO				/* Define PUBLISH demo */

#define AIO_SERVER		"io.adafruit.com"	/* Adafruit server */
#define AIO_SERVER_PORT		"1883"			/* Server port */
#define AIO_BASE_URL		"/api/v2"		/* Base URL for api */
#define AIO_USERNAME		"Enter Username"	/* Enter username here */
#define AIO_KEY			"Enter AIO key"		/* Enter AIO key here */
#define AIO_FEED		"Enter Feed Key"	/* Enter feed key */

#define APN			"internet"
#define USERNAME		""
#define PASSWORD		""

int16_t packet_id_counter = 0;


char clientID[] ="";
char will_topic[] = "";
char will_payload[] ="";
uint8_t will_qos = 1;
uint8_t will_retain = 0;

uint16_t StringToUint16(uint8_t* _String)
{
	uint16_t value = 0;
	while ('0' == *_String || *_String == ' ' || *_String == '"')
	{
		_String++;
	}
	while ('0' <= *_String && *_String <= '9')
	{
		value *= 10;
		value += *_String - '0';
		_String++;
	}
	return value;
}

uint8_t* AddStringToBuf(uint8_t *_buf, const char *_string)
{
	uint16_t _length = strlen(_string);
	_buf[0] = _length >> 8;
	_buf[1] = _length & 0xFF;
	_buf+=2;
	strncpy((char *)_buf, _string, _length);
	return _buf + _length;
}

bool sendPacket(uint8_t *buffer, uint16_t len)
{
	uint16_t sendlen;
	if (TCPClient_connected())
	{
		sendlen = min(len, 250);
		TCPClient_Send((char*)buffer, sendlen);
		return true;
	} 
	else
	return false;
}

uint16_t readPacket(uint8_t *buffer, int16_t _timeout) 
{
	uint16_t len = 0;
	while (TCPClient_DataAvailable() > 0 && _timeout >=0) 
	{
		buffer[len++] = TCPClient_DataRead();
		_timeout-= 10;	MSdelay(10);
	}
	buffer[len] = 0;
	return len;
}

uint8_t MQTT_ConnectToServer()
{
	return TCPClient_Start(1, AIO_SERVER, AIO_SERVER_PORT);
}

uint16_t MQTT_connectpacket(uint8_t* packet)
{
	uint8_t*_packet = packet;
	uint16_t _length;

	_packet[0] = (MQTT_CTRL_CONNECT << 4);
	_packet+=2;
	_packet = AddStringToBuf(_packet, "MQTT");
	_packet[0] = MQTT_PROTOCOL_LEVEL;
	_packet++;
	
	_packet[0] = MQTT_CONN_CLEANSESSION;
	if (will_topic && strlen(will_topic) != 0) {
		_packet[0] |= MQTT_CONN_WILLFLAG;
		if(will_qos == 1)
		_packet[0] |= MQTT_CONN_WILLQOS_1;
		else if(will_qos == 2)
		_packet[0] |= MQTT_CONN_WILLQOS_2;
		if(will_retain == 1)
		_packet[0] |= MQTT_CONN_WILLRETAIN;
	}
	if (strlen(AIO_USERNAME) != 0)
	_packet[0] |= MQTT_CONN_USERNAMEFLAG;
	if (strlen(AIO_KEY) != 0)
	_packet[0] |= MQTT_CONN_PASSWORDFLAG;
	_packet++;

	_packet[0] = MQTT_CONN_KEEPALIVE >> 8;
	_packet++;
	_packet[0] = MQTT_CONN_KEEPALIVE & 0xFF;
	_packet++;
	if (strlen(clientID) != 0) {
		_packet = AddStringToBuf(_packet, clientID);
		} else {
		_packet[0] = 0x0;
		_packet++;
		_packet[0] = 0x0;
		_packet++;
	}
	if (will_topic && strlen(will_topic) != 0) {
		_packet = AddStringToBuf(_packet, will_topic);
		_packet = AddStringToBuf(_packet, will_payload);
	}

	if (strlen(AIO_USERNAME) != 0) {
		_packet = AddStringToBuf(_packet, AIO_USERNAME);
	}
	if (strlen(AIO_KEY) != 0) {
		_packet = AddStringToBuf(_packet, AIO_KEY);
	}
	_length = _packet - packet;
	packet[1] = _length-2;

	return _length;
}

uint16_t MQTT_publishPacket(uint8_t *packet, const char *topic, char *data, uint8_t qos)
{
	uint8_t *_packet = packet;
	uint16_t _length = 0;
	uint16_t Datalen=strlen(data);

	_length += 2;
	_length += strlen(topic);
	if(qos > 0) {
		_length += 2;
	}
	_length += Datalen;
	_packet[0] = MQTT_CTRL_PUBLISH << 4 | qos << 1;
	_packet++;
	do {
		uint8_t encodedByte = _length % 128;
		_length /= 128;
		if ( _length > 0 ) {
			encodedByte |= 0x80;
		}
		_packet[0] = encodedByte;
		_packet++;
	} while ( _length > 0 );
	_packet = AddStringToBuf(_packet, topic);
	if(qos > 0) {
		_packet[0] = (packet_id_counter >> 8) & 0xFF;
		_packet[1] = packet_id_counter & 0xFF;
		_packet+=2;
		packet_id_counter++;
	}
	memmove(_packet, data, Datalen);
	_packet+= Datalen;
	_length = _packet - packet;

	return _length;
}

uint16_t MQTT_subscribePacket(uint8_t *packet, const char *topic, uint8_t qos) 
{
	uint8_t *_packet = packet;
	uint16_t _length;

	_packet[0] = MQTT_CTRL_SUBSCRIBE << 4 | MQTT_QOS_1 << 1;
	_packet+=2;

	_packet[0] = (packet_id_counter >> 8) & 0xFF;
	_packet[1] = packet_id_counter & 0xFF;
	_packet+=2;
	packet_id_counter++;

	_packet = AddStringToBuf(_packet, topic);

	_packet[0] = qos;
	_packet++;

	_length = _packet - packet;
	packet[1] = _length-2;

	return _length;
}

void PWM_Init()			/* Initialize PWM */
{
	TRISCbits.TRISC1 = 0;   /* Set CCP2 pin as output for PWM out */
	CCP2CON = 0x0C;         /* Set PWM mode */
}

int setPeriodTo(unsigned long FPWM)/* Set period */
{
    int clockSelectBits, TimerPrescaleBits;
    int TimerPrescaleValue;
    float period;
    unsigned long FOSC, _resolution = RESOLUTION;

    if (FPWM < MINTHR)    {TimerPrescaleBits = 2; TimerPrescaleValue = Timer2Prescale_16;}
    else                  {TimerPrescaleBits = 0; TimerPrescaleValue = Timer2Prescale_1;}

    if (FPWM > _resolution)               {clockSelectBits = 7; FOSC = InternalOsc_8MHz;}
    else if (FPWM > (_resolution >>= 1))  {clockSelectBits = 6; FOSC = InternalOsc_4MHz;}
    else if (FPWM > (_resolution >>= 1))  {clockSelectBits = 5; FOSC = InternalOsc_2MHz;}
    else if (FPWM > (_resolution >>= 1))  {clockSelectBits = 4; FOSC = InternalOsc_1MHz;}
    else if (FPWM > (_resolution >>= 1))  {clockSelectBits = 3; FOSC = InternalOsc_500KHz;}
    else if (FPWM > (_resolution >>= 1))  {clockSelectBits = 2; FOSC = InternalOsc_250KHz;}
    else if (FPWM > (_resolution >>= 1))  {clockSelectBits = 1; FOSC = InternalOsc_125KHz;}
    else                                  {clockSelectBits = 0; FOSC = InternalOsc_31KHz;}

    period = ((float)FOSC / (4.0 * (float)TimerPrescaleValue * (float)FPWM)) - 1.0;
    period = round(period);

    OSCCON = ((clockSelectBits & 0x07) << 4) | 0x02;
    PR2 = (int)period;
    T2CON = TimerPrescaleBits;
    TMR2 = 0;
    T2CONbits.TMR2ON = 1;  /* Turn ON Timer2 */
    return (int)period;
}

void SetDutyCycleTo(float Duty_cycle, int Period)/* Set Duty cycle for given period */
{
    int PWM10BitValue;
    
    PWM10BitValue = 4.0 * ((float)Period + 1.0) * (Duty_cycle/100.0);
    CCPR2L = (PWM10BitValue >> 2);
    CCP2CON = ((PWM10BitValue & 0x03) << 4) | 0x0C;
}

void ADC_Init()                 /* ADC initialization function */
{    
    TRISA = 0xFF;               /* Set as input port */
    ADCON1 = 0x07;              /* Set AN0-7 as Analog & other as digital */
    ADCON2 = 0x92;              /* Right Justified, 4Tad and Fosc/32. */
    ADRESH = 0;                 /* Flush ADC output Register */
    ADRESL = 0;   
}

int ADC_Read(char channel)      /* ADC channel read function */
{
    ADCON0 =(ADCON0 & 0b11000011)|((channel<<2) & 0b00111100); /* Channel is selected i.e (CHS3:CHS0) */
    ADCON0 |= ((1<<ADON)|(1<<GO)); /* Enable ADC and start conversion */
    while(ADCON0bits.GO_nDONE==1); /* wait for End of conversion i.e. Go/done'=0 conversion completed */        
    return((ADRESH*256) | (ADRESL));/* Return 10-bit ADC result */
}

int main()
{
	char buf[4];
	uint8_t _buffer[150];
	uint16_t len;
	
	#ifdef SUBSRCIBE_DEMO
	int period;
	long KeepAliveTime;
	PWM_Init();             /* Initialize PWM */
	period = setPeriodTo(10000);/* 10KHz PWM frequency */
	#endif

	#ifdef PUBLISH_DEMO
	int ADC_Value;
	ADC_Init();
	#endif

	OSCCON = 0x72;                              /* Set internal clock to 8MHz */
	USART_Init(9600);							/* Initiate 	USART with 9600 baud rate */
	INTCONbits.GIE=1;                           /* enable Global Interrupt */
	INTCONbits.PEIE=1;                          /* enable Peripheral Interrupt */
	PIE1bits.RCIE=1;                            /* enable Receive Interrupt */	
	
	while(!SIM900_Start());
	TCPClient_Shut();
	TCPClient_ConnectionMode(0);			/* 0 = Single; 1 = Multi */
	TCPClient_ApplicationMode(0);			/* 0 = Normal Mode; 1 = Transperant Mode */
	AttachGPRS();
	while(!(TCPClient_Connect(APN, USERNAME, PASSWORD)));

	while(1)
	{
		MQTT_ConnectToServer();
		len = MQTT_connectpacket(_buffer);
		sendPacket(_buffer, len);
		len = readPacket(_buffer, 1000);

		#ifdef PUBLISH_DEMO
		while(TCPClient_connected())
		{
			memset(_buffer, 0, 150);
			memset(buf, 0, 4);
			ADC_Value = (float)ADC_Read(0)/10.4;
			sprintf(buf, "%d", ADC_Value);	/* Read ADC channel 0 and publish it in range 0-100 */
			len = MQTT_publishPacket(_buffer, "Nivya151/feeds/test", buf, 1);/* topic format: "username/feeds/aio_feed" e.g. "Nivya151/feeds/test" */
			sendPacket(_buffer, len);
		}
		TCPClient_Close();
		MSdelay(1000);
		#endif
		
		#ifdef SUBSRCIBE_DEMO
		uint8_t valuePointer=0;
		memset(_buffer, 0, 150);
		len = MQTT_subscribePacket(_buffer, "Nivya151/feeds/test", 1);/* topic format: "username/feeds/aio_feed" e.g. "Nivya151/feeds/test" */
		sendPacket(_buffer, len);
		KeepAliveTime = (MQTT_CONN_KEEPALIVE * 1000L);
		while(KeepAliveTime > 0)		/* Read subscription packets till Alive time */
		{
			len = readPacket(_buffer, 1000);
			for (uint16_t i=0; i<len;i++)
			{
				for (uint8_t k = 0; k < 4; k++)/* here 4 is aio_feed char length e.g. aio_feed "test" has length of 4 char */
				buf[k] = _buffer[i + k];
				if (strstr(buf, AIO_FEED) != 0)
				{
					valuePointer = i + 4;
					SetDutyCycleTo(StringToUint16(_buffer + valuePointer), period);
					i = len;
				}
			}
			MSdelay(1);
			KeepAliveTime--;
		}
		if(TCPClient_connected()) TCPClient_Close();
		MSdelay(1000);
		#endif
	}
}