Workflow Monitoring -Dialog Flow and AWS Lambda
In this tutorial, we going to connect Google Assistant to Amazon AWS through Dialog Flow.
The goal is to make a program that can manage and respond to a user voice request.
Following are the things we are going to use:
Dialog flow is a Google-owned developer of human-computer interaction technologies based on natural language conversations. Give users new ways to interact with your product by building engaging voice and text-based conversational interfaces, such as voice apps and chatbots, powered by AI. Connect with users on your website, mobile app, the Google Assistant, Amazon Alexa, Facebook Messenger, and other popular platforms and devices.
Components we are going to use here:
User: We, Machines!
Text / Voice: The user interacts with an app like smartwatch/google home.
Agents: Help convert user requests into actionable data.
Intent: Support or the service that the user wants from the agent. The intent is configured by the developers. The intent determines the action by the code.
Fulfillment: This is a web-hook. This part of the conversation lets you pass on the request from your app to an external source and get a response and pass it back to the user. Setting up a webhook allows you to pass information from a matched intent into a web service and get a result from it.
AWS API Gateway:
- It allows for creating, publishing, maintaining, monitoring, and securing REST and WebSocket APIs at any scale.
- It provides multiple options for controlling access to your REST and WebSocket APIs.
- It implements standard HTTP methods such as GET, POST, PUT, PATCH, and DELETE.
Amazon DynamoDB:
- It is a hosted NoSQL database.
- It offers reliable and scalable performance.
- With a small, simple API, it allows for simple key-value access.
AWS Lambda
- It is an event-driven, serverless computing platform.
- It is a computing service that automatically manages the computing resources required by the code.
- AWS targets starting a Lambda instance within milliseconds of an event.
AWS IoT
- Managed Service
- No installation
- Automatic Scaling
- Redundant
- Message Broker
- Rule Engine
- Registry
The flow of the program
Step 1: Setting up DialogFlow account
- Go to https://dialogflow.com/
- Create an account with a Gmail account, and “agree” to the terms & conditions.
- Build the Agent: Agents translate user requests to actionable data i.e. intents. It’s essentially a module within dialog flow which incorporates Natural Language Processing to understand what the user meant and to figure out what “action” has to be carried out.
- To create an agent following are the steps
Entities:
Entities are used to extract parameter values from user queries. So when the user says “Please tell me the temperature ” we usually ask for the region, date, time, state etc.these are called entities.
Here our Entities will be called region which will be the name of the device 1, One, etc.
Fulfillment: Our Amazon AWS web service will receive a POST request from Dialogflow in the form of the response to a user query matched by intents with webhook enabled.
In the next part, we will explain how to configure AWS to receive Dialogflow POST requests and be able to respond.
AWS — Walkthrough
Create an AWS user and get in the AWS console.
STEP: 2
Create an IoT certificate to link the device to AWS IoT Cloud
When you created a certificate you will get certificates I.e private key, public key verisign.pem, certificate.pem file. Please save and use it when you send data from the device to AWS IoT Core.
STEP 3: Create a policy to work certificate:
Now attach that policy to the certificate and activate the certificate
Create an AWS IoT Rule you can find this in IoT Core → Act
Next, we should open the IoT console and link an MQTT topic to a certain rule which will trigger some actions. So we click on the “Act” tab, and we create a rule.
Set rule query statement
We define the rule query statement which will identify to which MQTT topic we are linked. We want to listen to certain topic messages and insert in the DB a new order
DynamoDB
First, let’s open the DynamoDB console, and let’s create a DB table called “IoTDeviceData” where we will store all the sensors data for a certain time. We will insert the primary key and the sort key.
Create a Lambda function
Now that we created an MQTT topic and an action we need to actually publish that topic. We will do so by using an AWS Lambda Function. Our function will be called by google assistant, our google action will send us a JSON request asking for certain tire size.
So next we need to open the Lambda Console and we create a lambda function from scratch, choosing as a language Python.
Allow Lambda to access AWS IoT:
For the lambda function to access DynamoDB or AWS IoT, it should have a particular role. We need to go to the IAM Console (Identity Management Console). We will see that a role for our function was already created in the list. We click on that role and we will attach an inline policy by pasting this JSON which will permit our function to publish on an IoT topic.
By refreshing our lambda page we can see that on the “output” side of the function we now have AWS IoT as a possible “output”.
Allow Lambda to access DynamoDB
Still, on the IAM Console on our function role, we will then add the micro-services policy which will give us the DynamoDB access permission.
Write the Lambda function
We now start by writing our function. We parse the JSON request looking for the tire size requested, we access the DB in the “stock” table (which we created earlier with some items, at the end of the post you can see a snapshot) to see if a tire with that size exists. If it exists we will respond to google with the model name and a picture of it.
Otherwise, we will answer Google Assistant by saying that we couldn’t find anything and we will publish an outOfStock event by requesting a certain size and a certain quantity.
Here’s the full code:
from __future__ import print_function
import json
from pprint import pprint
import boto3
from botocore.exceptions import ClientError
from boto3.dynamodb.conditions import Key,Attr
import datetime
dynamodb = boto3.resource('dynamodb')
table = dynamodb.Table('assignment_1')def lambda_handler(event, context):
print("Thish is event:",event)
s=(event['queryResult']['action'])
para = event["queryResult"]["parameters"]
print("This is parameter data: ",para)
if (s=='input.getTemperature'):
date1 = para['date']
date2 = date1[0:10]
time = para['date-time']
time1=time[11:19]
print("DATE:::",date2)
print("TIME::",time1)
response = table.query(
KeyConditionExpression=Key('Device_ID').eq(para['deviceid']) & Key('time').eq(time1)
)
a= response['Items'][0]
response1 = {
"fulfillmentMessages": [
{
"text": {
"text": [
" The " + para['attributes'] + " value of device "+ para['deviceid'] + " is " +str(a['temperature'])+" degree centigrade"
]
}
}
]
}
return (response1)
if(s=='input.deviceState'):
state= para['statedevice']
deviceId= para['deviceid']
client = boto3.client('iot-data', region_name='us-east-1')
response = client.publish(
topic='assignment',
qos=1,
payload=json.dumps({"deviceid": para['deviceid'],"state": para['statedevice']})
)
response2 = {
"fulfillmentMessages": [
{
"text": {
"text": [
" The state of device " + str(deviceId) + " is "+ str(state) + "."
]
}
}
]
}
return (response2)
Authorizer Lambda in NodeJS 12.x
/*
* Copyright 2015-2016 Amazon.com, Inc. or its affiliates. All Rights Reserved.
*
* Licensed under the Apache License, Version 2.0 (the "License"). You may not use this file except in compliance with the License. A copy of the License is located at
*
* http://aws.amazon.com/apache2.0/
*
* or in the "license" file accompanying this file. This file is distributed on an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the License for the specific language governing permissions and limitations under the License.
*/
console.log('Loading function');exports.handler = function(event, context, callback) {
console.log('Client token: ' + event.authorizationToken);
console.log('Method ARN: ' + event.methodArn);// validate the incoming token
// and produce the principal user identifier associated with the token// this could be accomplished in a number of ways:
// 1. Call out to OAuth provider
// 2. Decode a JWT token inline
// 3. Lookup in a self-managed DB
var principalId = 'user|a1b2c3d4'// you can send a 401 Unauthorized response to the client by failing like so:
// callback("Unauthorized", null);// if the token is valid, a policy must be generated which will allow or deny access to the client// if access is denied, the client will receive a 403 Access Denied response
// if access is allowed, API Gateway will proceed with the backend integration configured on the method that was called// build apiOptions for the AuthPolicy
var apiOptions = {};
var tmp = event.methodArn.split(':');
var apiGatewayArnTmp = tmp[5].split('/');
var awsAccountId = tmp[4];
apiOptions.region = tmp[3];
apiOptions.restApiId = apiGatewayArnTmp[0];
apiOptions.stage = apiGatewayArnTmp[1];
var method = apiGatewayArnTmp[2];
var resource = '/'; // root resource
if (apiGatewayArnTmp[3]) {
resource += apiGatewayArnTmp.slice(3, apiGatewayArnTmp.length).join('/');
}// this function must generate a policy that is associated with the recognized principal user identifier.
// depending on your use case, you might store policies in a DB, or generate them on the fly// keep in mind, the policy is cached for 5 minutes by default (TTL is configurable in the authorizer)
// and will apply to subsequent calls to any method/resource in the RestApi
// made with the same token// the example policy below denies access to all resources in the RestApi
var policy = new AuthPolicy(principalId, awsAccountId, apiOptions);
let data = event.authorizationToken.split(' ')[1];
let buff = new Buffer(data, 'base64');
let text = buff.toString('ascii');
console.log('Authorization Decoded Text: ' + text);
let username = text.split(':')[0];
let password = text.split(':')[1];
if (username === 'admin' && password === 'admin') {
policy.allowAllMethods();
} else {
policy.denyAllMethods();
}
// policy.allowMethod(AuthPolicy.HttpVerb.GET, "/users/username");// finally, build the policy
var authResponse = policy.build();// new! -- add additional key-value pairs
// these are made available by APIGW like so: $context.authorizer.<key>
// additional context is cached
authResponse.context = {
key : 'value', // $context.authorizer.key -> value
number : 1,
bool: true
};
// authResponse.context.arr = ['foo']; <- this is invalid, APIGW will not accept it
// authResponse.context.obj = {'foo':'bar'}; <- also invalid
console.log(JSON.stringify(authResponse));
callback(null, authResponse);
};/**
* AuthPolicy receives a set of allowed and denied methods and generates a valid
* AWS policy for the API Gateway authorizer. The constructor receives the calling
* user principal, the AWS account ID of the API owner, and an apiOptions object.
* The apiOptions can contain an API Gateway RestApi Id, a region for the RestApi, and a
* stage that calls should be allowed/denied for. For example
* {
* restApiId: "xxxxxxxxxx",
* region: "us-east-1",
* stage: "dev"
* }
*
* var testPolicy = new AuthPolicy("[principal user identifier]", "[AWS account id]", apiOptions);
* testPolicy.allowMethod(AuthPolicy.HttpVerb.GET, "/users/username");
* testPolicy.denyMethod(AuthPolicy.HttpVerb.POST, "/pets");
* context.succeed(testPolicy.build());
*
* @class AuthPolicy
* @constructor
*/
function AuthPolicy(principal, awsAccountId, apiOptions) {
/**
* The AWS account id the policy will be generated for. This is used to create
* the method ARNs.
*
* @property awsAccountId
* @type {String}
*/
this.awsAccountId = awsAccountId;/**
* The principal used for the policy, this should be a unique identifier for
* the end user.
*
* @property principalId
* @type {String}
*/
this.principalId = principal;/**
* The policy version used for the evaluation. This should always be "2012-10-17"
*
* @property version
* @type {String}
* @default "2012-10-17"
*/
this.version = "2012-10-17";/**
* The regular expression used to validate resource paths for the policy
*
* @property pathRegex
* @type {RegExp}
* @default '^\/[/.a-zA-Z0-9-\*]+$'
*/
this.pathRegex = new RegExp('^[/.a-zA-Z0-9-\*]+$');// these are the internal lists of allowed and denied methods. These are lists
// of objects and each object has 2 properties: A resource ARN and a nullable
// conditions statement.
// the build method processes these lists and generates the approriate
// statements for the final policy
this.allowMethods = [];
this.denyMethods = [];if (!apiOptions || !apiOptions.restApiId) {
this.restApiId = "*";
} else {
this.restApiId = apiOptions.restApiId;
}
if (!apiOptions || !apiOptions.region) {
this.region = "*";
} else {
this.region = apiOptions.region;
}
if (!apiOptions || !apiOptions.stage) {
this.stage = "*";
} else {
this.stage = apiOptions.stage;
}
};/**
* A set of existing HTTP verbs supported by API Gateway. This property is here
* only to avoid spelling mistakes in the policy.
*
* @property HttpVerb
* @type {Object}
*/
AuthPolicy.HttpVerb = {
GET : "GET",
POST : "POST",
PUT : "PUT",
PATCH : "PATCH",
HEAD : "HEAD",
DELETE : "DELETE",
OPTIONS : "OPTIONS",
ALL : "*"
};AuthPolicy.prototype = (function() {
/**
* Adds a method to the internal lists of allowed or denied methods. Each object in
* the internal list contains a resource ARN and a condition statement. The condition
* statement can be null.
*
* @method addMethod
* @param {String} The effect for the policy. This can only be "Allow" or "Deny".
* @param {String} he HTTP verb for the method, this should ideally come from the
* AuthPolicy.HttpVerb object to avoid spelling mistakes
* @param {String} The resource path. For example "/pets"
* @param {Object} The conditions object in the format specified by the AWS docs.
* @return {void}
*/
var addMethod = function(effect, verb, resource, conditions) {
if (verb != "*" && !AuthPolicy.HttpVerb.hasOwnProperty(verb)) {
throw new Error("Invalid HTTP verb " + verb + ". Allowed verbs in AuthPolicy.HttpVerb");
}if (!this.pathRegex.test(resource)) {
throw new Error("Invalid resource path: " + resource + ". Path should match " + this.pathRegex);
}var cleanedResource = resource;
if (resource.substring(0, 1) == "/") {
cleanedResource = resource.substring(1, resource.length);
}
var resourceArn = "arn:aws:execute-api:" +
this.region + ":" +
this.awsAccountId + ":" +
this.restApiId + "/" +
this.stage + "/" +
verb + "/" +
cleanedResource;if (effect.toLowerCase() == "allow") {
this.allowMethods.push({
resourceArn: resourceArn,
conditions: conditions
});
} else if (effect.toLowerCase() == "deny") {
this.denyMethods.push({
resourceArn: resourceArn,
conditions: conditions
})
}
};/**
* Returns an empty statement object prepopulated with the correct action and the
* desired effect.
*
* @method getEmptyStatement
* @param {String} The effect of the statement, this can be "Allow" or "Deny"
* @return {Object} An empty statement object with the Action, Effect, and Resource
* properties prepopulated.
*/
var getEmptyStatement = function(effect) {
effect = effect.substring(0, 1).toUpperCase() + effect.substring(1, effect.length).toLowerCase();
var statement = {};
statement.Action = "execute-api:Invoke";
statement.Effect = effect;
statement.Resource = [];return statement;
};/**
* This function loops over an array of objects containing a resourceArn and
* conditions statement and generates the array of statements for the policy.
*
* @method getStatementsForEffect
* @param {String} The desired effect. This can be "Allow" or "Deny"
* @param {Array} An array of method objects containing the ARN of the resource
* and the conditions for the policy
* @return {Array} an array of formatted statements for the policy.
*/
var getStatementsForEffect = function(effect, methods) {
var statements = [];if (methods.length > 0) {
var statement = getEmptyStatement(effect);for (var i = 0; i < methods.length; i++) {
var curMethod = methods[i];
if (curMethod.conditions === null || curMethod.conditions.length === 0) {
statement.Resource.push(curMethod.resourceArn);
} else {
var conditionalStatement = getEmptyStatement(effect);
conditionalStatement.Resource.push(curMethod.resourceArn);
conditionalStatement.Condition = curMethod.conditions;
statements.push(conditionalStatement);
}
}if (statement.Resource !== null && statement.Resource.length > 0) {
statements.push(statement);
}
}return statements;
};return {
constructor: AuthPolicy,/**
* Adds an allow "*" statement to the policy.
*
* @method allowAllMethods
*/
allowAllMethods: function() {
addMethod.call(this, "allow", "*", "*", null);
},/**
* Adds a deny "*" statement to the policy.
*
* @method denyAllMethods
*/
denyAllMethods: function() {
addMethod.call(this, "deny", "*", "*", null);
},/**
* Adds an API Gateway method (Http verb + Resource path) to the list of allowed
* methods for the policy
*
* @method allowMethod
* @param {String} The HTTP verb for the method, this should ideally come from the
* AuthPolicy.HttpVerb object to avoid spelling mistakes
* @param {string} The resource path. For example "/pets"
* @return {void}
*/
allowMethod: function(verb, resource) {
addMethod.call(this, "allow", verb, resource, null);
},/**
* Adds an API Gateway method (Http verb + Resource path) to the list of denied
* methods for the policy
*
* @method denyMethod
* @param {String} The HTTP verb for the method, this should ideally come from the
* AuthPolicy.HttpVerb object to avoid spelling mistakes
* @param {string} The resource path. For example "/pets"
* @return {void}
*/
denyMethod : function(verb, resource) {
addMethod.call(this, "deny", verb, resource, null);
},/**
* Adds an API Gateway method (Http verb + Resource path) to the list of allowed
* methods and includes a condition for the policy statement. More on AWS policy
* conditions here: http://docs.aws.amazon.com/IAM/latest/UserGuide/reference_policies_elements.html#Condition
*
* @method allowMethodWithConditions
* @param {String} The HTTP verb for the method, this should ideally come from the
* AuthPolicy.HttpVerb object to avoid spelling mistakes
* @param {string} The resource path. For example "/pets"
* @param {Object} The conditions object in the format specified by the AWS docs
* @return {void}
*/
allowMethodWithConditions: function(verb, resource, conditions) {
addMethod.call(this, "allow", verb, resource, conditions);
},/**
* Adds an API Gateway method (Http verb + Resource path) to the list of denied
* methods and includes a condition for the policy statement. More on AWS policy
* conditions here: http://docs.aws.amazon.com/IAM/latest/UserGuide/reference_policies_elements.html#Condition
*
* @method denyMethodWithConditions
* @param {String} The HTTP verb for the method, this should ideally come from the
* AuthPolicy.HttpVerb object to avoid spelling mistakes
* @param {string} The resource path. For example "/pets"
* @param {Object} The conditions object in the format specified by the AWS docs
* @return {void}
*/
denyMethodWithConditions : function(verb, resource, conditions) {
addMethod.call(this, "deny", verb, resource, conditions);
},/**
* Generates the policy document based on the internal lists of allowed and denied
* conditions. This will generate a policy with two main statements for the effect:
* one statement for Allow and one statement for Deny.
* Methods that includes conditions will have their own statement in the policy.
*
* @method build
* @return {Object} The policy object that can be serialized to JSON.
*/
build: function() {
if ((!this.allowMethods || this.allowMethods.length === 0) &&
(!this.denyMethods || this.denyMethods.length === 0)) {
throw new Error("No statements defined for the policy");
}var policy = {};
policy.principalId = this.principalId;
var doc = {};
doc.Version = this.version;
doc.Statement = [];doc.Statement = doc.Statement.concat(getStatementsForEffect.call(this, "Allow", this.allowMethods));
doc.Statement = doc.Statement.concat(getStatementsForEffect.call(this, "Deny", this.denyMethods));policy.policyDocument = doc;return policy;
}
};})();
REST API creation
In our last step, we need to give Google Dialogflow a link to which sends its requests. Google can only send POST requests so we open the API Gateway console of AWS and we create our API, choosing the protocol and a name.
API Resource creation
Then we click on “Actions” and “Create new Method”
Now select your lambda:
Now create authorizer it totally depends upon us to create authorizer:
Attach Authorizer with API method:
Deploy API:
After deployment you will see stages:
Pass this API gateway deployment URL to the Dialogflow fulfillment and pass basic auth credentials over here:
Try it from the test in the extreme right and the training phrases are
1. please on device 12. please off device 13. similarly for other devices like 2,3,4,54. what is the temperature of 1 on 2020-10-15 at 07:57:46 AM5. Similarly for other device 1,2,3,4,5
And also run device side code which contains both subscribe as well as publish code. whenever we call please on device 1 then it will call to API Gateway then API Gateway call Lambda and in lambda we have written publish code, Lambda will publish payload to device via AWS IoT.
from time import sleep
import random
import paho.mqtt.client as mqtt
import datetime
import time
import paho.mqtt.client as paho
import os
import socket
import ssl
import jsondef on_connect(client, userdata, flags, rc):
print("Connection returned result: " + str(rc) )
client.subscribe("assignment" , 1 )def on_message(client, userdata, msg):
print("topic: "+msg.topic)
#print("message",msg)
data= json.loads(msg.payload)
#print("data",data)
#print("type:",type(data))
print("State:",data['state'])
print("DeviceID:",data['deviceid'])
#print("payload: ",str(msg.payload))
#test_string= str(msg.payload)
#res = json.loads(test_string)
#print("The converted dictionary : " + str(res))
if(data['state']== "ON"):
current_date = str(datetime.datetime.now().date())
print(current_date)
e = datetime.datetime.now()
message = {}
message['timestamp'] =int(time.time())
message['Device_ID'] = data['deviceid'] #str(random.randint(1, 5))
message['time'] = time.strftime("%H:%M:%S")
message['temperature'] = random.randint(30, 50)
message['Humidity'] = random.randint(30, 50)
message['date'] = e.strftime("%Y-%m-%d")
message['state'] = data['state'] #'off'
messageJson = json.dumps(message)
print(messageJson)
client.publish("topic", messageJson)
time.sleep(0.5)
elif(data['state']== "OFF"):
current_date = str(datetime.datetime.now().date())
print(current_date)
e = datetime.datetime.now()
message = {}
message['timestamp'] =int(time.time())
message['Device_ID'] = data['deviceid'] . message['time'] = time.strftime("%H:%M:%S")
message['temperature'] = random.randint(30, 50)
message['Humidity'] = random.randint(30, 50)
message['date'] = e.strftime("%Y-%m-%d")
message['state'] = data['state'] #'off'
messageJson = json.dumps(message)
print(messageJson)
client.publish("topic", messageJson)mqttc = paho.Client()
mqttc.on_connect = on_connect
mqttc.on_message = on_message
client = mqtt.Client("awsiot")
client.on_connect=on_connect
client.on_message=on_messageawshost = "XXXXXXXXXXXXXX.iot.us-east-1.amazonaws.com"
awsport = 8883
clientId = "assignment"
thingName = "assignment"
caPath = "VeriSign.pem"
certPath = "XXXXXXXXXXXX-certificate.pem.crt"
keyPath = "XXXXXXXXXX-private.pem.key"mqttc.tls_set(caPath, certfile=certPath, keyfile=keyPath, cert_reqs=ssl.CERT_REQUIRED, tls_version=ssl.PROTOCOL_TLSv1_2, ciphers=None)mqttc.connect(awshost, awsport, keepalive=60)mqttc.loop_forever()
The output will be sended to AWS IoT and then that data is saved to AWS Dynamo DB using rule in AWS IoT
