update sensorID and deviceID in API call, override invaled IDs

client.py

@@ -1,5 +1,7 @@
 #!/usr/bin/python
 
+# example client application
+
 import logging
 import asyncio
 import numpy as np
@@ -40,9 +42,9 @@ def get_frame():
 async def test_coro():
     C = MQTTClient()
     print("START" + str(get_uptime()))
-    await C.connect('mqtt://100.79.112.65:1883')
+    await C.connect('mqtt://127.0.0.1:1883')
     tasks = [
-        asyncio.ensure_future(C.publish('sensors/pi1/camera', encoder({"data": str(get_frame()), "sensorID": 0, "timestamp": time.time_ns()}), qos=QOS_2)),
+        asyncio.ensure_future(C.publish('sensors/pi1/camera', encoder({"data": str(get_frame()), "sensorID": 0, "timestamp": time.time_ns() }), qos=QOS_2)),
         asyncio.ensure_future(C.publish('sensors/pi1/uptime', encoder({"data": get_uptime(), "sensorID": 0, "timestamp": time.time_ns()}), qos=QOS_2)),
     ]
     await asyncio.wait(tasks)

main.py

@@ -17,9 +17,12 @@ import cv2 as cv
 
 
 sublist = []
+sensorlist = []
+devicelist = []
+sensorcount = 0
+devicecount = 0
 
-sensortypes = ["microphone", "camera", "humidity"]
+showData = True # debug print
-showData = True
 
 async def broker_coro(brokerconfig):
     broker = Broker(config=brokerconfig)
@@ -27,62 +30,75 @@ async def broker_coro(brokerconfig):
 
 async def uptime_coro():
     C = MQTTClient()
-    await C.connect('mqtt://127.0.0.1:1883/')
+    await C.connect('mqtt://127.0.0.1:1883/') # connect to my own broker
-    # Subscribe to '$SYS/broker/uptime' with QOS=1
+    await C.subscribe(sublist) # subscribe to list of sensor topics
-    # Subscribe to '$SYS/broker/load/#' with QOS=2
-    #await C.subscribe([
-    #        ('a/b', QOS_1),
-    #     ])
-    await C.subscribe(sublist)
     try:
-        count = 0
         while True:
-            count = count + 1
             message = await C.deliver_message()
             packet = message.publish_packet
             dictout = decoder(packet.payload.data)
-            print("%d:  %s => %s" % (count, packet.variable_header.topic_name, dictout))
+            sensorID = 99999 # in case no sensor ID is found somehow
+            if 'sensorID' in dictout.keys() and dictout["sensorID"] > sensorcount:
+                sensorID = dictout["sensorID"]
+            else:
+                for sensor in sensorlist:
+                    if sensor[0] == packet.variable_header.topic_name:
+                        sensorID = sensor[1]
+
+            deviceID = 99999 # in case no device ID is found somehow
+            if 'deviceID' in dictout.keys() and dictout["deviceID"] > devicecount:
+                deviceID = dictout["deviceID"]
+            else:
+                for device in devicelist:
+                    print('/'.join(packet.variable_header.topic_name.split("/")[0:-1]) + "/")
+                    if device[0] == '/'.join(packet.variable_header.topic_name.split("/")[0:-1]) + "/": # remove the last field (sensor)
+                        deviceID = device[1]
+            timestamp = time.time_ns()
+            if 'timestamp' in dictout.keys():
+                timestamp = dictout["timestamp"]
+            #print("%s => %s" % (packet.variable_header.topic_name, dictout))
             if packet.variable_header.topic_name.split("/")[-1] == "camera": # trigger image loading
                 print(len(dictout["data"]))
                 img = np.frombuffer(base64.b64decode(dictout["data"]), np.uint8)
-                #height, width = img.shape[:2]
+                cv.imwrite('./output.jpg', img) # write to file for testing
-                #print(height, width)
+            publishData(str(dictout["data"]), str(sensorID), str(deviceID), str(packet.variable_header.topic_name.split("/")[-1]), timestamp)
-
-                #cv.imshow('frame', img)
-                cv.imwrite('./output.jpg', img)
-                #if cv.waitKey(1) == ord('q'):
-                #    break
-            else:
-                publishData(str(dictout["data"]), str(dictout["sensorID"]), str(packet.variable_header.topic_name.split("/")[-1]), dictout["timestamp"])
         #await C.unsubscribe(['a/b'])
         #await C.disconnect()
     except ClientException as ce:
         logger.error("Client exception: %s" % ce)
 
 
-def publishData(data, sensorID, sensorType, dataTimestamp):
+def publishData(data, sensorID, deviceID, sensorType, dataTimestamp):
     with Plugin() as plugin:
             if showData is True:
-                print("publishing network.bridge.sensor." + sensorType + " with metadata:", {"sensorID": sensorID, "sensorType": sensorType}, "and timestamp:", dataTimestamp, "and data:", data)
+                print("publishing network.bridge.sensor." + sensorType + " with metadata:", {"sensorID": sensorID, "deviceID": deviceID, "sensorType": sensorType}, "and timestamp:", dataTimestamp, "and data:", data)
             else:
-                print("publishing network.bridge.sensor." + sensorType + "with metadata", {"sensorID": sensorID, "sensorType": sensorType})
+                print("publishing network.bridge.sensor." + sensorType + "with metadata", {"sensorID": sensorID, "deviceID": deviceID, "sensorType": sensorType})
+            # send data to waggle:
+            plugin.publish("network.bridge.sensor." + sensorType, data, meta={"sensorID": str(sensorID), "deviceID": str(deviceID), "sensorType": str(sensorType)}, timestamp=dataTimestamp)
+
 
-            plugin.publish("network.bridge.sensor." + sensorType, data, meta={"sensorID": str(sensorID), "sensorType": str(sensorType)}, timestamp=dataTimestamp)
-            #time.sleep(1)
-            print("DONE")
 
 def subdict(d, header):
-  # coordinate mqtt topic names and waggle topics / sensor names
+    # coordinate mqtt topic names and waggle topics / sensor names
-  for k,v in d.items():        
+    global sensorcount
-     if isinstance(v, dict):
+    global devicecount
-        if header is not None:
+    for k,v in d.items():        
-            subdict(v, header + k + "/") # recurse further
+        if isinstance(v, dict):
+            if header is not None:
+                subdict(v, header + k + "/") # recurse further
+            else:
+                subdict(v, k + "/")
         else:            
-            subdict(v, k + "/")
+            for sensor in v:
-     else:            
+                sublist.append((header + k + "/" + sensor, QOS_2))
-        for sensor in v:
+                sensorlist.append((header + k + "/" + sensor, sensorcount))
-            sublist.append((header + k + "/" + sensor, QOS_2))
+                sensorcount = sensorcount + 1
-            #print (header + k + "/" + sensor)
+                if (header + k + "/", devicecount - 1) not in devicelist:
+                    devicelist.append((header + k + "/", devicecount))
+                    devicecount = devicecount + 1
+
+                #print (((header + k + "/" + sensor, QOS_2), sensorcount))
 
 def decoder(data):
     print (data)
@@ -97,7 +113,9 @@ if __name__ == "__main__":
     with open('subscribe.yaml', 'r') as fileread:
         clientconfig = yaml.safe_load(fileread)
     subdict(clientconfig, None)
-    print(sublist)
+    print("sublist:", sublist)
+    print("sensors:", sensorlist)
+    print("devices:", devicelist)
 
     formatter = "[%(asctime)s] :: %(levelname)s :: %(name)s :: %(message)s"
     logging.basicConfig(level=logging.ERROR, format=formatter)

subscribe.yaml

@@ -2,10 +2,13 @@ sensors:
   pico1:
     - "distance"
     - "uptime"
-  pico2:
-    cameras:
-      - "1"
-      - "2"
   pi1:
     - "camera"
     - "uptime"
+  subnet2:
+    pi2:
+      - "microphone"
+      - "uptime"
+    pico1:
+      - "soil"
+      - "uptime"