Grove* Sensors, AWS Greengrass* Group and Device-to-Cloud Communication

Introduction

This article explores a method for monitoring the environment using sensor data rules and issuing alerts for abnormal readings. To that end, we will setup a continuous MQTT communication for passing sensor data using the UP Squared* board and Grove* shield, the AWS Greengrass* group and device-to-cloud communication. The Grove sensors used in this article are the loudness sensor, the barometer sensor, and the IR distance interrupter. The environment’s loudness, ambient temperature, barometric pressure, and altitude will be the data collected from the sensors. The rules are set according to the sensor readings to determine the abnormal values and to issue alerts. First, we will configure the Greengrass devices via the AWS* console, and then we will configure them on both the UP Squared* board and another Linux Ubuntu* 16.04 machine. On the UP Squared board, we will run a Python* script to collect the sensor data, and filter it out based on rules. Then, the publisher device will send the sensor data to the subscriber device (the Linux Ubuntu machine), as well as send the alerts for abnormal readings to the AWS IoT cloud via MQTT topics.

Learn more about the AWS Greengrass

Learn more about the UP Squared board

Prerequisites

Hardware:

AWS Greengrass:

Grove* Sensors

On the UP Squared board, install the MRAA and UPM libraries to interface with Grove sensors:

sudo add-apt-repository ppa:mraa/mraa
sudo apt-get update
sudo apt-get install libmraa1 libmraa-dev mraa-tools python-mraa python3-mraa
sudo apt-get install libupm-dev libupm-java python-upm python3-upm node-upm upm-example

Code 1. Commands to install Grove dependencies

To enable non-privileged access to the Grove sensors, run the following commands:

sudo add-apt-repository ppa:ubilinux/up
sudo apt update
sudo apt install upboard-extras
sudo usermod -a -G i2c ggc_user
sudo usermod -a -G grio ggc_user
sudo reboot

Code 2. Commands to enable non-privileged access to Grove sensors

AWS Greengrass* Setup

To install AWS Greengrass on the UP Squared board, follow these instructions

Check that you have installed all the needed dependencies:

sudo apt update
git clone https://github.com/aws-samples/aws-greengrass-samples.git
cd aws-greengrass-samples
cd greengrass-dependency-checker-GGCv1.3.0
sudo ./check_ggc_dependencies

Code 3. Commands to check AWS dependencies

Go to the AWS IoT console. Choose Greengrass from leftside menu, select Groups underneath it, and select your group from main window:

Commands to install Grove Dependencies
Figure 1. AWS Greengrass Groups view

Select Devices from the left-side menu. Click Add Device button, on the top right corner:

Commands to enable non-privileged access to Grove Sensors
Figure 2. Greengrass devices view

Choose Create New Device:

Creating a new device view
Figure 3. Creating a new device view

Enter the name, pub, in the field and clickNext:

Creating a registry entry for a device
Figure 4. Creating a Registry entry for a device

Click on Use Defaults button:

Set up Security View
Figure 5. Set up security view

Download the security credentials, we will use them in the next module. Click Finish:

Download Security Credentials
Figure 6. Download security credentials

You should see the new device on the screen:

Greengrass Devices View
Figure 7. Greengrass Devices view

Add another new device and name it sub. When you’re done, you should see the following screen, with two new devices:

 Updated Greengrass Devices View
Figure 8. Updated Greengrass Devices view

On the left-side menu, select Subscriptions. Click on Add Subscription:

 Greengrass Subscriptions View
Figure 9. Greengrass Subscriptions view

For the source, go to Devices tab and select pub. For the target, go to Devices tab and select sub. Click Next:

Selecting Source and Target View
Figure 10. Creating subscription: selecting source and target view

Add topic, sensors/data/pubsub:

Adding Topic View
Figure 11. Creating subscription: adding topic view

Review the subscription and click Next:

Confirm and Save Subscription View
Figure 12. Confirm and save Subscription view

You can see the subscription:

Subscriptions View
Figure 13. Subscriptions view

Create another subscription by following the steps below. Choose pub as a source and IoT Cloud as a target. For topic, enter sensors/data/alerts. After you done, you should see a similar screen:

Updated Subscriptions View
Figure 14. Updated Subscriptions view

On the group header, click Actions, select Deploy and wait until it is successfully completed:

Deploying the Greengrass Group View
Figure 15. Deploying the Greengrass Group view

Publisher Setup

In this module, we will configure the Greengrass device to be a MQTT publisher. In this case, we are using the UP Squared board both as a Greengrass core, and as a publisher device, so we can get the sensor data from a device. The other Linux machine will be used as a subscriber device and configured in the next module.

Copy pub’s tar file, which was saved in a previous module, and untar it. Save the files in the publisher device (UP Squared board) and rename them for readability:

tar –xzvf <pub-credentials-id>-setup.tar.gz
mv <pub-credentials-id>.cert.pem pub.cert.pem
mv <pub-credentials-id>.private.pem pub.private.pem
mv <pub-credentials-id>.public.pem pub.public.pem

Code 4. Commands to Save Publisher’s Credentials

In the publisher folder, get a root certificate and save it as root-ca-cert.pem:

wget https://www.symantec.com/content/en/us/enterprise/verisign/roots/VeriSign-Class%203-Public-Primary-Certification-Authority-G5.pem –O root-ca-cert.pem

Code 5. Commands to get a root certificate

On Up Squared board, install AWS IoT SDK for Python:

python
>>> import ssl
>>> ssl.OPENSSL_VERSION
# output should be version of OpenSSL 1.0.1+:‘OpenSSL 1.0.2g 1 Mar 2016’
>>> exit()
cd ~ 
git clone https://github.com/aws/aws-iot-device-sdk-python.git
cd aws-iot-device-sdk-python
python setup.py install

Code 6. Commands to install AWS IoT SDK for Python

The following rules will allow us to filter the sensor data values, and capture messages if the values are abnormal. We will determine the range of normal readings. For temperature, we define the normal range to be below 25 and above 20 degrees Celsius:

class TemperatureOver25(Rule):
    def predicate(self, sensorValue):
        return sensorValue > 25

    def action(self, sensorValue):
        message = “Temperature Over 25 Rule activated on “ + self.sensorId + “ with sensor value “ + str(sensorValue)
        return message

class TemperatureUnder20(Rule):
    def predicate(self, sensorValue):
        return sensorValue < 20

    def action(self, sensorValue):
        message = “Temperature Under 20 Rule activated on “ + self.sensorId + “ with sensor value “ + str(sensorValue)
        return message

Code 7. Code snippet with temperature classes

We will filter the rules by sensor:

def filterBySensorId(sensorId, rules):
    “Filter a list of rules by sensorId”
    return [rule for rule in rules if rule.sensorId == sensorId]

Code 8. Code snippet for filtering rules by sensor

The rules will be imported and instantiated in the next script, greengrassCommunication.py. Save the following Python script as sensor_rules.py to the publisher folder:

class Rule:
    “””
    A Base Class for defining IoT automation rules.
    “””
    
    def __init__(self, sensorId):
        “””
        Constructor function that takes an id 
        that uniquely identifies a sensor.
        “””
        self.sensorId = sensorId
    
    def predicate(self, sensorValue):
        “In the base Rule class, the predicate always returns False”
        return False
                    
    def action(self, sensorValue):
        message = “Generic Rule activiation on “ + self.sensorId + “ with senor value “ + str(sensorValue)
        return message

class TemperatureOver25(Rule):
    def predicate(self, sensorValue):
        return sensorValue > 25

    def action(self, sensorValue):
        message = “Temperature Over 25 Rule activated on “ + self.sensorId + “ with sensor value “ + str(sensorValue)
        return message

class TemperatureUnder20(Rule):
    def predicate(self, sensorValue):
        return sensorValue < 20

    def action(self, sensorValue):
        message = “Temperature Under 20 Rule activated on “ + self.sensorId + “ with sensor value “ + str(sensorValue)
        return message

class PressureOver96540(Rule):
    def predicate(self, sensorValue):
        return sensorValue > 96540
    def action(self, sensorValue):
        message = “Pressure Over 96540 Rule activated on “ + self.sensorId + “ with sensor value “ + str(sensorValue)
        return message

class PressureUnder96534(Rule):
    def predicate(self, sensorValue):
        return sensorValue < 96534
    def action(self, sensorValue):
        message = “Pressure Under 96534 Rule activated on “ + self.sensorId + “ with sensor value “ + str(sensorValue)
        return message

class AltitudeOver1214(Rule):
    def predicate(self, sensorValue):
        return sensorValue > 1214
    def action(self, sensorValue):
        message = “Altitude Over 1214 Rule activated on “ + self.sensorId + “ with sensor value “ + str(sensorValue)
        return message

class AltitudeUnder1214(Rule):
    def predicate(self, sensorValue):
        return sensorValue < 1214
    def action(self, sensorValue):
        message = “Altitude Under 1214 Rule activated on “ + self.sensorId + “ with sensor value “ + str(sensorValue)
        return message

class ObjectDetected(Rule):
    def predicate(self, sensorValue):
        return sensorValue == True
    def action(self, sensorValue):
        message = “Object Detected Rule activated on “ + self.sensorId + “ with sensor value “ + str(sensorValue)
        return message

class LoudnessOver3(Rule):
    def predicate(self, sensorValue):
        return sensorValue > 3
    def action(self, sensorValue):
        message = “Loudness Over 3 Rule activated on “ + self.sensorId + “ with sensor value “ + str(sensorValue)
        return message

class LoudnessUnder05(Rule):
    def predicate(self, sensorValue):
        return sensorValue < 0.5
    def action(self, sensorValue):
        message = “Loudness Under 0.5 Rule activated on “ + self.sensorId + “ with sensor value “ + str(sensorValue)
        return message

def filterBySensorId(sensorId, rules):
    “Filter a list of rules by sensorId”
    return [rule for rule in rules if rule.sensorId == sensorId]

Code 9. sensor_rules.py, Python script to create rules

We will need the following imports to get Grove sensor data:

from upm import pyupm_bmp280 as bmp280
from upm import pyupm_rfr359f as rfr359f
from upm import pyupm_loudness as loudness
import mraa

Code 10. Import statements for Grove sensors

We will interface with Grove shield and instantiate the sensors. Loudness sensor is connected to pin A2 which needs to be offset by 512. The IR distance interrupter, which will be used for object detection, is connected to pin D2 and needs to be offset by 512:

mraa.addSubplatform(mraa.GROVEPI, “0”)
# bmp is a barometer sensor
bmp = bmp280.BMP280(0, 0x76)
loudness_sensor = loudness.Loudness(514, 5.0)
# IR distance interruptor
object_detector = rfr359f.RFR359F(514)

Code 11. Interfacing with Grove shield and instantiating its sensors

Rules will be instantiated for each one class and will be saved in rules list:

r1 = sensor_rules.Rule(“generic_Sensor”)  # Generic Rule assigned to a sensor name “generic_Sensor”
r2 = sensor_rules.TemperatureOver25(“temperature”) # Derived rule assigned to a sensor name “temperature”
r3 = sensor_rules.TemperatureUnder20(“temperature”) # Derived rule assigned to a sensor name “temperature”
r4 = sensor_rules.PressureOver96540(“pressure”) # PressureOver96534 assigned to a sensor name “pressure”
r5 = sensor_rules.PressureUnder96534(“pressure”) # PressureUnder96534 assigned to a sensor name “pressure”
r6 = sensor_rules.AltitudeOver1214(“altitude”) # AltitudeOver1214 assigned to a sensor name “altitude”
r7 = sensor_rules.AltitudeUnder1214(“altitude”) # AltitudeUnder1214 assigned to a sensor name “altitude”
r8 = sensor_rules.ObjectDetected(“object detection”) # ObjectDetected is assigned to a sensor “object detection”
r9 = sensor_rules.LoudnessOver3(“loudness”) # LoudnessOver3 is assigned to a sensor “loudness”
r10 = sensor_rules.LoudnessUnder05(“loudness”) # LoudnessUnder05 is assigned to a sensor “loudness”

rules = [r1, r2, r3, r4, r5, r6, r7, r8, r9, r10]

Code 12. Instantiating rules

The following code snippet reads the sensor data and saves them in a JSON format:

def get_sensor_data():
    # Getting new readings from barometer sensor
    bmp.update()
    pressure_value = bmp.getPressurePa()
    temperature_value = bmp.getTemperature()
    # Translating altitude value from meters to feet
    altitude_value = int(bmp.getAltitude() * 3.2808)
    # Get IR object detection data

    # returns True or False
    object_detection_value = object_detector.objectDetected()

    loudness_value = loudness_sensor.loudness()

    timestamp = time.time()

    sensor_data = {“values”:[
                     {“sensor_id”: “pressure”, “value”: pressure_value, “timestamp”: timestamp}, 
                     {“sensor_id”: “temperature”, “value”: temperature_value, “timestamp”: timestamp},
                     {“sensor_id”: “altitude”, “value”: altitude_value, “timestamp”: timestamp},
                     {“sensor_id”: “object detection”, “value”: object_detection_value, “timestamp”: timestamp},
                     {“sensor_id”: “loudness”, “value”: loudness_value, “timestamp”: timestamp}
                    ]
                   }
    sensor_data_json = json.loads(json.dumps(sensor_data[“values”]))

    return sensor_data_json

Code 13. Code snippet to get sensor data

This code snippet filters the JSON object with sensor data and creates alert messages for abnormal readings:

def apply_rules(sensor_data_json):
    rules_message = []
    
    for item in sensor_data_json:
        sensor_id = item[‘sensor_id’]
        sensor_value = item[‘value’]
        filteredRules = sensor_rules.filterBySensorId(sensor_id, rules)
        for r in filteredRules:            
            if r.predicate(sensor_value) == True:                
                rules_message.append(r.action(sensor_value))
    return rules_message

Code 14. Code snippet to filter rules by sensor

The while loop will run continuously to publish the sensor data in the sensors/data/pubsub topic and alerts in sensors/data/alerts topic:

while True:
    if args.mode == ‘both’ or args.mode == ‘publish’:
        message = {}
        sensor_data_json = get_sensor_data()
        # message[‘message’] = args.message
        message[‘message’] = get_message(sensor_data_json)
        message[‘alerts’] = apply_rules(sensor_data_json)
        message[‘sequence’] = loopCount
        messageJson = json.dumps(message)
        myAWSIoTMQTTClient.publish(topic, messageJson, 0)
        if args.mode == ‘publish’:
            print(‘Published topic %s: %s\n’ % (topic, messageJson))
        cloud_topic = “sensors/data/alerts”
        alerts_json = json.dumps(message[‘alerts’])
        myAWSIoTMQTTClient.publish(cloud_topic, alerts_json, 0)
        loopCount += 1
    time.sleep(1)

Code 15. Continuous while loop for publishing sensor data messages and alerts

The Code 16 Python script will create the sensor rules, collect the sensor data, and send MQTT messages with the sensor data values to the Greengrass subscriber device as well as send alerts with abnormal data to the IoT cloud. Save this Python script as greengrassCommunication.py to the same folder:

from __future__ import print_function
import os
import sys
import time
import uuid
import json
import argparse
from AWSIoTPythonSDK.core.greengrass.discovery.providers import DiscoveryInfoProvider
from AWSIoTPythonSDK.core.protocol.connection.cores import ProgressiveBackOffCore
from AWSIoTPythonSDK.MQTTLib import AWSIoTMQTTClient
from AWSIoTPythonSDK.exception.AWSIoTExceptions import DiscoveryInvalidRequestException
import signal, atexit
from upm import pyupm_bmp280 as bmp280
from upm import pyupm_rfr359f as rfr359f
from upm import pyupm_loudness as loudness
import mraa
import sensor_rules


mraa.addSubplatform(mraa.GROVEPI, “0”)
# bmp is a barometer sensor
bmp = bmp280.BMP280(0, 0x76)
loudness_sensor = loudness.Loudness(514, 5.0)
# IR distance interruptor
object_detector = rfr359f.RFR359F(514)

r1 = sensor_rules.Rule(“generic_Sensor”)  # Generic Rule assigned to a sensor name “generic_Sensor”
r2 = sensor_rules.TemperatureOver25(“temperature”) # Derived rule assigned to a sensor name “temperature”
r3 = sensor_rules.TemperatureUnder20(“temperature”) # Derived rule assigned to a sensor name “temperature”
r4 = sensor_rules.PressureOver96540(“pressure”) # PressureOver96534 assigned to a sensor name “pressure”
r5 = sensor_rules.PressureUnder96534(“pressure”) # PressureUnder96534 assigned to a sensor name “pressure”
r6 = sensor_rules.AltitudeOver1214(“altitude”) # AltitudeOver1214 assigned to a sensor name “altitude”
r7 = sensor_rules.AltitudeUnder1214(“altitude”) # AltitudeUnder1214 assigned to a sensor name “altitude”
r8 = sensor_rules.ObjectDetected(“object detection”) # ObjectDetected is assigned to a sensor “object detection”
r9 = sensor_rules.LoudnessOver3(“loudness”) # LoudnessOver3 is assigned to a sensor “loudness”
r10 = sensor_rules.LoudnessUnder05(“loudness”) # LoudnessUnder05 is assigned to a sensor “loudness”

rules = [r1, r2, r3, r4, r5, r6, r7, r8, r9, r10]

def get_sensor_data():
    # Getting new readings from barometer sensor
    bmp.update()
    pressure_value = bmp.getPressurePa()
    temperature_value = bmp.getTemperature()
    # Translating altitude value from meters to feet
    altitude_value = int(bmp.getAltitude() * 3.2808)
    # Get IR object detection data

    # returns True or False
    object_detection_value = object_detector.objectDetected()

    loudness_value = loudness_sensor.loudness()

    timestamp = time.time()

    sensor_data = {“values”:[
                     {“sensor_id”: “pressure”, “value”: pressure_value, “timestamp”: timestamp}, 
                     {“sensor_id”: “temperature”, “value”: temperature_value, “timestamp”: timestamp},
                     {“sensor_id”: “altitude”, “value”: altitude_value, “timestamp”: timestamp},
                     {“sensor_id”: “object detection”, “value”: object_detection_value, “timestamp”: timestamp},
                     {“sensor_id”: “loudness”, “value”: loudness_value, “timestamp”: timestamp}
                    ]
                   }
    sensor_data_json = json.loads(json.dumps(sensor_data[“values”]))

    return sensor_data_json


def get_message(sensor_data_json):    
    sensor_data_message = []
    for item in sensor_data_json:
        sensor_id = item[‘sensor_id’]
        sensor_value = item[‘value’]        
        sensor_data_message.append(item)
    return sensor_data_message

def apply_rules(sensor_data_json):
    rules_message = []
    
    for item in sensor_data_json:
        sensor_id = item[‘sensor_id’]
        sensor_value = item[‘value’]
        filteredRules = sensor_rules.filterBySensorId(sensor_id, rules)
        for r in filteredRules:            
            if r.predicate(sensor_value) == True:                
                rules_message.append(r.action(sensor_value))
    return rules_message

AllowedActions = [‘both’, ‘publish’, ‘subscribe’]

# General message notification callback
def customOnMessage(message):
    print(‘Received message on topic %s: %s\n’ % (message.topic, message.payload))

MAX_DISCOVERY_RETRIES = 10
GROUP_CA_PATH = “./groupCA/”

# Read in command-line parameters
parser = argparse.ArgumentParser()
parser.add_argument(“-e”, “–endpoint”, action=”store”, required=True, dest=”host”, help=”Your AWS IoT custom endpoint”)
parser.add_argument(“-r”, “–rootCA”, action=”store”, required=True, dest=”rootCAPath”, help=”Root CA file path”)
parser.add_argument(“-c”, “–cert”, action=”store”, dest=”certificatePath”, help=”Certificate file path”)
parser.add_argument(“-k”, “–key”, action=”store”, dest=”privateKeyPath”, help=”Private key file path”)
parser.add_argument(“-n”, “–thingName”, action=”store”, dest=”thingName”, default=”Bot”, help=”Targeted thing name”)
parser.add_argument(“-m”, “–mode”, action=”store”, dest=”mode”, default=”both”,
                    help=”Operation modes: %s”%str(AllowedActions))

args = parser.parse_args()
host = args.host
rootCAPath = args.rootCAPath
certificatePath = args.certificatePath
privateKeyPath = args.privateKeyPath
clientId = args.thingName
thingName = args.thingName
topic = “sensors/data/pubsub”

if args.mode not in AllowedActions:
    parser.error(“Unknown –mode option %s. Must be one of %s” % (args.mode, str(AllowedActions)))
    exit(2)

if not args.certificatePath or not args.privateKeyPath:
    parser.error(“Missing credentials for authentication.”)
    exit(2)

# Progressive back off core
backOffCore = ProgressiveBackOffCore()

# Discover GGCs
discoveryInfoProvider = DiscoveryInfoProvider()
discoveryInfoProvider.configureEndpoint(host)
discoveryInfoProvider.configureCredentials(rootCAPath, certificatePath, privateKeyPath)
discoveryInfoProvider.configureTimeout(10)  # 10 sec

retryCount = MAX_DISCOVERY_RETRIES
discovered = False
groupCA = None
coreInfo = None
while retryCount != 0:
    try:
        discoveryInfo = discoveryInfoProvider.discover(thingName)
        caList = discoveryInfo.getAllCas()
        coreList = discoveryInfo.getAllCores()

        # We only pick the first ca and core info
        groupId, ca = caList[0]
        coreInfo = coreList[0]
        print(“Discovered GGC: %s from Group: %s” % (coreInfo.coreThingArn, groupId))

        print(“Now we persist the connectivity/identity information…”)
        groupCA = GROUP_CA_PATH + groupId + “_CA_” + str(uuid.uuid4()) + “.crt”
        if not os.path.exists(GROUP_CA_PATH):
            os.makedirs(GROUP_CA_PATH)
        groupCAFile = open(groupCA, “w”)
        groupCAFile.write(ca)
        groupCAFile.close()

        discovered = True
        print(“Now proceed to the connecting flow…”)
        break
    except DiscoveryInvalidRequestException as e:
        print(“Invalid discovery request detected!”)
        print(“Type: %s” % str(type(e)))
        print(“Error message: %s” % e.message)
        print(“Stopping…”)
        break
    except BaseException as e:
        print(“Error in discovery!”)
        print(“Type: %s” % str(type(e)))
        print(“Error message: %s” % e.message)
        retryCount -= 1
        print(“\n%d/%d retries left\n” % (retryCount, MAX_DISCOVERY_RETRIES))
        print(“Backing off…\n”)
        backOffCore.backOff()

if not discovered:
    print(“Discovery failed after %d retries. Exiting…\n” % (MAX_DISCOVERY_RETRIES))
    sys.exit(-1)

# Iterate through all connection options for the core and use the first successful one
myAWSIoTMQTTClient = AWSIoTMQTTClient(clientId)
myAWSIoTMQTTClient.configureCredentials(groupCA, privateKeyPath, certificatePath)
myAWSIoTMQTTClient.onMessage = customOnMessage

connected = False
for connectivityInfo in coreInfo.connectivityInfoList:
    currentHost = connectivityInfo.host
    currentPort = connectivityInfo.port
    print(“Trying to connect to core at %s:%d” % (currentHost, currentPort))
    myAWSIoTMQTTClient.configureEndpoint(currentHost, currentPort)
    try:
        myAWSIoTMQTTClient.connect()
        connected = True
        break
    except BaseException as e:
        print(“Error in connect!”)
        print(“Type: %s” % str(type(e)))
        print(“Error message: %s” % e.message)

if not connected:
    print(“Cannot connect to core %s. Exiting…” % coreInfo.coreThingArn)
    sys.exit(-2)

# Successfully connected to the core
if args.mode == ‘both’ or args.mode == ‘subscribe’:
    myAWSIoTMQTTClient.subscribe(topic, 0, None)
time.sleep(2)

loopCount = 0
while True:
    if args.mode == ‘both’ or args.mode == ‘publish’:
        message = {}
        sensor_data_json = get_sensor_data()
        # message[‘message’] = args.message
        message[‘message’] = get_message(sensor_data_json)
        message[‘alerts’] = apply_rules(sensor_data_json)
        message[‘sequence’] = loopCount
        messageJson = json.dumps(message)
        myAWSIoTMQTTClient.publish(topic, messageJson, 0)
        if args.mode == ‘publish’:
            print(‘Published topic %s: %s\n’ % (topic, messageJson))
        cloud_topic = “sensors/data/alerts”
        alerts_json = json.dumps(message[‘alerts’])
        myAWSIoTMQTTClient.publish(cloud_topic, alerts_json, 0)
        loopCount += 1
    time.sleep(1)

Code 16. greengrassCommunication.py, Python script to get Sensor Data and Publish the MQTT Messages

Subscriber Setup

In this module, we will configure the Greengrass device to be a MQTT subscriber. On the subscriber device, the non-UP Squared Linux machine, do the same: copy sub’s tar file which was saved in a previous module and untar it, save the files in the publisher device, and rename them for readability:

tar –xzvf <sub-credentials-id>-setup.tar.gz
mv <sub-credentials-id>.cert.pem sub.cert.pem
mv <sub-credentials-id>.private.pem sub.private.pem
mv <sub-credentials-id>.public.pem sub.public.pem

Code 17. Commands to save subscriber credentials

In the subscriber folder, get a root certificate and save it as root-ca-cert.pem:

wget https://www.symantec.com/content/en/us/enterprise/verisign/roots/VeriSign-Class%203-Public-Primary-Certification-Authority-G5.pem –O root-ca-cert.pem

Code 18. Commands to get a root certificate

On the subscriber device, install AWS IoT SDK for Python:

python
>>> import ssl
>>> ssl.OPENSSL_VERSION
# output should be version of OpenSSL 1.0.1+:‘OpenSSL 1.0.2g 1 Mar 2016’
>>> exit()
cd ~ 
git clone https://github.com/aws/aws-iot-device-sdk-python.git
cd aws-iot-device-sdk-python
python setup.py install

Code 19. Commands to install AWS IoT SDK for Python

The subscriber device will listen to sensors/data/pubsub topic continuously, printing “Waiting for the message.” every 10 seconds to confirm the script is still running:

while True:
    print(“Waiting for the message.”)
    time.sleep(10)

Code 20. Code snippet to continuously wait for the message

Copy the following Python script into the publisher device’s folder where publisher keys are stored, and save it as subscription.py:

from __future__ import print_function
import os
import sys
import time
import uuid
import json
import argparse
from AWSIoTPythonSDK.core.greengrass.discovery.providers import DiscoveryInfoProvider
from AWSIoTPythonSDK.core.protocol.connection.cores import ProgressiveBackOffCore
from AWSIoTPythonSDK.MQTTLib import AWSIoTMQTTClient
from AWSIoTPythonSDK.exception.AWSIoTExceptions import DiscoveryInvalidRequestException
import signal, atexit


AllowedActions = [‘both’, ‘publish’, ‘subscribe’]

# General message notification callback
def customOnMessage(message):
    print(‘Received message on topic %s: %s\n’ % (message.topic, message.payload))

MAX_DISCOVERY_RETRIES = 10
GROUP_CA_PATH = “./groupCA/”

# Read in command-line parameters
parser = argparse.ArgumentParser()
parser.add_argument(“-e”, “–endpoint”, action=”store”, required=True, dest=”host”, help=”Your AWS IoT custom endpoint”)
parser.add_argument(“-r”, “–rootCA”, action=”store”, required=True, dest=”rootCAPath”, help=”Root CA file path”)
parser.add_argument(“-c”, “–cert”, action=”store”, dest=”certificatePath”, help=”Certificate file path”)
parser.add_argument(“-k”, “–key”, action=”store”, dest=”privateKeyPath”, help=”Private key file path”)
parser.add_argument(“-n”, “–thingName”, action=”store”, dest=”thingName”, default=”Bot”, help=”Targeted thing name”)
parser.add_argument(“-m”, “–mode”, action=”store”, dest=”mode”, default=”both”,
                    help=”Operation modes: %s”%str(AllowedActions))

args = parser.parse_args()
host = args.host
rootCAPath = args.rootCAPath
certificatePath = args.certificatePath
privateKeyPath = args.privateKeyPath
clientId = args.thingName
thingName = args.thingName
topic = “sensors/data/pubsub”

if args.mode not in AllowedActions:
    parser.error(“Unknown –mode option %s. Must be one of %s” % (args.mode, str(AllowedActions)))
    exit(2)

if not args.certificatePath or not args.privateKeyPath:
    parser.error(“Missing credentials for authentication.”)
    exit(2)

# Progressive back off core
backOffCore = ProgressiveBackOffCore()

# Discover GGCs
discoveryInfoProvider = DiscoveryInfoProvider()
discoveryInfoProvider.configureEndpoint(host)
discoveryInfoProvider.configureCredentials(rootCAPath, certificatePath, privateKeyPath)
discoveryInfoProvider.configureTimeout(10)  # 10 sec

retryCount = MAX_DISCOVERY_RETRIES
discovered = False
groupCA = None
coreInfo = None
while retryCount != 0:
    try:
        discoveryInfo = discoveryInfoProvider.discover(thingName)
        caList = discoveryInfo.getAllCas()
        coreList = discoveryInfo.getAllCores()

        # We only pick the first ca and core info
        groupId, ca = caList[0]
        coreInfo = coreList[0]
        print(“Discovered GGC: %s from Group: %s” % (coreInfo.coreThingArn, groupId))

        print(“Now we persist the connectivity/identity information…”)
        groupCA = GROUP_CA_PATH + groupId + “_CA_” + str(uuid.uuid4()) + “.crt”
        if not os.path.exists(GROUP_CA_PATH):
            os.makedirs(GROUP_CA_PATH)
        groupCAFile = open(groupCA, “w”)
        groupCAFile.write(ca)
        groupCAFile.close()

        discovered = True
        print(“Now proceed to the connecting flow…”)
        break
    except DiscoveryInvalidRequestException as e:
        print(“Invalid discovery request detected!”)
        print(“Type: %s” % str(type(e)))        
​        print(“Error message: %s” % e.message)
        print(“Stopping…”)
        break
    except BaseException as e:
        print(“Error in discovery!”)
        print(“Type: %s” % str(type(e)))
        print(“Error message: %s” % e.message)
        retryCount -= 1
        print(“\n%d/%d retries left\n” % (retryCount, MAX_DISCOVERY_RETRIES))
        print(“Backing off…\n”)
        backOffCore.backOff()

if not discovered:
    print(“Discovery failed after %d retries. Exiting…\n” % (MAX_DISCOVERY_RETRIES))
    sys.exit(-1)

# Iterate through all connection options for the core and use the first successful one
myAWSIoTMQTTClient = AWSIoTMQTTClient(clientId)
myAWSIoTMQTTClient.configureCredentials(groupCA, privateKeyPath, certificatePath)
myAWSIoTMQTTClient.onMessage = customOnMessage

connected = False
for connectivityInfo in coreInfo.connectivityInfoList:
    currentHost = connectivityInfo.host
    currentPort = connectivityInfo.port
    print(“Trying to connect to core at %s:%d” % (currentHost, currentPort))
    myAWSIoTMQTTClient.configureEndpoint(currentHost, currentPort)
    try:
        myAWSIoTMQTTClient.connect()
        connected = True
        break
    except BaseException as e:
        print(“Error in connect!”)
        print(“Type: %s” % str(type(e)))
        print(“Error message: %s” % e.message)

if not connected:
    print(“Cannot connect to core %s. Exiting…” % coreInfo.coreThingArn)
    sys.exit(-2)

# Successfully connected to the core
if args.mode == ‘both’ or args.mode == ‘subscribe’:
    myAWSIoTMQTTClient.subscribe(topic, 0, None)
    print(“after subscription”)
time.sleep(2)

loopCount = 0
while True:
    print(“Waiting for the message.”)
    time.sleep(10)

Code 21. subscription.py, Python script to subscribe to MQTT messages

Run the Scripts

In this module, we will run the Python scripts and view the MQTT messages with sensor data.

On the UP Squared board, start the Greengrass service:

cd <path-to-greengrass>/greengrass/ggc/core
sudo ./greengrassd start

Code 22. Commands to start Greengrass service

Go to the publisher folder:

cd <path-to-publisher-folder>

Code 23. Command to navigate to publisher folder

Get your AWS IoT endpoint ID by going to the AWS console, then to IoT Core page. On the bottom left-side menu, select Settings. Copy your endpoint value:

Settings View
Figure 16. Settings view

Substitute your AWS IoT endpoint ID and run the following command:

python greengrassCommunication.py --endpoint <your-aws-iot-endpoint-id>.iot.us-west-2.amazonaws.com --rootCA root-ca-cert.pem --cert pub.cert.pem --key pub.private.key --thingName pub --mode publish

Code 24. Command to run greengrassCommunication.py

You should see the following screen:

Command to Run greengrassCommunication.py
Figure 17. Command to run greengrassCommunication.py and MQTT messages

On the subscriber device, go to the subscriber folder:

cd <path-to-subscriber-folder>

Code 25. Command to navigate to subscriber folder

Substitute your AWS IoT endpoint ID and run the following command:

python subscription.py --endpoint <your-aws-iot-endpoint-id>.iot.us-west-2.amazonaws.com --rootCA root-ca-cert.pem --cert sub.cert.pem --key sub.private.key --thingName sub --mode subscribe 

Code 26. Command to run subscription.py

You should see the following screen:

Command to Run subscription.py
Figure 18. Command to run subscription.py and received MQTT messages

Go to the AWS IoT console. Select Test from the left-side menu. Type sensors/data/alerts in the topic field, change MQTT payload display to display it as strings, and click Subscribe to topic:

 

MQTT Subscription View
Figure 19. MQTT subscription view

After some time, messages should display on the bottom of the screen:

MQTT Messages View
Figure 20. MQTT messages view

As you can see, the rules were activated for some abnormal sensor readings. You may wish to create some action logic to return the values back to normal. This setup will allow you to monitor your environment and ensure the data readings are in the normal range.

Learn More on UP Squared

About the Author

Rozaliya Everstova is a software engineer at Intel in the Core and Visual Computing Group working on scale enabling projects for Internet of Things.

Для получения подробной информации о возможностях оптимизации компилятора обратитесь к нашему Уведомлению об оптимизации.
Возможность комментирования русскоязычного контента была отключена. Узнать подробнее.