Make it work

broker.yaml

@@ -10,7 +10,7 @@ listeners:
 timeout-disconnect-delay: 2
 auth:
     plugins: ['auth.anonymous'] #List of plugins to activate for authentication among all registered plugins
-    allow-anonymous: true
+    allow-anonymous: true # for now
     #password-file: /some/passwd_file
 topic-check:
     enabled: false  # Set to False if topic filtering is not needed

client.py

@@ -5,14 +5,20 @@ import asyncio
 
 from amqtt.client import MQTTClient
 from amqtt.mqtt.constants import QOS_0, QOS_1, QOS_2
+import time
+import json
 
 logger = logging.getLogger(__name__)
 
+def encoder(inputdict):
+    return json.dumps(inputdict).encode('utf-8')
+
 async def test_coro():
     C = MQTTClient()
-    await C.connect('mqtt://test.mosquitto.org/')
+    await C.connect('mqtt://127.0.0.1:1883/')
     tasks = [
-        asyncio.ensure_future(C.publish('a/b', b'TEST MESSAGE WITH QOS_0')),
+
+        asyncio.ensure_future(C.publish('sensors/pico1/distance', encoder({"data": 23.0, "sensorID": 0, "timestamp": time.time_ns()}))),
         asyncio.ensure_future(C.publish('a/b', b'TEST MESSAGE WITH QOS_1', qos=QOS_1)),
         asyncio.ensure_future(C.publish('a/b', b'TEST MESSAGE WITH QOS_2', qos=QOS_2)),
     ]

main.py

@@ -10,8 +10,9 @@ from amqtt.broker import Broker
 import yaml
 from amqtt.client import MQTTClient, ClientException
 from amqtt.mqtt.constants import QOS_1, QOS_2
+import json
 
-
+sublist = []
 
 sensortypes = ["microphone", "camera", "humidity"]
 showData = True
@@ -25,16 +26,21 @@ async def uptime_coro():
     await C.connect('mqtt://127.0.0.1:1883/')
     # Subscribe to '$SYS/broker/uptime' with QOS=1
     # Subscribe to '$SYS/broker/load/#' with QOS=2
-    await C.subscribe([
+    #await C.subscribe([
-            ('a/b', QOS_1),
+    #        ('a/b', QOS_1),
-         ])
+    #     ])
+    await C.subscribe(sublist)
     try:
-        for i in range(1, 100):
+        count = 0
+        while True:
+            count = count + 1
             message = await C.deliver_message()
             packet = message.publish_packet
-            print("%d:  %s => %s" % (i, packet.variable_header.topic_name, str(packet.payload.data)))
+            dictout = decoder(packet.payload.data)
-        await C.unsubscribe(['a/b'])
+            print("%d:  %s => %s" % (count, packet.variable_header.topic_name, dictout))
-        await C.disconnect()
+            publishData(dictout["data"], dictout["sensorID"], 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)
 
@@ -46,16 +52,38 @@ def publishData(data, sensorID, sensorType, dataTimestamp):
             else:
                 print("publishing network.bridge.sensor." + sensorType + "with metadata", {"sensorID": sensorID, "sensorType": sensorType})
             
-            plugin.publish("network.bridge.sensor." + sensorType, data, meta={"sensorID": sensorID, "sensorType": sensorType}, timestamp=dataTimestamp)
+            plugin.publish("network.bridge.sensor." + sensorType, data, meta={"sensorID": str(sensorID), "sensorType": str(sensorType)}, timestamp=dataTimestamp)
             #time.sleep(1)
 
+def subdict(d, header):
+  for k,v in d.items():        
+     if isinstance(v, dict):
+        if header is not None:
+            subdict(v, header + k + "/")
+        else:
+            subdict(v, k + "/")
+     else:            
+        for sensor in v:
+            sublist.append((header + k + "/" + sensor, QOS_2))
+            #print (header + k + "/" + sensor)
+
+def decoder(data):
+    print (data)
+    return json.loads(data.decode("utf-8"))
+
 if __name__ == "__main__":
-    # testing
+    # mqtt broker config
     with open('broker.yaml', 'r') as fileread:
         brokerconfig = yaml.safe_load(fileread)
 
+    # clients config
+    with open('subscribe.yaml', 'r') as fileread:
+        clientconfig = yaml.safe_load(fileread)
+    subdict(clientconfig, None)
+    print(sublist)
+
     formatter = "[%(asctime)s] :: %(levelname)s :: %(name)s :: %(message)s"
-    logging.basicConfig(level=logging.INFO, format=formatter)
+    logging.basicConfig(level=logging.ERROR, format=formatter)
     asyncio.get_event_loop().run_until_complete(broker_coro(brokerconfig))
     asyncio.get_event_loop().run_until_complete(uptime_coro())
     asyncio.get_event_loop().run_forever()

run.sh

@@ -1,4 +1,9 @@
 #!/bin/bash
 
 export PYWAGGLE_LOG_DIR=test-run
-python main.py
+python main.py &
+sleep 1
+python client.py 2&>/dev/null > /dev/null
+sleep 5
+jobs -p
+kill $(jobs -p)

subscribe.yaml

@@ -1,3 +1,10 @@
 sensors:
   pico1:
-    - "distance"
+    - "distance"
+    - "uptime"
+  pico2:
+    cameras:
+      - "1"
+      - "2"
+  pi1:
+    - "humidity"

test-run/data.ndjson

@@ -3,3 +3,6 @@
 {"meta":{"sensorID":"0","sensorType":"microphone"},"name":"network.bridge.sensor.microphone","timestamp":"2023-03-21T14:52:40.885835367","value":1}
 {"meta":{"sensorID":"0","sensorType":"microphone"},"name":"network.bridge.sensor.microphone","timestamp":"2023-03-21T14:53:10.464592551","value":1}
 {"meta":{"sensorID":"0","sensorType":"microphone"},"name":"network.bridge.sensor.microphone","timestamp":"2023-03-21T14:53:17.921731597","value":1}
+{"meta":{"sensorID":"0","sensorType":"distance"},"name":"network.bridge.sensor.distance","timestamp":"2023-03-23T16:25:51.990142198","value":23.0}
+{"meta":{"sensorID":"0","sensorType":"distance"},"name":"network.bridge.sensor.distance","timestamp":"2023-03-23T16:27:40.431504756","value":23.0}
+{"meta":{"sensorID":"0","sensorType":"distance"},"name":"network.bridge.sensor.distance","timestamp":"2023-03-23T16:28:15.329268361","value":23.0}