Azure
/
Verified-Telemetry-Solution-Sample
зеркало из https://github.com/Azure/Verified-Telemetry-Solution-Sample.git
1
0
Форкнуть 0
Код Задачи Пакеты Проекты Релизы Вики Активность

lint format (#6) 

* lint format

* remove unused comments, set dashboard refresh to 1s.

* move app code into app dir

* add 1s default refresh for dashboard

* dont bring services up by default
This commit is contained in:
Ryan Winter 2021-05-12 11:26:31 -07:00 коммит произвёл GitHub
Родитель 826d54c87f
Коммит e3731a6734
Не найден ключ, соответствующий данной подписи
Идентификатор ключа GPG: 4AEE18F83AFDEB23
22 изменённых файлов: 466 добавлений и 376 удалений
Показать статистику Скачать Patch файл Скачать Diff файл Показать все файлы Свернуть все файлы

4
.devcontainer/devcontainer.json
Просмотреть файл

@ -3,10 +3,10 @@
{
"name": "Node.js",
// Compose settings
"dockerComposeFile": "../docker-compose.yml",
"runServices": [],
"service": "web",
"workspaceFolder": "/home/app",
// Set *default* container specific settings.json values on container create.

20
.eslintrc.json Normal file
Просмотреть файл

@ -0,0 +1,20 @@
{
"env": {
"browser": true,
"es2021": true
},
"extends": [
"standard"
],
"parserOptions": {
"ecmaVersion": 12,
"sourceType": "module"
},
"rules": {
"no-console": "off",
"no-prototype-builtins": "off",
"indent": [ "error", 4 ],
"brace-style": [ "error", "allman" ],
"max-len": [ "warn", { "code": 130 } ]
}
}

1
.gitignore поставляемый
Просмотреть файл

@ -50,3 +50,4 @@ modules.order
Module.symvers
Mkfile.old
dkms.conf
node_modules

8
Dockerfile
Просмотреть файл

@ -1,8 +0,0 @@
FROM node:10.23.1-alpine
COPY ./package.json /home/app/package.json
COPY . /home/app/
WORKDIR /home/app
RUN npm install
EXPOSE 8080
ENTRYPOINT ["/usr/local/bin/node", "index.js"]

88
HTTPServer.js
Просмотреть файл

@ -1,88 +0,0 @@
// Copyright (c) Microsoft Corporation.
// Licensed under the MIT License.
var constants = require('./constants');
var http = require('http');
let { inspect } = require('util');
var url = require('url');
var PORT = 8080;
var TYPE_PROPERTY = "Property"
var TYPE_COMMAND = "Command"
var ROOT_COMPONENT = "Root"
let propertiesCommandsAPI = function (dtServiceclient){
http.createServer(function (req, res) {
res.writeHead(200, {});
res.end();
console.log("Recieved URL: %s",req.url)
var query = url.parse(req.url, true).query
console.log(query.type);
console.log(query.componentName);
console.log(query.methodName);
console.log(query.value);
if(query.type == TYPE_PROPERTY){
UpdateDigitalTwin(dtServiceclient, query.componentName,query.methodName,query.value);
console.log('Updating Property..');
}
else if(query.type == TYPE_COMMAND){
SendCommand(dtServiceclient, query.componentName,query.methodName,query.value);
console.log('Updating Command..');
}
res.end()
}).listen(PORT);
};
async function UpdateDigitalTwin(dtServiceclient, componentName,propertyName,propertyValue) {
var patch;
if(componentName == ROOT_COMPONENT)
{
patch = [{
op: 'add',
path: '/' + propertyName,
value: JSON.parse(propertyValue)
}];
}
else
{
patch = [{
op: 'add',
path: '/' + componentName +'/'+ propertyName,
value: JSON.parse(propertyValue)
}];
}
console.log(patch);
await dtServiceclient.updateDigitalTwin(constants.deviceId, patch);
console.log('Patch has been successfully applied');
};
async function SendCommand(dtServiceclient, componentName,commandName,commandValue) {
const options = {
connectTimeoutInSeconds: 0,
responseTimeoutInSeconds: 30 // The responseTimeoutInSeconds must be within [5; 300]
};
var commandResponse;
if(componentName == ROOT_COMPONENT)
{
commandResponse = await dtServiceclient.invokeCommand(constants.deviceId, commandName, JSON.parse(commandValue), options);
}
else
{
commandResponse = await dtServiceclient.invokeComponentCommand(constants.deviceId, componentName, commandName, JSON.parse(commandValue), options);
}
// Print result of the command
console.log(inspect(commandResponse));
};
module.exports = { propertiesCommandsAPI:propertiesCommandsAPI }

8
app/Dockerfile Normal file
Просмотреть файл

@ -0,0 +1,8 @@
FROM node:10.23.1-alpine
ADD . /home/app
WORKDIR /home/app
EXPOSE 8080
RUN npm install
CMD node index.js

89
app/HTTPServer.js Normal file
Просмотреть файл

@ -0,0 +1,89 @@
// Copyright (c) Microsoft Corporation.
// Licensed under the MIT License.
const constants = require('./constants')
const http = require('http')
const { inspect } = require('util')
const url = require('url')
const PORT = 8080
const TYPE_PROPERTY = 'Property'
const TYPE_COMMAND = 'Command'
const ROOT_COMPONENT = 'Root'
const propertiesCommandsAPI = function (dtServiceclient)
{
http.createServer(function (req, res)
{
res.writeHead(200, {})
res.end()
console.log('Received URL: %s', req.url)
const query = url.parse(req.url, true).query
console.log(query.type)
console.log(query.componentName)
console.log(query.methodName)
console.log(query.value)
if (query.type === TYPE_PROPERTY)
{
UpdateDigitalTwin(dtServiceclient, query.componentName, query.methodName, query.value)
console.log('Updating Property..')
}
else if (query.type === TYPE_COMMAND)
{
SendCommand(dtServiceclient, query.componentName, query.methodName, query.value)
console.log('Updating Command..')
}
res.end()
}).listen(PORT)
}
async function UpdateDigitalTwin (dtServiceclient, componentName, propertyName, propertyValue)
{
let patch
if (componentName === ROOT_COMPONENT)
{
patch = [{
op: 'add',
path: '/' + propertyName,
value: JSON.parse(propertyValue)
}]
}
else
{
patch = [{
op: 'add',
path: '/' + componentName + '/' + propertyName,
value: JSON.parse(propertyValue)
}]
}
console.log(patch)
await dtServiceclient.updateDigitalTwin(constants.deviceId, patch)
console.log('Patch has been successfully applied')
};
async function SendCommand (dtServiceclient, componentName, commandName, commandValue)
{
const options = {
connectTimeoutInSeconds: 0,
responseTimeoutInSeconds: 30 // The responseTimeoutInSeconds must be within [5; 300]
}
let commandResponse
if (componentName === ROOT_COMPONENT)
{
commandResponse = await dtServiceclient.invokeCommand(constants.deviceId, commandName, JSON.parse(commandValue), options)
}
else
{
commandResponse = await dtServiceclient.invokeComponentCommand(
constants.deviceId, componentName, commandName, JSON.parse(commandValue), options)
}
// Print result of the command
console.log(inspect(commandResponse))
};
module.exports = { propertiesCommandsAPI: propertiesCommandsAPI }

3
constants.js → app/constants.js
Просмотреть файл

@ -1,6 +1,7 @@
// Copyright (c) Microsoft Corporation.
// Licensed under the MIT License.
module.exports = {
connectionString: '',
deviceId: ''
};
}

46
app/eventProcessor.js Normal file
Просмотреть файл

@ -0,0 +1,46 @@
// Copyright (c) Microsoft Corporation.
// Licensed under the MIT License.
const influxwriter = require('./influxWriter')
const { checkVerifiedTelemetrySupport, getVerifiedTelemetryStatus } = require('./verifiedTelemetryProcessor')
const constants = require('./constants')
const printError = function (err)
{
console.log(err.message)
}
const processMessage = function (message)
{
// console.log(message);
const body = message.body
const additionalProperties = message.applicationProperties
// console.log(additionalProperties);
const deviceId = message.annotations['iothub-connection-device-id']
let componentName = ''
try
{
componentName = message.annotations['dt-subject']
}
catch (e)
{
componentName = 'Default Component'
}
console.log('Received Telemetry')
if (deviceId === constants.deviceId)
{
console.log('Received Telemetry for device:', constants.deviceId)
for (const key of Object.keys(body))
{
influxwriter.writeTelemetryToInfluxDB(
key,
body[key],
deviceId,
componentName,
checkVerifiedTelemetrySupport(key, additionalProperties),
getVerifiedTelemetryStatus(key, additionalProperties))
}
}
}
module.exports = { printError: printError, processMessage: processMessage }

35
app/index.js Normal file
Просмотреть файл

@ -0,0 +1,35 @@
// Copyright (c) Microsoft Corporation.
// Licensed under the MIT License.
const { EventHubClient, EventPosition } = require('@azure/event-hubs')
const IoTHubTokenCredentials = require('azure-iothub').IoTHubTokenCredentials
const DigitalTwinServiceClient = require('azure-iothub').DigitalTwinClient
const iothubreader = require('./eventProcessor')
const { propertiesCommandsAPI } = require('./HTTPServer')
const { processVerifiedTelemetryProperties } = require('./verifiedTelemetryProcessor')
const constants = require('./constants')
const credentials = new IoTHubTokenCredentials(constants.connectionString)
const dtServiceclient = new DigitalTwinServiceClient(credentials)
let ehClient
EventHubClient.createFromIotHubConnectionString(constants.connectionString).then(function (client)
{
console.log('Successfully created the EventHub Client from iothub connection string.')
ehClient = client
return ehClient.getPartitionIds()
}).then(function (ids)
{
console.log('The partition ids are: ', ids)
return ids.map(function (id)
{
return ehClient.receive(id, iothubreader.processMessage, iothubreader.printError,
{ eventPosition: EventPosition.fromEnqueuedTime(Date.now()) }
)
})
}).catch(iothubreader.printError)
setInterval(processVerifiedTelemetryProperties, 10000, dtServiceclient)
propertiesCommandsAPI(dtServiceclient)

127
app/influxWriter.js Normal file
Просмотреть файл

@ -0,0 +1,127 @@
// Copyright (c) Microsoft Corporation.
// Licensed under the MIT License.
const Influx = require('influx')
// So this is some generic influxDB schema for IoT Data.
const influx = new Influx.InfluxDB({
host: 'influxdb',
database: 'vt',
schema: [
{
measurement: 'telemetry_messages',
fields: {
jsonvalue: Influx.FieldType.STRING,
jsonasnumber: Influx.FieldType.FLOAT
},
tags: [
'telemetry',
'deviceId',
'componentName',
'verifiedTelemetrySupport',
'verifiedTelemetryStatus'
]
},
{
measurement: 'property_messages',
fields: {
jsonvalue: Influx.FieldType.STRING,
jsonasnumber: Influx.FieldType.FLOAT
},
tags: [
'property',
'deviceId',
'componentName'
]
}
]
})
/**
* Writes a generic json key/value pair to InfluxDB...
* @param {string} key key of the json pair
* @param {string} value value of the json pair
* @param {string} deviceId iothub deviceId
* @param {string} componentName component name
* @param {string} verifiedTelemetrySupport verifiedTelemetrySupport
* @param {string} verifiedTelemetryStatus verifiedTelemetryStatus
*/
const writeTelemetryToInfluxDB = function (key, value, deviceId, componentName, verifiedTelemetrySupport, verifiedTelemetryStatus)
{
let parsedNumber = 0
try
{
parsedNumber = parseFloat(value)
influx.writePoints([
{
measurement: 'telemetry_messages',
fields: { jsonvalue: value, jsonasnumber: parsedNumber },
tags: {
telemetry: key,
deviceId: deviceId,
componentName: componentName,
verifiedTelemetrySupport: verifiedTelemetrySupport,
verifiedTelemetryStatus: verifiedTelemetryStatus
}
}]
)
// console.log('Telemetry with key: ',
// key, ', value: ', parsedNumber, 'and vTStatus: ', verifiedTelemetryStatus, 'stored in DB');
}
catch (e)
{
// couldn't parse, so send string only
influx.writePoints([
{
measurement: 'telemetry_messages',
fields: { jsonvalue: value },
tags: {
telemetry: key,
deviceId: deviceId,
componentName: componentName,
verifiedTelemetrySupport: verifiedTelemetrySupport,
verifiedTelemetryStatus: verifiedTelemetryStatus
}
}
])
// console.log('PARSING ERROR!, ','Telemetry with key: ',
// key, ', string Value: ', value, 'and vTStatus: ', verifiedTelemetryStatus, 'stored in DB');
}
}
const writePropertyToInfluxDB = function (key, value, deviceId, componentName, timestampString)
{
try
{
const parsedDatetime = Date.parse(timestampString)
const adjustedstartTime = parsedDatetime
influx.writePoints([
{
measurement: 'property_messages',
fields: { jsonvalue: value },
tags: { property: key, deviceId: deviceId, componentName: componentName },
timestamp: adjustedstartTime
}],
{
precision: 'ms'
}
)
console.log('Property with key: ', key, 'and value: ', value, 'stored in DB')
}
catch (e)
{
// couldnt parse, so send string only
influx.writePoints([
{
measurement: 'property_messages',
fields: { jsonvalue: value },
tags: { property: key, deviceId: deviceId, componentName: componentName }
}
])
console.log('PARSING ERROR!, ', 'Property with key: ', key, 'and string Value: ', value, 'stored in DB')
}
}
module.exports = { writeTelemetryToInfluxDB: writeTelemetryToInfluxDB, writePropertyToInfluxDB: writePropertyToInfluxDB }

10
package.json → app/package.json
Просмотреть файл

@ -4,7 +4,8 @@
"description": "test",
"main": "index.js",
"scripts": {
"lint": "eslint -c ../../../../.eslintrc.json .",
"lint": "eslint **/*.js",
"format": "eslint **/*.js --fix",
"npmlockrefresh": "npm i --package-lock-only",
"ci": "npm -s run lint",
"test": "echo \"Error: no test specified\" && exit 1"
@ -18,7 +19,10 @@
"influx": "^5.8.0"
},
"devDependencies": {
"eslint": "^6.8.0",
"eslint-config-google": "^0.13.0"
"eslint": "^7.25.0",
"eslint-config-standard": "^16.0.2",
"eslint-plugin-import": "^2.22.1",
"eslint-plugin-node": "^11.1.0",
"eslint-plugin-promise": "^4.3.1"
}
}

102
app/verifiedTelemetryProcessor.js Normal file
Просмотреть файл

@ -0,0 +1,102 @@
// Copyright (c) Microsoft Corporation.
// Licensed under the MIT License.
const influxwriter = require('./influxWriter')
const constants = require('./constants')
let digitalTwinLocalCopy
const checkVerifiedTelemetrySupport = function (telemetryName, additionalProperties)
{
const verifiedTelemetryComponentName = 'vT' + telemetryName
if (digitalTwinLocalCopy.hasOwnProperty(verifiedTelemetryComponentName) &&
digitalTwinLocalCopy.hasOwnProperty('vTDevice') &&
digitalTwinLocalCopy.vTDevice.hasOwnProperty('enableVerifiedTelemetry'))
{
console.log('Verified Telemetry: Entering New loop 2')
if (digitalTwinLocalCopy[verifiedTelemetryComponentName].hasOwnProperty('fingerprintTemplate') &&
digitalTwinLocalCopy.vTDevice.enableVerifiedTelemetry === true)
{
console.log('Verified Telemetry: Reference Fingerprint not collected')
return (true)
}
else
{
console.log('Verified Telemetry: Reference Fingerprint collected')
return (false)
}
}
else
{
return (false)
}
}
const getVerifiedTelemetryStatus = function (telemetryName, additionalProperties)
{
const verifiedTelemetryComponentName = 'vT' + telemetryName
if (additionalProperties.hasOwnProperty(verifiedTelemetryComponentName))
{
console.log('Verified Telemetry Status fetched from Enriched Telemetry Message')
return (additionalProperties[verifiedTelemetryComponentName])
}
else if (digitalTwinLocalCopy.hasOwnProperty(verifiedTelemetryComponentName))
{
console.log('Verified Telemetry Status fetched from Digital Twin')
return (digitalTwinLocalCopy[verifiedTelemetryComponentName].telemetryStatus)
}
else
{
return (false)
}
}
async function processVerifiedTelemetryProperties (dtServiceclient)
{
digitalTwinLocalCopy = await dtServiceclient.getDigitalTwin(constants.deviceId)
if (digitalTwinLocalCopy.hasOwnProperty('vTDevice') &&
digitalTwinLocalCopy.vTDevice.hasOwnProperty('enableVerifiedTelemetry'))
{
if (digitalTwinLocalCopy.hasOwnProperty('vTsoilMoistureExternal1') &&
digitalTwinLocalCopy.hasOwnProperty('vTsoilMoistureExternal2'))
{
if (digitalTwinLocalCopy.vTsoilMoistureExternal1.hasOwnProperty('fingerprintTemplate') &&
digitalTwinLocalCopy.vTsoilMoistureExternal2.hasOwnProperty('fingerprintTemplate') &&
digitalTwinLocalCopy.vTDevice.enableVerifiedTelemetry === true)
{
influxwriter.writePropertyToInfluxDB(
'deviceStatus',
digitalTwinLocalCopy.vTDevice.deviceStatus,
constants.deviceId,
'vTDevice',
digitalTwinLocalCopy.vTDevice.$metadata.deviceStatus.lastUpdateTime)
}
else
{
influxwriter.writePropertyToInfluxDB(
'deviceStatus',
'unknown',
constants.deviceId,
'vTDevice',
digitalTwinLocalCopy.vTDevice.$metadata.deviceStatus.lastUpdateTime)
}
}
}
if (digitalTwinLocalCopy.hasOwnProperty('vTDevice'))
{
influxwriter.writePropertyToInfluxDB(
'enableVerifiedTelemetry',
digitalTwinLocalCopy.vTDevice.enableVerifiedTelemetry,
constants.deviceId,
'vTDevice',
digitalTwinLocalCopy.vTDevice.$metadata.enableVerifiedTelemetry.lastUpdateTime)
}
};
module.exports =
{
checkVerifiedTelemetrySupport: checkVerifiedTelemetrySupport,
getVerifiedTelemetryStatus: getVerifiedTelemetryStatus,
processVerifiedTelemetryProperties: processVerifiedTelemetryProperties
}

29
docker-compose.yml
Просмотреть файл

@ -1,32 +1,30 @@
version: "2"
services:
web:
build:
context: .
build: ./app
container_name: app
depends_on:
- grafana
- influxdb
ports:
- 8080:8080
# networks:
# - monitoring
links:
- influxdb
grafana:
image: grafana/grafana
environment:
GF_SERVER_HTTP_PORT: 3030
GF_INSTALL_PLUGINS: https://github.com/cloudspout/cloudspout-button-panel/releases/download/7.0.2/cloudspout-button-panel.zip;cloudspout-button-panel
container_name: grafana
restart: always
ports:
- 3030:3030
# networks:
# - monitoring
volumes:
- ./grafana:/etc/grafana/provisioning/
environment:
GF_SERVER_HTTP_PORT: 3030
GF_INSTALL_PLUGINS: https://github.com/cloudspout/cloudspout-button-panel/releases/download/7.0.2/cloudspout-button-panel.zip;cloudspout-button-panel
- ./grafana/provisioning:/etc/grafana/provisioning/
- ./grafana/grafana.ini:/etc/grafana/grafana.ini
influxdb:
image: influxdb:1.8
environment:
INFLUXDB_DB: vt
INFLUXDB_ADMIN_ENABLED: "true"
@ -34,18 +32,13 @@ services:
INFLUXDB_ADMIN_PASSWORD: supersecretpassword
INFLUXDB_USER: telegraf
INFLUXDB_USER_PASSWORD: secretpassword
image: influxdb:1.8
container_name: influxdb
restart: always
ports:
- 8086:8086
# networks:
# - monitoring
volumes:
- influxdb-volume:/var/lib/influxdb
# networks:
# monitoring:
# external: false
volumes:
influxdb-volume:
external: false

35
eventProcessor.js
Просмотреть файл

@ -1,35 +0,0 @@
// Copyright (c) Microsoft Corporation.
// Licensed under the MIT License.
let influxwriter = require('./influxWriter');
const { checkVerifiedTelemetrySupport, getVerifiedTelemetryStatus } = require('./verifiedTelemetryProcessor');
const constants = require('./constants');
let printError = function (err) {
console.log(err.message);
};
let processMessage = function (message) {
// console.log(message);
let body = message.body;
let additionalProperties = message.applicationProperties;
// console.log(additionalProperties);
let deviceId = message.annotations["iothub-connection-device-id"];
let componentName = ""
try {
componentName = message.annotations["dt-subject"];
} catch (e) {
componentName = "Default Component";
}
console.log("Received Telemetry");
if(deviceId == constants.deviceId)
{
console.log("Received Telemetry for device:",constants.deviceId);
for (const key of Object.keys(body))
{
influxwriter.writeTelemetryToInfluxDB(key, body[key], deviceId,componentName, checkVerifiedTelemetrySupport(key, additionalProperties), getVerifiedTelemetryStatus(key, additionalProperties));
}
}
};
module.exports = { printError: printError, processMessage: processMessage}

2
grafana/grafana.ini Normal file
Просмотреть файл

@ -0,0 +1,2 @@
[dashboards]
min_refresh_interval = 1s

27
grafana/dashboards/VerifiedTelemetryGSGDashboard.json → grafana/provisioning/dashboards/VerifiedTelemetryGSGDashboard.json
Просмотреть файл

@ -15,7 +15,7 @@
"editable": true,
"gnetId": null,
"graphTooltip": 0,
"iteration": 1618868816684,
"iteration": 1620410228816,
"links": [],
"panels": [
{
@ -125,7 +125,7 @@
"text": {},
"textMode": "auto"
},
"pluginVersion": "7.5.4",
"pluginVersion": "7.5.5",
"targets": [
{
"groupBy": [],
@ -321,7 +321,7 @@
"text": {},
"textMode": "auto"
},
"pluginVersion": "7.5.4",
"pluginVersion": "7.5.5",
"targets": [
{
"groupBy": [],
@ -490,7 +490,7 @@
"alertThreshold": true
},
"percentage": false,
"pluginVersion": "7.5.4",
"pluginVersion": "7.5.5",
"pointradius": 8,
"points": true,
"renderer": "flot",
@ -791,7 +791,7 @@
"alertThreshold": true
},
"percentage": false,
"pluginVersion": "7.5.4",
"pluginVersion": "7.5.5",
"pointradius": 8,
"points": true,
"renderer": "flot",
@ -1276,7 +1276,7 @@
"alertThreshold": true
},
"percentage": false,
"pluginVersion": "7.5.4",
"pluginVersion": "7.5.5",
"pointradius": 8,
"points": true,
"renderer": "flot",
@ -1433,7 +1433,7 @@
"alertThreshold": true
},
"percentage": false,
"pluginVersion": "7.5.4",
"pluginVersion": "7.5.5",
"pointradius": 8,
"points": true,
"renderer": "flot",
@ -1590,7 +1590,7 @@
"alertThreshold": true
},
"percentage": false,
"pluginVersion": "7.5.4",
"pluginVersion": "7.5.5",
"pointradius": 8,
"points": true,
"renderer": "flot",
@ -1747,7 +1747,7 @@
"alertThreshold": true
},
"percentage": false,
"pluginVersion": "7.5.4",
"pluginVersion": "7.5.5",
"pointradius": 8,
"points": true,
"renderer": "flot",
@ -1869,7 +1869,7 @@
}
}
],
"refresh": "5s",
"refresh": "1s",
"schemaVersion": 27,
"style": "dark",
"tags": [],
@ -1879,8 +1879,8 @@
"allValue": null,
"current": {
"selected": false,
"text": "MyMXChipDevice",
"value": "MyMXChipDevice"
"text": "stm",
"value": "stm"
},
"datasource": "InfluxDB",
"definition": "",
@ -1911,6 +1911,7 @@
},
"timepicker": {
"refresh_intervals": [
"1s",
"5s",
"10s",
"30s",
@ -1925,6 +1926,6 @@
},
"timezone": "",
"title": "Verified Telemetry",
"uid": "BhqdsjvMZ",
"uid": "AzureIoT",
"version": 1
}

0
grafana/dashboards/dashboard.yml → grafana/provisioning/dashboards/dashboard.yml
Просмотреть файл

0
grafana/datasources/datasource.yml → grafana/provisioning/datasources/datasource.yml
Просмотреть файл

27
index.js
Просмотреть файл

@ -1,27 +0,0 @@
// Copyright (c) Microsoft Corporation.
// Licensed under the MIT License.
var { EventHubClient, EventPosition } = require('@azure/event-hubs');
let IoTHubTokenCredentials = require('azure-iothub').IoTHubTokenCredentials;
let DigitalTwinServiceClient = require('azure-iothub').DigitalTwinClient;
let iothubreader = require('./eventProcessor');
const { propertiesCommandsAPI } = require('./HTTPServer');
const {processVerifiedTelemetryProperties } = require('./verifiedTelemetryProcessor');
var constants = require('./constants');
const credentials = new IoTHubTokenCredentials(constants.connectionString);
const dtServiceclient = new DigitalTwinServiceClient(credentials);
EventHubClient.createFromIotHubConnectionString(constants.connectionString).then(function (client) {
console.log("Successully created the EventHub Client from iothub connection string.");
ehClient = client;
return ehClient.getPartitionIds();
}).then(function (ids) {
console.log("The partition ids are: ", ids);
return ids.map(function (id) {
return ehClient.receive(id, iothubreader.processMessage, iothubreader.printError, { eventPosition: EventPosition.fromEnqueuedTime(Date.now()) });
});
}).catch(iothubreader.printError);
setInterval(processVerifiedTelemetryProperties, 10000, dtServiceclient);
propertiesCommandsAPI(dtServiceclient);

105
influxWriter.js
Просмотреть файл

@ -1,105 +0,0 @@
// Copyright (c) Microsoft Corporation.
// Licensed under the MIT License.
const Influx = require('influx');
//So this is some generic influxDB schema for IoT Data.
const influx = new Influx.InfluxDB({
host: 'influxdb',
database: 'vt',
schema: [
{
measurement: 'telemetry_messages',
fields: {
jsonvalue: Influx.FieldType.STRING,
jsonasnumber: Influx.FieldType.FLOAT
},
tags: [
'telemetry',
'deviceId',
'componentName',
'verifiedTelemetrySupport',
'verifiedTelemetryStatus'
]
},
{
measurement: 'property_messages',
fields: {
jsonvalue: Influx.FieldType.STRING,
jsonasnumber: Influx.FieldType.FLOAT
},
tags: [
'property',
'deviceId',
'componentName'
]
}
]
});
/**
* Writes a generic json key/value pair to InfluxDB...
* @param {string} key key of the json pair
* @param {string} value value of the json pair
* @param {string} deviceId iothub deviceId
* @param {string} componentName component name
* @param {string} verifiedTelemetrySupport verifiedTelemetrySupport
* @param {string} verifiedTelemetryStatus verifiedTelemetryStatus
*/
let writeTelemetryToInfluxDB = function (key, value, deviceId, componentName, verifiedTelemetrySupport, verifiedTelemetryStatus) {
let parsedNumber = 0;
try {
parsedNumber = parseFloat(value);
influx.writePoints([
{
measurement: 'telemetry_messages',
fields: { jsonvalue: value, jsonasnumber: parsedNumber },
tags: { telemetry: key, deviceId: deviceId, componentName: componentName, verifiedTelemetrySupport: verifiedTelemetrySupport, verifiedTelemetryStatus: verifiedTelemetryStatus},
}]
)
// console.log('Telemetry with key: ', key, ', value: ', parsedNumber, 'and vTStatus: ', verifiedTelemetryStatus, 'stored in DB');
} catch (e){
//couldnt parse, so send string only
influx.writePoints([
{
measurement: 'telemetry_messages',
fields: { jsonvalue: value },
tags: { telemetry: key, deviceId: deviceId, componentName: componentName, verifiedTelemetrySupport: verifiedTelemetrySupport, verifiedTelemetryStatus: verifiedTelemetryStatus}
}
])
// console.log('PARSING ERROR!, ','Telemetry with key: ', key, ', string Value: ', value, 'and vTStatus: ', verifiedTelemetryStatus, 'stored in DB');
}
};
let writePropertyToInfluxDB = function (key, value, deviceId, componentName, timestampString) {
let parsedNumber = 0;
try {
let parsedDatetime = Date.parse(timestampString);
var adjustedstartTime = parsedDatetime;
influx.writePoints([
{
measurement: 'property_messages',
fields: { jsonvalue: value },
tags: { property: key, deviceId: deviceId, componentName: componentName },
timestamp: adjustedstartTime
}],
{
precision: 'ms'
}
)
console.log('Property with key: ', key, 'and value: ', value, 'stored in DB');
} catch (e){
//couldnt parse, so send string only
influx.writePoints([
{
measurement: 'property_messages',
fields: { jsonvalue: value },
tags: { property: key, deviceId: deviceId, componentName: componentName}
}
])
console.log('PARSING ERROR!, ','Property with key: ', key, 'and string Value: ', value, 'stored in DB');
}
};
module.exports = { writeTelemetryToInfluxDB: writeTelemetryToInfluxDB, writePropertyToInfluxDB: writePropertyToInfluxDB}

76
verifiedTelemetryProcessor.js
Просмотреть файл

@ -1,76 +0,0 @@
// Copyright (c) Microsoft Corporation.
// Licensed under the MIT License.
let influxwriter = require('./influxWriter');
var constants = require('./constants');
var digitalTwinLocalCopy;
let checkVerifiedTelemetrySupport = function (telemetryName, additionalProperties) {
var verifiedTelemetryComponentName = 'vT' + telemetryName;
if(digitalTwinLocalCopy.hasOwnProperty(verifiedTelemetryComponentName) && digitalTwinLocalCopy.hasOwnProperty('vTDevice') && digitalTwinLocalCopy.vTDevice.hasOwnProperty('enableVerifiedTelemetry'))
{
console.log("Verified Telemetry: Entering New loop 2")
if(digitalTwinLocalCopy[verifiedTelemetryComponentName].hasOwnProperty('fingerprintTemplate') && digitalTwinLocalCopy.vTDevice.enableVerifiedTelemetry == true)
{
console.log("Verified Telemetry: Reference Fingerprint not collected");
return(true);
}
else
{
console.log("Verified Telemetry: Reference Fingerprint collected");
return (false);
}
}
else
{
return(false);
}
};
let getVerifiedTelemetryStatus = function (telemetryName, additionalProperties) {
var verifiedTelemetryComponentName = 'vT' + telemetryName;
if(additionalProperties.hasOwnProperty(verifiedTelemetryComponentName))
{
console.log("Verified Telemetry Status fetched from Enriched Telemetry Message");
return(additionalProperties[verifiedTelemetryComponentName])
}
else if(digitalTwinLocalCopy.hasOwnProperty(verifiedTelemetryComponentName))
{
console.log("Verified Telemetry Status fetched from Digital Twin");
return(digitalTwinLocalCopy[verifiedTelemetryComponentName].telemetryStatus);
}
else
{
return(false);
}
};
async function processVerifiedTelemetryProperties(dtServiceclient) {
digitalTwinLocalCopy = await dtServiceclient.getDigitalTwin(constants.deviceId);
if(digitalTwinLocalCopy.hasOwnProperty('vTDevice') && digitalTwinLocalCopy.vTDevice.hasOwnProperty('enableVerifiedTelemetry'))
{
if(digitalTwinLocalCopy.hasOwnProperty('vTsoilMoistureExternal1') && digitalTwinLocalCopy.hasOwnProperty('vTsoilMoistureExternal2'))
{
if(digitalTwinLocalCopy.vTsoilMoistureExternal1.hasOwnProperty('fingerprintTemplate') && digitalTwinLocalCopy.vTsoilMoistureExternal2.hasOwnProperty('fingerprintTemplate') && digitalTwinLocalCopy.vTDevice.enableVerifiedTelemetry == true)
{
influxwriter.writePropertyToInfluxDB('deviceStatus', digitalTwinLocalCopy.vTDevice.deviceStatus, constants.deviceId, 'vTDevice', digitalTwinLocalCopy.vTDevice.$metadata.deviceStatus.lastUpdateTime);
}
else
{
influxwriter.writePropertyToInfluxDB('deviceStatus', 'unknown', constants.deviceId, 'vTDevice', digitalTwinLocalCopy.vTDevice.$metadata.deviceStatus.lastUpdateTime);
}
}
}
if(digitalTwinLocalCopy.hasOwnProperty('vTDevice'))
{
influxwriter.writePropertyToInfluxDB('enableVerifiedTelemetry', digitalTwinLocalCopy.vTDevice.enableVerifiedTelemetry, constants.deviceId, 'vTDevice', digitalTwinLocalCopy.vTDevice.$metadata.enableVerifiedTelemetry.lastUpdateTime);
}
};
module.exports = { checkVerifiedTelemetrySupport: checkVerifiedTelemetrySupport, getVerifiedTelemetryStatus: getVerifiedTelemetryStatus, processVerifiedTelemetryProperties: processVerifiedTelemetryProperties}