File content as of revision 24:74f170dfd425:

/*******************************************************************************
 * Copyright (c) 2014, 2015 IBM Corp.
 *
 * All rights reserved. This program and the accompanying materials
 * are made available under the terms of the Eclipse Public License v1.0
 * and Eclipse Distribution License v1.0 which accompany this distribution.
 *
 * The Eclipse Public License is available at
 *    http://www.eclipse.org/legal/epl-v10.html
 * and the Eclipse Distribution License is available at
 *   http://www.eclipse.org/org/documents/edl-v10.php.
 *
 * Contributors:
 *    Ian Craggs - initial API and implementation and/or initial documentation
 *    Ian Craggs - make sure QoS2 processing works, and add device headers
 *******************************************************************************/

 /**
  This is a sample program to illustrate the use of the MQTT Client library
  on the mbed platform.  The Client class requires two classes which mediate
  access to system interfaces for networking and timing.  As long as these two
  classes provide the required public programming interfaces, it does not matter
  what facilities they use underneath. In this program, they use the mbed
  system libraries.

 */

 // change this to 1 to output messages to LCD instead of serial
#define USE_LCD 1

#if USE_LCD
#include "C12832.h"

// the actual pins are defined in mbed_app.json and can be overridden per target
C12832 lcd(LCD_MOSI, LCD_SCK, LCD_MISO, LCD_A0, LCD_NCS);

#define logMessage lcd.cls();lcd.locate(0,0);lcd.printf

#else

#define logMessage printf

#endif

#define MQTTCLIENT_QOS2 0

#include "easy-connect.h" //Ethernet Connection library
#include "MQTTNetwork.h" //MQTT Connection
#include "MQTTmbed.h"  //MQTT Connection
#include "MQTTClient.h" //MQTT Connection
#include "string.h";

#include "mbed.h" 
//#include "platform/mbed_thread.h"
#include "FP.h" //For function pointer usage
#include "MQ2.h" //Smoke and Gas detector library
#include "LM75B.h" //Temperature sensor

#define ON 1 
#define OFF 0

#define PUMP PTC1 //Relay Signal for controlling pump
#define TRIGGER PTB18 //Trigger pin for ultrasonic sensor
#define ECHO PTB19 //Echo pin for ultrasonic sensor
#define MQ2_SIG PTC10 //Smoke and Gas sensor output pin
#define ALARM_SIG PTC8 //Alarm Signal Pin
#define FAN_SIG PTC9 //Fan signal pin

//Global variables to update the ThingSpeak channel fields.
int WATER_LEVEL = 0; 
float CO_level = 0;
float SMOKE_level = 0;
float GAS_level = 0;
float TEMPERATURE = 0;
int FAN_STATE = 0;
int ALARM_STATE = 0;
int PUMP_STATE = 0;

class indicator{ //class to allow easy usage of RGB LED 
    private:
        DigitalOut RED; //LED1
        DigitalOut BLUE; //LED2
        DigitalOut GREEN; //LED3
        Ticker blink; //Ticker to start blinking
        bool red_set; //bool to check if RED is ON
        bool green_set; //bool to check if GREEN is ON
        bool blue_set; //bool to check if BLUE is ON
        
        void blink_effect(){ //function to flip LEDs to generate blink effect
            if(red_set)
                RED = !RED;
            if(green_set)
                GREEN = !GREEN;
            if(blue_set)
                BLUE = !BLUE;
        }
 
    public:
         indicator(PinName _LED1, PinName _LED2, PinName _LED3):RED(_LED1, 1), GREEN(_LED2, 1), BLUE(_LED3, 1){ //Initialize RGB LEDs with requried PINS.
             red_set = false; green_set = false; blue_set = false; //All LED intially OFF.
        }
        
        void red_led(){ //Set Only RED LED
            RED = 0; red_set = true; 
            GREEN = 1; green_set = false;
            BLUE = 1; blue_set = false;
        }
        
        void green_led(){ //Set Only GREEN LED
            RED = 1; red_set = false;
            GREEN = 0; green_set = true;
            BLUE = 1; blue_set = false;
        }
        void blue_led(){//Set Only BLUE LED
            RED = 1; red_set = false;
            GREEN = 1; green_set = false;
            BLUE = 0; blue_set = true;
        }
        void amber_led(){//Set Only AMBER LED
            RED = 0; red_set = true;
            GREEN = 0; green_set = true;
            BLUE = 1; blue_set = false;
        }
        void pink_led(){//Set Only PINK LED
            RED = 0; red_set = true;
            GREEN = 1; green_set = false;
            BLUE = 0; blue_set = true;
        }
        void cyan_led(){//Set Only CYAN LED
            RED = 1; red_set = false;
            GREEN = 0; green_set = true;
            BLUE = 0; blue_set = true;
        }
        void white_led(){//Set Only WHITE LED
            RED = 0; red_set = true;
            GREEN = 0; green_set = true;
            BLUE = 0; blue_set = true;
        }
        void all_led_off(){//Set All LED OFF
            RED = 1; red_set = false;
            GREEN = 1; green_set = false;
            BLUE = 1; blue_set = false;
        }
        void start_blink(float blink_interval){ //Start blinking on current LED using tikcer to avoid code blocking
            blink.attach(callback(this, &indicator::blink_effect), blink_interval);
        }
        void stop_blink(){ //Stop blinkink. Detach Ticker function call
            blink.detach();
            if(red_set) //Set last LED being used as ON.
                RED = 0;
            if(green_set)
                GREEN = 0;
            if(blue_set)
                BLUE = 0;        
        }
};

indicator led_indicator(LED1, LED2, LED3); //Use K64f RGB LED for water pump
indicator shield_led_indicator(D5, D9, D8); //Use Application Shield's RGB LED for MQ2 sensor

class ultrasonic{ //Ultrasonic sensot for measuring depth of water in tank
    private:
        DigitalOut trigger;//DigitalOut trigger; //trigger pin for UltarSonic sensor
        InterruptIn echo; //Echo pin for water_level measurement. Uses as Interrupt to avoid code execution blocking.
        Ticker measure; //Ticker to trigger water_level measurement periodically.
        Timeout trigger_low; //Timeout interrupt to set trigger pin low after 10us of HIGH.
        Timer echoTimer; //Timer to get the time elapsed between echo sent and echo received back
        int echo_time_us; //Timer taken by echo to resturn abck to module.
        float interval; //Interval after which periodic measurement should start
        volatile bool measuring;
        FP<void,int> fp;
         
        void set_trigger_high(){ //Function to set Trigger Pin HIGH for 10us and set measuring = true
            measuring = true;
            trigger = ON;
            trigger_low.attach_us(callback(this, &ultrasonic::set_trigger_low), 10); //set trigger pin OFF after 10us   
        }
        
        void set_trigger_low(void){ //Function to set trigger pin LOW to fiish trigger after 10us. This will be called by low timer object.
            trigger = OFF;
        }
        
        void start_echo(void){//Function to start listening for echo back. This will start timer echoTimer. Will be based on Echo Pin RISE(or HIGH)
            if(!measuring) return; //avoid fluctuations on echo pin if any. Continue echo calculation only if measuring is TRUE.
            echoTimer.reset();
            echoTimer.start(); //Timer start since Echo pin is HIGH
        }
        void stop_echo(void){//Function to stop echoTimer after echo received back. Will be based on Echo Pin FALL(or LOW).
            if (!measuring)return; //avoid fluctuations on echo pin if any
            echoTimer.stop(); //stop timer since echo pin back to LOW
            echo_time_us = echoTimer.read_us(); //Calculate time for ultrasonic wave RTT.
            if(echo_time_us > 2000 || echo_time_us < 125)return; //ignore reading in case abnormally high or low RTT time
            measuring = false; //Marke measurement completed
            fp(echo_time_us / (29 * 2));
        }
public: 
        ultrasonic(PinName _trigger, PinName _echo,float _interval):trigger(_trigger), echo(_echo){ //Constructor: initialize trigger(default value OFF) & echo pin
            interval = _interval;
            echo.rise(callback(this, &ultrasonic::start_echo)); //Attach function to Interrupt on Echo pin Rise
            echo.fall(callback(this, &ultrasonic::stop_echo)); //Attach function to Interrupt on Echo pin Fall
            measuring = false;
        }

        void start_measure(){ //function to Set Trigger Pin high repeatedly every 'time' seconds
            measure.attach(callback(this, &ultrasonic::set_trigger_high),interval);//ticker to trigger measurement with callback function
        }
        void stop_measure(void){ //Dettach from Ticker and stop measurements
            measuring = false;
            measure.detach();
        }
        void req_callBack(FP<void, int> _fp){
            fp = _fp;
        }
};

class pump{
    private:
        DigitalOut signal; //Signal for relay to set PUMP ON and OFF
        ultrasonic sensor; //Ultrasonic module objet for measuring depth of water
        volatile bool pump_on_request; //Bool to inform whether ita PUMP on request or only water depth meas.
        const int LEVEL_LOW = 4; //Low Water level defined as 4CM. Signifies Tank FULL.
        const int LEVEL_HIGH = 9; //High Water level defined as 11CM. Signifies Tank EMPTY.
        FP<void, int> fp; //Function pointer to receive call back with depth of water level in tank.
        
    public:        
        //Initialize Ultrasonic module with TRIG and ECHO pin along with measurement interval.
        pump(PinName _signal, PinName _trigger, PinName _echo, float _interval):signal(_signal), sensor(_trigger, _echo, _interval){ 
            pump_on_request = false;
            fp.attach(this, &pump::update_depth); //update_depth function will be called bu ultrasonic module when depth meas is completed.
            sensor.req_callBack(fp); //set call back.
        }
        void start_measurement(int interval){
            sensor.start_measure(); //Start depth measurements to calculate the water level.
        }
        void stop_measurement(){
            sensor.stop_measure(); //Start depth measurements to calculate the water level.
        }
    
        void pump_on(){ //Pump On request by User.
            pump_on_request = true; //bool to confirm that Pump On request is raised
            //sensor.start_measure(); //Start depth measurements to calculate the water level.
        }
        void pump_off(){ //Switch OFF the pump
            pump_on_request = false; //Set User request to flase.
            //sensor.stop_measure(); //Stop the ultrasonic measurements
            signal = OFF; //Relay Signal for pump OFF.
            PUMP_STATE = OFF;
        }
        
        void get_depth(){ //Only depth of water requried. no need to switch ON pump.
            if(signal == OFF) //If Already PUMP ONdont start measurement again. 
                sensor.start_measure(); //Start Ultrasonic Measurements
        }
        
        void update_depth(int depth){ //Depth from ultrasonic is available
            WATER_LEVEL = depth;
            if(depth <= LEVEL_LOW) // set GREEN LED to inidicate the tank is FULL
                led_indicator.green_led();
            else if(depth <= LEVEL_HIGH)// set Amber LED to inidicate the tank is more than HALF
                led_indicator.amber_led();
            else 
                led_indicator.red_led();// set RED LED to inidicate the tank is less than HALF
            
            if(depth <= LEVEL_LOW && signal == ON)//Tank FULL and PUMP ON. Switch OFF PUMP.
                pump_off();
            else if(signal == OFF && pump_on_request && depth > LEVEL_HIGH ) //Pump is OFF and User have requested to on PUMP
                signal = ON;  //Pump ON 
            else if(signal == ON && !pump_on_request) //If PUMP is ON but not requster by User.
                pump_off(); //Pump OFF
            else if(signal == OFF && !pump_on_request) // Only for depth measurement. Pump on is not requested by user.
                pump_off();
            PUMP_STATE = signal;
        }
};

class control{ //Control object for fan and alarm. Only Switch ON and OFF requried.
    private:
        DigitalOut signal; //Signal pin for object
    public:
        control(PinName _signal, int state):signal(_signal,state){ //Initialize Signal PIN with default state(ON or OFF)
        }
        
        void set_state(int state){ //Set the state of control
            if (state >= 1){ //Switch ON
                signal = ON;
                led_indicator.pink_led(); //Blink RED LED
            }
            else //Switch OFF
                signal = OFF; //Set Control OFF
            //publish data
        }
        int get_state(){ //Get the state of the control
            return signal.read(); 
        }
};

class _temperature{ //Object for measuring current temperature and taking action accordingly
    private:
        LM75B temp_sensor; //LM75B temperature library used.
        float temperature_threshold_fan; // Threshold value for switching ON the FAN
        float temperature_threshold_alarm; // Threshold value for switching ON the ALARM
        Ticker periodic_meas; //Ticker for periodic measurement of the temperature
        control* alarm; //Poniter to control the alarm
        control* fan; //Poniter to control the fan
        void read_data(){ //Read the temperature from sensor
            TEMPERATURE = temp_sensor.temp(); //Use global variable TEMPERATURE which will be sent to ThingSpeak.
            if(TEMPERATURE >= temperature_threshold_alarm){ //Check if alarm threshold met.
                alarm->set_state(ON); //Set Alarm ON
                shield_led_indicator.red_led(); //Start RED LED Blinking
            }
            else if(TEMPERATURE >= temperature_threshold_fan){//Check if fan threshold met.
                shield_led_indicator.amber_led();//Set Amber LED Blinking
                fan->set_state(ON);//Set Alarm ON
                FAN_STATE = ON;
            }
            else
                shield_led_indicator.green_led();
        }
    public:
        _temperature(control* _alarm, control* _fan):temp_sensor(D14,D15){ //Initialize temp sensor with shield PINs D14, D15
            alarm = _alarm; //received object for alarm from Main function
            fan = _fan;//received object for fan from Main function
            temperature_threshold_fan = 30; //Fan threshold
            temperature_threshold_alarm = 40; //Alarm threshold
        }
        void start_measurement(int interval){ 
            periodic_meas.attach(callback(this, &_temperature::read_data), interval); //Start ticker based periodic temperature measurement
        }
        void stop_measurement(int interval){
            periodic_meas.detach();  //Stop ticker.
        }         
};

class _smoke_detector{ //Class to control the MQ2 Gas and Smoke detector
    private:
        MQ2 mq2; //Existing MQ2 library is used
        MQ2_data_t data; //data field to read the CO, SMOKE & GAS levels
        float CO_threshold_fan; //FAN Threshold for Carbon Monoxide
        float CO_threshold_alarm;//ALARM Threshold for Carbon Monoxide
        float GAS_threshold_fan; //FAN Threshold for combustible GASES
        float GAS_threshold_alarm; //ALARM Threshold for combustible GASES
        float SMOKE_threshold_fan; //FAN Threshold for SMOKE
        float SMOKE_threshold_alarm;//ALARM Threshold for SMOKE
        control* fan; //Poniter to control the fan
        control* alarm; //Poniter to control the alarm
        Ticker periodic_meas; //Ticker to complete periodic measurements
        
        void read_data(){ //Read the GAS/CO/SMOKE level from sensor
            mq2.read(&data);
            CO_level = mq2.readCO(); //CO Level
            GAS_level = mq2.readLPG(); //GAS Level
            SMOKE_level = mq2.readSmoke(); //SMOKE Level
            //Check if alarm threhold is crossed for any measurements
            if(data.co >= CO_threshold_alarm || data.lpg >= GAS_threshold_alarm || data.smoke >= SMOKE_threshold_alarm){ 
                shield_led_indicator.red_led(); //Set RED Blinker
                alarm->set_state(ON); //Set alarm ON
            }
            //Check if fan threhold is crossed for any measurements
            else if(data.co >= CO_threshold_fan || data.lpg >= GAS_threshold_fan || data.smoke >= SMOKE_threshold_fan){ 
                shield_led_indicator.amber_led(); //Set AMBER LED
                fan->set_state(ON); //Set FAN ON
            }
        }
        
    public:
        //Initialize MQ2 sensor using Analog pin.
        _smoke_detector(PinName _analog, control* _alarm, control* _fan):mq2(_analog){ 
            mq2.begin(); //mq2 begin method to Calibrate sensor
            fan = _fan; //Fan control object from Main function
            alarm = _alarm;//Alarm control object from Main function
            CO_threshold_alarm = 3;
            CO_threshold_fan = 2;
            GAS_threshold_fan = 2;
            GAS_threshold_alarm = 3;
            SMOKE_threshold_alarm = 3;
            SMOKE_threshold_fan = 2;
        }
               
        void start_measurement(int interval){ //Start Ticker based periodic measurements
            periodic_meas.attach(callback(this, &_smoke_detector::read_data), interval);
        }
        void stop_measurement(){//Stop Ticker based periodic measurements
            periodic_meas.detach();
        }  
};

int arrivedcount = 0;

pump water_pump(PUMP,TRIGGER, ECHO, 1); //Water Pump object
control alarm(ALARM_SIG, ON); //Alarm Pump object
control fan(FAN_SIG, OFF); //Fan object
_smoke_detector smoke_detector(MQ2_SIG, &alarm, &fan); //Smoke Detector object
_temperature temperature(&alarm, &fan);//Uses fixed PIN on shiled- D14, D15. So no need to pass from main
NetworkInterface* network = easy_connect(true); //Network object. Creates ethernt conncetion
MQTTNetwork mqttNetwork(network); //MQTTNetwork object. Takes Network object as input for connection over ethernet
MQTT::Client<MQTTNetwork, Countdown> client(mqttNetwork); //MQTT Client for Connction with broker, Subscribe and Publsih service.

 
void field1_pump_state(MQTT::MessageData& md) //Callback functionf for Field1 from thingspeak.
{
    MQTT::Message &message = md.message; //Message recived using MQTT
    logMessage("Message arrived: qos %d, retained %d, dup %d, packetid %d\r\n", message.qos, message.retained, message.dup, message.id);
    logMessage("Message Field1 %s\r\n", message.payload); //Message Payload
    if(strtol((char *)message.payload, '\0', 10) == 1){  //convert to int and check if power ON command
        logMessage("Water Pump ON\r\n"); 
        water_pump.stop_measurement();
        water_pump.start_measurement(2); //Start Water Pump measurements. At intervals of 2s. This is small value because for test purpose very small tank is being used. In real life this could be in mins.
        water_pump.pump_on(); //Start the pump for 1st time
    }
    else{
        logMessage("Water Pump OFF\r\n"); 
        water_pump.pump_off(); //Power off pump otherwise
    }
}

void field2_fan_state(MQTT::MessageData& md)//Callback functionf for Field2 from thingspeak.
{
    MQTT::Message &message = md.message; //Message recived using MQTT
    logMessage("Message arrived: qos %d, retained %d, dup %d, packetid %d\r\n", message.qos, message.retained, message.dup, message.id);
    logMessage("Message Field2 %s\r\n", message.payload);//Message Payload
    if(strtol((char *)message.payload, '\0', 10) == 1){//convert to int and check if power ON command
        logMessage("FAN ON\r\n");
        fan.set_state(OFF);//Power On FAN
    }
    else{
        logMessage("FAN OFF\r\n");
        fan.set_state(ON);//Power off fan otherwise
    }
}

void field3_alarm_state(MQTT::MessageData& md)
{
    MQTT::Message &message = md.message;
    logMessage("Message arrived: qos %d, retained %d, dup %d, packetid %d\r\n", message.qos, message.retained, message.dup, message.id);
    logMessage("Message Field3 %s\r\n", message.payload);
//    RED = strtol((char *)message.payload, '\0', 10);
    if(strtol((char *)message.payload, '\0', 10) == 1){//convert to int and check if power ON command
        logMessage("Alarm ON\r\n");;//Power On Alarm. This scenario should never occur,
        alarm.set_state(ON);
    }
    else{
        logMessage("Alarm OFF\r\n");
        alarm.set_state(OFF); //Power off Alarm otherwise
    }
    ++arrivedcount;
}

void publish_field1(){ //Function to publish the values to ThingSpeak fields
    static int last_update = -1;
    if(PUMP_STATE == last_update)
        return; //Ignore unncessary update in case no chnage in value.
    last_update = PUMP_STATE;
    logMessage("Publish Routine\r\n");
    int rc = 0;
    MQTT::Message message; 
    char* topic = "channels/1608275/publish";//Topic contaning Channel ID
    char buf[100]; //Buffer for storing data
    sprintf(buf, "field1=%d&status=MQTTPUBLISH",PUMP_STATE);
    message.qos = MQTT::QOS0; //MQTT QOS 0 used. Best effort QOS
    message.retained = false;
    message.dup = false;
    message.payload = (void*)buf; //Copy buffer to Message Payload
    message.payloadlen = strlen(buf)+1;
    rc = client.publish(topic, message); //Publsih data using MQTT client
    logMessage("Publish rc: %d, Buf: %s\r\n", rc, buf);
}

void publish_field2(){ //Function to publish the values to ThingSpeak fields
    static int last_update = -1;
    if(FAN_STATE == last_update)
        return; //Ignore unncessary update in case no chnage in value.
    last_update = FAN_STATE;
    logMessage("Publish Routine\r\n");
    int rc = 0;
    MQTT::Message message; 
    char* topic = "channels/1608275/publish";//Topic contaning Channel ID
    char buf[100]; //Buffer for storing data
    sprintf(buf, "field2=%d&status=MQTTPUBLISH",FAN_STATE);
    message.qos = MQTT::QOS0; //MQTT QOS 0 used. Best effort QOS
    message.retained = false;
    message.dup = false;
    message.payload = (void*)buf; //Copy buffer to Message Payload
    message.payloadlen = strlen(buf)+1;
    rc = client.publish(topic, message); //Publsih data using MQTT client
    logMessage("Publish rc: %d, Buf: %s\r\n", rc, buf);
}

void publish_field3(){ //Function to publish the values to ThingSpeak fields
    static int last_update = -1;
    if(ALARM_STATE == last_update)
        return; //Ignore unncessary update in case no chnage in value.
    last_update = ALARM_STATE;
    logMessage("Publish Routine\r\n");
    int rc = 0;
    MQTT::Message message; 
    char* topic = "channels/1608275/publish";//Topic contaning Channel ID
    char buf[100]; //Buffer for storing data
    sprintf(buf, "field3=%d&status=MQTTPUBLISH",ALARM_STATE);
    message.qos = MQTT::QOS0; //MQTT QOS 0 used. Best effort QOS
    message.retained = false;
    message.dup = false;
    message.payload = (void*)buf; //Copy buffer to Message Payload
    message.payloadlen = strlen(buf)+1;
    rc = client.publish(topic, message); //Publsih data using MQTT client
    logMessage("Publish rc: %d, Buf: %s\r\n", rc, buf);
}

void publish_field4(){ //Function to publish the values to ThingSpeak fields
    static int last_update = -1;
    last_update = WATER_LEVEL;
    //logMessage("Publish Routine\r\n");
    int rc = 0;
    MQTT::Message message; 
    char* topic = "channels/1608275/publish";//Topic contaning Channel ID
    char buf[100]; //Buffer for storing data
    //field5=%0.1f&field6=%0.1f&field7=%0.1f&field8=%0.1f&
    sprintf(buf, "field4=%d&field1=%d&field2=%d&field3=%d&status=MQTTPUBLISH",WATER_LEVEL, PUMP_STATE,FAN_STATE,ALARM_STATE);
    message.qos = MQTT::QOS0; //MQTT QOS 0 used. Best effort QOS
    message.retained = false;
    message.dup = false;
    message.payload = (void*)buf; //Copy buffer to Message Payload
    message.payloadlen = strlen(buf)+1;
    rc = client.publish(topic, message); //Publsih data using MQTT client
    //logMessage("Publish rc: %d, Buf: %s\r\n", rc, buf);
}

void publish_field5(){ //Function to publish the values to ThingSpeak fields
    static int last_update = -1;
    last_update = GAS_level;
    //logMessage("Publish Routine\r\n");
    int rc = 0;
    MQTT::Message message; 
    char* topic = "channels/1608275/publish";//Topic contaning Channel ID
    char buf[100]; //Buffer for storing data
    sprintf(buf, "field5=%d&field6=%d&field7=%d&field8=%d&status=MQTTPUBLISH",GAS_level, CO_level, SMOKE_level, TEMPERATURE);
    message.qos = MQTT::QOS0; //MQTT QOS 0 used. Best effort QOS
    message.retained = false;
    message.dup = false;
    message.payload = (void*)buf; //Copy buffer to Message Payload
    message.payloadlen = strlen(buf)+1;
    rc = client.publish(topic, message); //Publsih data using MQTT client
    //logMessage("Publish rc: %d, Buf: %s\r\n", rc, buf);
}

void publish_field6(){ //Function to publish the values to ThingSpeak fields
    static int last_update = -1;
    if(CO_level == last_update)
        return; //Ignore unncessary update in case no chnage in value.
    last_update = CO_level;
    logMessage("Publish Routine\r\n");
    int rc = 0;
    MQTT::Message message; 
    char* topic = "channels/1608275/publish";//Topic contaning Channel ID
    char buf[100]; //Buffer for storing data
    sprintf(buf, "field6=%d&status=MQTTPUBLISH",CO_level);
    message.qos = MQTT::QOS0; //MQTT QOS 0 used. Best effort QOS
    message.retained = false;
    message.dup = false;
    message.payload = (void*)buf; //Copy buffer to Message Payload
    message.payloadlen = strlen(buf)+1;
    rc = client.publish(topic, message); //Publsih data using MQTT client
    logMessage("Publish rc: %d, Buf: %s\r\n", rc, buf);
}

void publish_field8(){ //Function to publish the values to ThingSpeak fields
    static int last_update = -1;
    if(TEMPERATURE == last_update)
        return; //Ignore unncessary update in case no chnage in value.
    last_update = TEMPERATURE;
    logMessage("Publish Routine\r\n");
    int rc = 0;
    MQTT::Message message; 
    char* topic = "channels/1608275/publish";//Topic contaning Channel ID
    char buf[100]; //Buffer for storing data
    sprintf(buf, "field8=%d&status=MQTTPUBLISH",TEMPERATURE);
    message.qos = MQTT::QOS0; //MQTT QOS 0 used. Best effort QOS
    message.retained = false;
    message.dup = false;
    message.payload = (void*)buf; //Copy buffer to Message Payload
    message.payloadlen = strlen(buf)+1;
    rc = client.publish(topic, message); //Publsih data using MQTT client
    logMessage("Publish rc: %d, Buf: %s\r\n", rc, buf);
}

void publish_field7(){ //Function to publish the values to ThingSpeak fields
    static int last_update = -1;
    if(SMOKE_level == last_update)
        return; //Ignore unncessary update in case no chnage in value.
    last_update = SMOKE_level;
    logMessage("Publish Routine\r\n");
    int rc = 0;
    MQTT::Message message; 
    char* topic = "channels/1608275/publish";//Topic contaning Channel ID
    char buf[100]; //Buffer for storing data
    sprintf(buf, "field7=%d&status=MQTTPUBLISH",SMOKE_level);
    message.qos = MQTT::QOS0; //MQTT QOS 0 used. Best effort QOS
    message.retained = false;
    message.dup = false;
    message.payload = (void*)buf; //Copy buffer to Message Payload
    message.payloadlen = strlen(buf)+1;
    rc = client.publish(topic, message); //Publsih data using MQTT client
    logMessage("Publish rc: %d, Buf: %s\r\n", rc, buf);
}

int main(int argc, char* argv[])
{
    smoke_detector.start_measurement(2); //Start periodic Smoke detector measurements at intervals of 10s
   temperature.start_measurement(20); //Start periodic Temperaturre readings. At interval of 20s
   water_pump.start_measurement(2); //Start Water Pump measurements. At intervals of 2s. This is small value because for test purpose very small tank is being used. In real life this could be in mins.
   water_pump.pump_on(); //Start the pump for 1st time
    float version = 0.6;
    logMessage("HelloMQTT: version is %.2f\r\n", version);
    int ret = 0;
    
    if (!network) { //Check if ethernet connection successful
        return -1;
    }

    const char* hostname = "mqtt3.thingspeak.com"; //Thingspeak host name
    int port = 1883; //MQTT default port number
    logMessage("Connecting to %s:%d\r\n", hostname, port);
    int rc = mqttNetwork.connect(hostname, port); //Attempt to connect to MQTT host
    if (rc != 0)
        logMessage("rc- TCP connect: %d\r\n", rc);

    MQTTPacket_connectData data = MQTTPacket_connectData_initializer; //Build MQTT Client connection data
    data.MQTTVersion = 3; //MQTT Version
    data.clientID.cstring = "FA8KKzQ1CDY6EjENATYTESc"; //Thingspeak MQTT Client ID
    data.username.cstring = "FA8KKzQ1CDY6EjENATYTESc";//Thingspeak MQTT Username
    data.password.cstring = "vVdiiGscVIGLEgQIZ250Unxc";//Thingspeak MQTT Password
    
    
    if ((rc = client.connect(data)) != 0) //Connect Client
        logMessage("rc- connect: %d\r\n", rc);
        
    char* topic = "channels/1608275/subscribe/fields/field1"; //Subscribe for field1
    if ((rc = client.subscribe(topic, MQTT::QOS0, field1_pump_state)) != 0)//Use field1_pump_state function pointer as call back function
        logMessage("rc1: %d\r\n", rc);
    
    topic = "channels/1608275/subscribe/fields/field2";
    if ((rc = client.subscribe(topic, MQTT::QOS0, field2_fan_state)) != 0)//Use field2_fan_state function pointer as call back function
        logMessage("rc2: %d\r\n", rc);

    topic = "channels/1608275/subscribe/fields/field3";
    if ((rc = client.subscribe(topic, MQTT::QOS0, field3_alarm_state)) != 0)//Use field3_alarm_state function pointer as call back function
        logMessage("rc: %d\r\n", rc);
    
    while (1){
        logMessage("GAS: %d, CO: %d, Smoke: %d, TEMP = %0.1f, Water = %d", GAS_level, CO_level, SMOKE_level, TEMPERATURE, WATER_LEVEL);
        //publish_field1();
        //wait(1);
        //publish_field2();
        //wait(1);
        //publish_field3();
        //wait(1);
        //publish_field4();
        //wait(5);
        //publish_field5();
        //wait(5);
        //publish_field5();
        //wait(1);
        //publish_field6();
        //wait(1);
        //publish_field7();
        //wait(1);
        //publish_field8();
        client.yield(50);
        //client.connect(data);
    }
    if ((rc = client.unsubscribe(topic)) != 0)
        logMessage("rc from unsubscribe was %d\r\n", rc);
    if ((rc = client.disconnect()) != 0)
        logMessage("rc from disconnect was %d\r\n", rc);
    mqttNetwork.disconnect();
    logMessage("Version %.2f: finish %d msgs\r\n", version, arrivedcount);
    return 0;
}