[feat] Add cloud functions to ingest cloud iot data

Author: alvarowolfx

.DS_Store

@@ -0,0 +1,5 @@
+{
+  "projects": {
+    "production": "gcloud-iot-edge"
+  }
+}

.firebaserc

@@ -1,2 +1,3 @@
 *.pem
-.DS_Store
+.DS_Store
+node_modules

.gitignore

@@ -0,0 +1,6 @@
+{
+  "rules": {
+    ".read": "auth != null",
+    ".write": "auth != null"
+  }
+}

database.rules.json

@@ -55,13 +55,21 @@ class CloudIoTCoreGateway {
 
     // Create a client, and connect to the Google MQTT bridge
     this.iatTime = parseInt( Date.now() / 1000 )    
-    this.client = mqtt.connect( connectionArgs )    
+    this.client = mqtt.connect( connectionArgs )
+    this.client.on( 'connect', ( success ) => {
+      if ( success ) {
+        logger.info( 'Client connected.' )
+  
+      } else {
+        logger.info( 'Client not connected.' )
+      }
+    } )
   }
 
-  checkConnection() {
+  checkConnection() {    
     const secsFromIssue = parseInt( Date.now() / 1000 ) - this.iatTime
     if ( secsFromIssue > this.tokenExpMins * 60 ) {      
-      logger.info( `\tRefreshing token after ${secsFromIssue} seconds.` )
+      logger.info( `Refreshing token after ${secsFromIssue} seconds.` )
       this.connect()
     }
   }

edge-server/CloudIoTCoreGateway.js

@@ -17,7 +17,7 @@ class EdgeServer {
     this.classifier = new ImageClassifier( classifier )
     this.gateway = new CloudIoTCoreGateway( gateway )
     this.web = new WebInterface( web )      
-    this.limiter = new RateLimiter( 2, 'minute' )
+    this.limiter = new RateLimiter( 1, 'minute' )
 
     this.deviceQueue = {}
   }
@@ -27,16 +27,16 @@ class EdgeServer {
     await this.classifier.load()
     this.deviceListener.start()
     this.web.start()
-    this.gateway.start()
+    await this.gateway.start()
 
-    this.deviceListener.onDeviceAdded( ( deviceId ) => {      
-      this.gateway.attachDevice( deviceId )
+    this.deviceListener.onDeviceAdded( async ( deviceId ) => {      
+      await this.gateway.attachDevice( deviceId )
       this.gateway.publishDeviceState( deviceId, { status : 'online' } )
     } )
 
-    this.deviceListener.onDeviceRemoved( ( deviceId ) => {      
-      this.gateway.detachDevice( deviceId )
+    this.deviceListener.onDeviceRemoved( async ( deviceId ) => {      
       this.gateway.publishDeviceState( deviceId, { status : 'offline' } )
+      this.gateway.detachDevice( deviceId )
     } )
 
     this.run()    
@@ -49,11 +49,11 @@ class EdgeServer {
     return Object.keys( this.deviceQueue ).length > 0 
   }
 
-  queueData( device, { classes, trackedClasses, countClasses } ) { 
-    if ( classes.length === 0 ) {
-      return
+  queueData( device, { classes, trackedClasses, countClasses } ) {     
+    if ( classes.length === 0 ) {     
+      classes = [ 'empty' ] 
+      countClasses = { empty : 1 }      
     }       
-
     const { name } = device
     const deviceData = this.deviceQueue[name]
     if ( !deviceData ) {
@@ -134,12 +134,18 @@ class EdgeServer {
           logger.info( '[PublishData] Sending data to cloud iot core.' )
           const publishPromises = Object.keys( this.deviceQueue ).map( ( deviceId ) => {
             const res = this.deviceQueue[deviceId]
-            return this.gateway.publishDeviceTelemetry( deviceId, res.countClasses )
-          } )    
-          logger.info( `Promises ${publishPromises}` )
-          this.clearQueue()                    
+            // Check if is just empty
+            if ( res.countClasses.empty && Object.keys( res.countClasses ).length === 1 ) {
+              logger.info( '[Empty] Message ' )
+              return this.gateway.publishDeviceTelemetry( deviceId, { classes : {} } )
+            } 
+            delete res.countClasses.empty
+            return this.gateway.publishDeviceTelemetry( deviceId, { classes : res.countClasses } )
+          } )              
           await Promise.all( publishPromises )
+          this.clearQueue()                    
         } else {
+          
           logger.info( '[PublishData] Publishing throttled.' )
         }
       }
@@ -166,14 +172,15 @@ class EdgeServer {
 
     logger.info( 'Sending offline events' )
     try {
+      logger.info( 'Sending gateway offline event' )
+      await this.gateway.publishGatewayState( { status : 'offline' } )    
       const publishPromises = devices.map( ( device ) => {
         logger.info( `Sending offline event for device ${device.name}` )
         return this.gateway.publishDeviceState( device.name, { status : 'offline' } )
       } )      
       await Promise.all( publishPromises )
-      logger.info( 'Sending gateway offline event' )
-      await this.gateway.publishGatewayState( { status : 'offline' } )    
       logger.info( 'All offline events sent' )
+      await new Promise( resolve => setTimeout( resolve, 1000 ) )
     } catch ( err ) {
       logger.error( `Error sending data to cloud iot core ${err}`, err )
     }

edge-server/EdgeServer.js

@@ -7,6 +7,7 @@
     "sdkconfig.h": "c",
     "qr_recoginize.h": "c",
     "ov2640.h": "c",
-    "sensor.h": "c"
+    "sensor.h": "c",
+    "esp_wifi.h": "c"
   }
 }

esp32-camera-firmware/.vscode/settings.json

@@ -0,0 +1,24 @@
+{
+  "database": {
+    "rules": "database.rules.json"
+  },
+  "functions": {
+    "predeploy": [
+      "npm --prefix \"$RESOURCE_DIR\" run lint"
+    ]
+  },
+  "hosting": {
+    "public": "public",
+    "ignore": [
+      "firebase.json",
+      "**/.*",
+      "**/node_modules/**"
+    ],
+    "rewrites": [
+      {
+        "source": "**",
+        "destination": "/index.html"
+      }
+    ]
+  }
+}

firebase.json

@@ -0,0 +1,123 @@
+{
+  "parserOptions": {
+    // Required for certain syntax usages
+    "ecmaVersion": 6
+  },
+  "plugins": [
+    "promise"
+  ],
+  "extends": "eslint:recommended",
+  "rules": {
+    // Removed rule "disallow the use of console" from recommended eslint rules
+    "no-console": "off",
+
+    // Removed rule "disallow multiple spaces in regular expressions" from recommended eslint rules
+    "no-regex-spaces": "off",
+
+    // Removed rule "disallow the use of debugger" from recommended eslint rules
+    "no-debugger": "off",
+
+    // Removed rule "disallow unused variables" from recommended eslint rules
+    "no-unused-vars": "off",
+
+    // Removed rule "disallow mixed spaces and tabs for indentation" from recommended eslint rules
+    "no-mixed-spaces-and-tabs": "off",
+
+    // Removed rule "disallow the use of undeclared variables unless mentioned in /*global */ comments" from recommended eslint rules
+    "no-undef": "off",
+
+    // Warn against template literal placeholder syntax in regular strings
+    "no-template-curly-in-string": 1,
+
+    // Warn if return statements do not either always or never specify values
+    "consistent-return": 1,
+
+    // Warn if no return statements in callbacks of array methods
+    "array-callback-return": 1,
+
+    // Require the use of === and !==
+    "eqeqeq": 2,
+
+    // Disallow the use of alert, confirm, and prompt
+    "no-alert": 2,
+
+    // Disallow the use of arguments.caller or arguments.callee
+    "no-caller": 2,
+
+    // Disallow null comparisons without type-checking operators
+    "no-eq-null": 2,
+
+    // Disallow the use of eval()
+    "no-eval": 2,
+
+    // Warn against extending native types
+    "no-extend-native": 1,
+
+    // Warn against unnecessary calls to .bind()
+    "no-extra-bind": 1,
+
+    // Warn against unnecessary labels    
+    "no-extra-label": 1,
+
+    // Disallow leading or trailing decimal points in numeric literals
+    "no-floating-decimal": 2,
+
+    // Warn against shorthand type conversions
+    "no-implicit-coercion": 1,
+
+    // Warn against function declarations and expressions inside loop statements
+    "no-loop-func": 1,
+
+    // Disallow new operators with the Function object
+    "no-new-func": 2,
+
+    // Warn against new operators with the String, Number, and Boolean objects
+    "no-new-wrappers": 1,
+
+    // Disallow throwing literals as exceptions
+    "no-throw-literal": 2,
+
+    // Require using Error objects as Promise rejection reasons
+    "prefer-promise-reject-errors": 2,
+
+    // Enforce “for” loop update clause moving the counter in the right direction
+    "for-direction": 2,
+
+    // Enforce return statements in getters
+    "getter-return": 2,
+
+    // Disallow await inside of loops
+    "no-await-in-loop": 2,
+
+    // Disallow comparing against -0
+    "no-compare-neg-zero": 2,
+
+    // Warn against catch clause parameters from shadowing variables in the outer scope
+    "no-catch-shadow": 1,
+
+    // Disallow identifiers from shadowing restricted names
+    "no-shadow-restricted-names": 2,
+
+    // Enforce return statements in callbacks of array methods
+    "callback-return": 2,
+
+    // Require error handling in callbacks
+    "handle-callback-err": 2,
+
+    // Warn against string concatenation with __dirname and __filename
+    "no-path-concat": 1,
+
+    // Prefer using arrow functions for callbacks
+    "prefer-arrow-callback": 1,
+
+    // Return inside each then() to create readable and reusable Promise chains.
+    // Forces developers to return console logs and http calls in promises. 
+    "promise/always-return": 2,
+
+    //Enforces the use of catch() on un-returned promises
+    "promise/catch-or-return": 2,
+
+    // Warn against nested then() or catch() statements
+    "promise/no-nesting": 1
+  }
+}

functions/.eslintrc.json

@@ -0,0 +1,31 @@
+const functions = require('firebase-functions')
+const admin = require('firebase-admin')
+
+admin.initializeApp()
+
+const db = admin.database()
+
+exports.processTelemetry = functions
+  .pubsub
+  .topic('telemetry')
+  .onPublish( handleMessage('data') )
+
+exports.processState = functions
+  .pubsub
+  .topic('state')
+  .onPublish( handleMessage('meta') )
+
+function handleMessage( topic ) {
+  return ( message, context ) => {
+    const attributes = message.attributes
+    const payload = message.json
+
+    const deviceId = attributes['deviceId']
+
+    const data = Object.assign({}, payload, {          
+      updated: context.timestamp
+    })
+
+    return db.ref(`/devices/${deviceId}/${topic}`).update(data)
+  }
+}