Linked autos telemetry – Processing Protobuf messages with AWS IoT Core

August 22, 2022

Introduction

In join autos purposes, telemetry knowledge is normally very in depth, containing construction and unstructured knowledge. To ship knowledge over to the Cloud you need to use Protocol Buffers (Protobuf – binary format). Protobuf gives the appliance with an environment friendly but properly structured compressing mechanism. The built-in protocol documentation makes knowledge serialization and deserialization extra manageable than JavaScript object notation (JSON). Nonetheless, producer and client should function on an outlined shared schema to encode and decode it correctly.

On this weblog we’ll cowl the most effective practices whereas utilizing Protobuf to encoding and decoding. Additionally, you will be taught step-by-step the right way to use AWS IoT Core and AWS Lambda to ingest and course of Protobuf for consumption

Resolution Structure

Determine 1. Structure diagram

Resolution overview

You’ll simulate a related automobile and authenticate it to AWS IoT Core. The machine will first encode the payload and ship it over Message Queuing Telemetry Transport (MQTT) to AWS IoT Core
As soon as the message is acquired by AWS IoT Core you’ll use AWS IoT Rule which is able to invoke an AWS Lambda operate with the intention to decode the payload.
The rule sends messages to Amazon Kinesis Information Firehouse after which shops it in Amazon S3
Every time a brand new file is written on Easy Storage Service (Amazon S3), AWS Glue Crawler will crawl the info to deduce the schema and make it accessible within the AWS Glue Information Catalog.
We’ll the use Amazon Athena to do Advert-hoc querying an visualize it in Amazon fast sight.

AWS IoT Core

AWS IoT Core securely connects your simulated IoT machine and routes the encoded messages to AWS providers with out managing the underlying infrastructure. You’ll be able to then use guidelines for AWS IoT to decode your payload knowledge and ahead it to Amazon Kinesis Information Firehose.

Amazon Kinesis Information Firehose

Amazon Kinesis Information Firehose captures the incoming knowledge from your rule for AWS IoT and cargo it as batch in parquet format in our Amazon S3 Bucket.

Amazon S3

Amazon S3 serves as an information lake to your knowledge that you need to use for additional evaluation and visualization.

AWS Glue

The AWS Glue Information Catalog is your persistent retailer for the metadata (e.g., schema and placement of the info). It’s a managed service that permits you to retailer, annotate, and share metadata within the AWS Cloud.

For writing information to Amazon S3, you need to use AWS Glue crawler to scan knowledge, classify it, carry out schema extractions, and retailer the metadata robotically within the AWS Glue Information Catalog.

Amazon Athena

Amazon Athena makes use of the metadata classification from AWS Glue Information Catalog to carry out ad-hoc queries on the info.

Amazon QuickSight

You’ll be able to visualize your knowledge and construct a customized dashboard utilizing Amazon QuickSight

Resolution Walkthrough

Pre-requisite

You want a PC with an online browser, ideally with the most recent model of Chrome / FireFox
You could have entry to an AWS account with Administrator Entry privileges
For those who don’t have an AWS Account comply with the directions to create one.
You’ll use Cloud formation template to create the setup setting and you may delete the setting as soon as accomplished
Following AWS providers can be used:
- AWS IoT Core
- Amazon Kinesis Information Firehose
- Amazon S3
- AWS Glue
- Amazon Athena
- Amazon QuickSight
- Amazon Cloud9

Setup resolution

Creating and setup AWS Cloud9 setting

Use the next hyperlink to setup the take a look at setting utilizing AWS Cloud9 for this weblog AWS IoT Gadget Shopper Workshop (IoT quickstart) (workshops.aws). You might decide any area near your location.

Setup AWS IoT Factor and SDK

Open Cloud9 terminal and let’s setup Python SDK for us to make use of.

Create the folder you’ll use to attach the IoT factor utilizing Cloud9 terminal window.

mkdir -p /dwelling/ubuntu/setting/protobuf-python-aws-iot-device/certs
cd /dwelling/ubuntu/setting/protobuf-python-aws-iot-device/

Setup the dependencies:

copy and paste the next necessities.txt

AWSIoTPythonSDK==1.5.2
numpy==1.19.5
protobuf==3.19.4

after which run the next:

python3 -m venv venv
supply ./venv/bin/activate
pip set up -r necessities.txt
deactivate

Setup your AWS IoT Factor comply with steps outlined right here.

As soon as we’ve created the factor let’s add these certificates in our Cloud9 occasion for us to attach from there.

Add the newly created certificates and RootCA into ‘certs’ folder created earlier.

Gadget and Schema

Right here is the Protobuf schema that we are going to use. Create file the next file automotive.proto file and replica and paste the next content material.

syntax = "proto2";

package deal automotive;

message Automotive {
  required float battery_level = 1;
  required float battery_health = 2;
  required float battery_discharge_rate = 3;
  required float wheel_rpm = 4;
  required float mileage_left = 5;
}

You will want to compile and generate the suitable library, right here is the corresponding file you need to use and save into following file automotive_pb2.py

# -*- coding: utf-8 -*-
# Generated by the protocol buffer compiler.  DO NOT EDIT!
# supply: automotive.proto
"""Generated protocol buffer code."""
from google.protobuf import descriptor as _descriptor
from google.protobuf import descriptor_pool as _descriptor_pool
from google.protobuf import message as _message
from google.protobuf import reflection as _reflection
from google.protobuf import symbol_database as _symbol_database
# @@protoc_insertion_point(imports)
_sym_db = _symbol_database.Default()

DESCRIPTOR = _descriptor_pool.Default().AddSerializedFile(b'nx10x61utomotive.protox12nautomotive"x84x01nnAutomotivex12x15nrbattery_levelx18x01 x02(x02x12x16nx0ex62x61ttery_healthx18x02 x02(x02x12x1enx16x62x61ttery_discharge_ratex18x03 x02(x02x12x11ntwheel_rpmx18x04 x02(x02x12x14nx0cmileage_leftx18x05 x02(x02')


_AUTOMOTIVE = DESCRIPTOR.message_types_by_name['Automotive']
Automotive = _reflection.GeneratedProtocolMessageType('Automotive', (_message.Message,), {
  'DESCRIPTOR' : _AUTOMOTIVE,
  '__module__' : 'automotive_pb2'
  # @@protoc_insertion_point(class_scope:automotive.Automotive)
  })
_sym_db.RegisterMessage(Automotive)
if _descriptor._USE_C_DESCRIPTORS == False:
  DESCRIPTOR._options = None
  _AUTOMOTIVE._serialized_start=33
  _AUTOMOTIVE._serialized_end=165
# @@protoc_insertion_point(module_scope)

Let’s create our file that may execute our machine simulation. Copy and paste the next content material in a file named major.py

'''
/*
 * Copyright 2010-2017 Amazon.com, Inc. or its associates. All Rights Reserved.
 *
 * Licensed underneath the Apache License, Model 2.0 (the "License").
 * You might not use this file besides in compliance with the License.
 * A duplicate of the License is situated at
 *
 *  http://aws.amazon.com/apache2.0
 *
 * or within the "license" file accompanying this file. This file is distributed
 * on an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, both
 * categorical or implied. See the License for the particular language governing
 * permissions and limitations underneath the License.
 */
 '''

from AWSIoTPythonSDK.MQTTLib import AWSIoTMQTTClient
import logging
import time
import argparse
import json
import automotive_pb2
import numpy as np

AllowedActions = ['both', 'publish', 'subscribe']

# Customized MQTT message callback
def customCallback(shopper, userdata, message):
    print("Acquired a brand new message: ")
    print(message.payload)
    print("from subject: ")
    print(message.subject)
    print("--------------nn")


# Learn in command-line parameters
parser = argparse.ArgumentParser()
parser.add_argument("-e", "--endpoint", motion="retailer", required=True, dest="host", assist="Your AWS IoT customized endpoint")
parser.add_argument("-r", "--rootCA", motion="retailer", required=True, dest="rootCAPath", assist="Root CA file path")
parser.add_argument("-c", "--cert", motion="retailer", dest="certificatePath", assist="Certificates file path")
parser.add_argument("-k", "--key", motion="retailer", dest="privateKeyPath", assist="Non-public key file path")
parser.add_argument("-p", "--port", motion="retailer", dest="port", kind=int, assist="Port quantity override")
parser.add_argument("-w", "--websocket", motion="store_true", dest="useWebsocket", default=False,
                    assist="Use MQTT over WebSocket")
parser.add_argument("-id", "--clientId", motion="retailer", dest="clientId", default="basicPubSub",
                    assist="Focused shopper id")
parser.add_argument("-t", "--topic", motion="retailer", dest="subject", default="sdk/take a look at/Python", assist="Focused subject")
parser.add_argument("-m", "--mode", motion="retailer", dest="mode", default="each",
                    assist="Operation modes: %s"%str(AllowedActions))
parser.add_argument("-M", "--message", motion="retailer", dest="message", default="Hi there World!",
                    assist="Message to publish")

args = parser.parse_args()
host = args.host
rootCAPath = args.rootCAPath
certificatePath = args.certificatePath
privateKeyPath = args.privateKeyPath
port = args.port
useWebsocket = args.useWebsocket
clientId = args.clientId
subject = args.subject

if args.mode not in AllowedActions:
    parser.error("Unknown --mode choice %s. Should be one among %s" % (args.mode, str(AllowedActions)))
    exit(2)

if args.useWebsocket and args.certificatePath and args.privateKeyPath:
    parser.error("X.509 cert authentication and WebSocket are mutual unique. Please decide one.")
    exit(2)

if not args.useWebsocket and (not args.certificatePath or not args.privateKeyPath):
    parser.error("Lacking credentials for authentication.")
    exit(2)

# Port defaults
if args.useWebsocket and never args.port:  # When no port override for WebSocket, default to 443
    port = 443
if not args.useWebsocket and never args.port:  # When no port override for non-WebSocket, default to 8883
    port = 8883

# Configure logging
logger = logging.getLogger("AWSIoTPythonSDK.core")
logger.setLevel(logging.DEBUG)
streamHandler = logging.StreamHandler()
formatter = logging.Formatter('%(asctime)s - %(title)s - %(levelname)s - %(message)s')
streamHandler.setFormatter(formatter)
logger.addHandler(streamHandler)

# Init AWSIoTMQTTClient
myAWSIoTMQTTClient = None
if useWebsocket:
    myAWSIoTMQTTClient = AWSIoTMQTTClient(clientId, useWebsocket=True)
    myAWSIoTMQTTClient.configureEndpoint(host, port)
    myAWSIoTMQTTClient.configureCredentials(rootCAPath)
else:
    myAWSIoTMQTTClient = AWSIoTMQTTClient(clientId)
    myAWSIoTMQTTClient.configureEndpoint(host, port)
    myAWSIoTMQTTClient.configureCredentials(rootCAPath, privateKeyPath, certificatePath)

# AWSIoTMQTTClient connection configuration
myAWSIoTMQTTClient.configureAutoReconnectBackoffTime(1, 32, 20)
myAWSIoTMQTTClient.configureOfflinePublishQueueing(-1)  # Infinite offline Publish queueing
myAWSIoTMQTTClient.configureDrainingFrequency(2)  # Draining: 2 Hz
myAWSIoTMQTTClient.configureConnectDisconnectTimeout(10)  # 10 sec
myAWSIoTMQTTClient.configureMQTTOperationTimeout(5)  # 5 sec

# Join and subscribe to AWS IoT
myAWSIoTMQTTClient.join()
if args.mode == 'each' or args.mode == 'subscribe':
    myAWSIoTMQTTClient.subscribe(subject, 1, customCallback)
time.sleep(2)

# Publish to the identical subject in a loop ceaselessly
loopCount = 0
automotive = automotive_pb2.Automotive()
dataPointsSin = np.linspace(-np.pi, np.pi,100)
whereas True:
    if args.mode == 'each' or args.mode == 'publish':
        # 100 linearly spaced numbers
        automotive.battery_level = abs(dataPointsSin[loopCount % 100]) / np.pi
        automotive.battery_health = 100 - loopCount % 100
        automotive.battery_discharge_rate = 4.8
        automotive.wheel_rpm = 3000
        automotive.mileage_left = loopCount % 100
        message = bytearray(automotive.SerializeToString())
        myAWSIoTMQTTClient.publish(subject, message, 1)
        if args.mode == 'publish':
            print('Printed subject %s: %sn' % (subject, message))
        loopCount += 1
    time.sleep(1)
    # Copyright Amazon.com, Inc. or its associates. All Rights Reserved.
# SPDX-License-Identifier: Apache-2.0.

Lambda layer

Create a lambda layer that can be used to retailer protobuf libraries, then execute the next command:

mkdir -p ./protobuf/python/customized
cd ./protobuf/python
pip3 set up protobuf --target .                                      
cd customized
cp ../../../automotive_pb2.py ./
echo 'customized' >> ../protobuf-*.dist-info/namespace_packages.txt
echo 'customized/demo_pb2.py' >> ../protobuf-*.dist-info/RECORD
echo 'customized' >> ../protobuf-*.dist-info/top_level.txt
cd ../../
zip -r protobuf.zip .
aws lambda publish-layer-version --layer-name protobuf --zip-file fileb://protobuf.zip --compatible-runtimes python3.8

Setup S3 bucket

We’ll in a primary time create an S3 bucket the place will retailer our knowledge and which we’ll question it from. Create an S3 bucket, fill out a reputation, for the title can be “ACCOUNT-ID-connected-evs”, go away the remainder as default. and click on on create bucket. Please word your bucket title as we’ll reusing it by way of the entire this weblog.

Determine 2.

Setup Kinesis knowledge firehose

Create a supply stream, the supply stream will write the acquired knowledge from the related vehicles to Amazon S3. Choose supply Direct PUT, vacation spot Amazon S3, and fill out a reputation, for me it will likely be: “my-delivery-stream-connected-evs“.

Determine 3.

In vacation spot settings, choose the S3 bucket that you simply beforehand created, as bucket prefix, fill out “uncooked“ and error prefix as ”errors“. Pass over the remainder as default and wait jiffy earlier than this completes.

Determine 4.

Setup AWS IoT Rule

Create the AWS IoT Rule, we’ll use the IoT rule through the lambda creation, please word your rule title. It is advisable to choose all knowledge coming from the subject connected-cars/knowledge, after which invoke the incoming knowledge with a lambda operate with the intention to decode the protobuf encoded payload. You first must encode the binary string in base64. For the SQL assertion please copy and paste the next, please substitute ACCOUNT_ID together with your account ID

Choose Message Routing

Choose Guidelines
Choose Create rule
Give Rule title (i.e. we’re utilizing “MyConnectedEVSRuleToFirehose”)
Give Rule description
Use the next question for the rule: SELECT aws_lambda("arn:aws:lambda:us-east-1:ACCOUNT_ID:operate:my-protobuf-decoder", {"knowledge": encode(*, "base64"), "clientId": clientId()}) as payload, timestamp() as p_time FROM 'connected-cars/knowledge'
Choose Subsequent
In Connect rule actions
- Choose settings as per Determine 6
- Choose Add motion
Choose Subsequent
In Overview and Create

Determine 5.

Determine 6.

Setup lambda

Create AWS Lambda operate and provides the identical title as earlier when creating AWS IoT Rule. Decide Python 3.8 for runtime.

Determine 7.

After creating the layer for the protobuf half, please use the next code:

import json
import base64
from customized import automotive_pb2

print('Loading operate')

def lambda_handler(occasion, context):
    print("Acquired occasion: " + json.dumps(occasion, indent=2))

    ret = {}    

    knowledge = occasion["data"]
    payload_data_decoded = base64.b64decode(knowledge)
    automotive = automotive_pb2.Automotive()
    automotive.ParseFromString(payload_data_decoded)
    elems = automotive.ListFields()
    for elem in elems:
        ret[elem[0].title] = elem[1]
    ret["clientId"] = occasion["clientId"]
    return ret

Within the configuration tab and permissions, go to the resource-based coverage and click on on add permission. We have to add the mandatory permission to permit the iot rule to invoke our operate. when specifying the arn, please use the identical title for the rule you created. Click on on save.

Determine 8.

Lastly, we’ll use the beforehand created layer, for that, go within the layer half and Choose ‘Add a layer‘.

Determine 9.

Determine 10.

Protobuf decode/encode

Following JSON can be used for additional encoding as Protobuf binary message.

{
    "battery_level": 100,
    "battery_health" : 50,
    "battery_discharge_rate" : 4.8,
    "wheel_rpm" : 3000,
    "mileage_left" : 88
}

Let’s publish our first message and test if all the pieces is working:

Pattern command:

Utilizing certificates and AWS IoT factor created earlier, these certificates used within the parameter to ship the message (substitute xxxx with related values to your setup).

supply ./venv/bin/activate
python3 major.py -e xxxx-ats.iot.us-east-1.amazonaws.com -c ./certs/xxxx-certificate.pem.crt -r ./certs/AmazonRootCA1.pem -t connected-cars/knowledge -m eza -k ./certs/xxxx-private.pem.key --mode publish

Determine 11.

Go to your bucket after jiffy, you must see information being written.

Setup AWS Glue Crawler

We’re going to now create the Glue Crawler that can be liable for creating and updating the schema of our knowledge. You’ll be able to create a crawler on the next hyperlink. For crawler title, mine can be: ‘my_connected_evs_crawler’.

For the Crawler supply kind decide Information shops, for Repeat crawls of S3 knowledge shops decide Crawl all folders. Within the Add Information retailer go away all the pieces by default however for the embody path choose your S3 bucket and the uncooked folder. For me it will likely be s3://ACCOUNT_ID-connected-evs/uncooked. click on on subsequent. Don’t add one other datastore. Give a reputation to your function. For the frequency go away as default.

For Configure the crawler’s output, click on on add database, add a database title my_connected_evs_db and go away the remainder clean. Depart the remainder as default and click on subsequent.

Choose your crawler, and click on on run your crawler. The standing of your crawler ought to be displaying beginning, When the standing of your crawler is stopping, go test your desk in your database. It’s best to see the next to your uncooked desk:

Determine 12.

Setup Amazon Athena

Go to the Amazon Athena console, you possibly can setup your Amazon Athena question outcomes by following this hyperlink.

Choose your database and desk that you simply used for the crawler. Run the next question:

SELECT * FROM uncooked;

Determine 13.

Visualize knowledge utilizing Amazon QuickSight

To setup QuickSight, please comply with this hyperlink.

In QuickSight, let’s first create our dataset. Click on on Dataset on the left. The supply of our dataset can be Amazon Athena that we used beforehand to preview our knowledge. If you wish to test the opposite sources which might be supported, please comply with the next hyperlink. Please word that in our case we use Amazon Athena for simplicity to do advert hoc querying and fast dash-boarding.

Determine 14.

On the next display, click on on new dataset.

Determine 15.

Then click on on Athena.

Determine 16.

Then give a reputation to your knowledge supply, for us, it will likely be: ‘my-evs-athena-data-source’. Make to certain to the validate connection. Then click on on Create Information supply.

Determine 17.

Selected the AwsDataCatalog and our db my_connected_evs_db and the uncooked desk. Click on on Use customized SQL.

Determine 18.

We’ll flatten the payload struct with the next question. Copy and paste the question and title the customized SQL question and click on on Verify question.

SELECT payload.battery_level as battery_level, payload.battery_health as battery_health, payload.battery_discharge_rate as battery_discharge_rate, payload.wheel_rpm as wheel_rpm, payload.mileage_left as mileage_left, p_time, payload.clientId as client_id FROM "my_connected_evs_db"."uncooked";

Determine 19.

Depart the remainder by default and click on on visualize.

Determine 20.

Listed here are some examples of visualization

Click on on the decrease left hand facet on the desk diagram and on every dimension. It’s best to see your knowledge in a desk format.

Determine 21.

Conclusion

On this weblog submit we took a pattern JSON payload and encoded into Protobuf binary format and despatched it over to AWS IoT Core, the place we deserialized and decoded utilizing AWS Lambda.The info was then our knowledge lake in Amazon S3 utilizing Kinesis Datafirehose, lastly utilizing this knowledge we visualized our telemetry knowledge of this related car. By following this weblog, you realized how one can compress, serialize and deserialize JSON dataset of related autos. Utilizing this system you possibly can obtain decrease latency together with compatibility between your finish machine and your client.

Concerning the Authors

Syed Rehan is a International Sr. Specialist Options Architect at Amazon Net Providers and is predicated in London. He’s protecting world span of shoppers and supporting them as lead IoT Resolution Architect. Syed has in-depth information of IoT and cloud and works on this function with world clients starting from start-up to enterprises to allow them to construct IoT options with the AWS eco system.

Kevin Polossat is a Options Architect at AWS. He works with clients in France to assist them embrace and undertake the cloud. Outdoors of labor, he enjoys wine and cheese.