diff --git a/broker.yaml b/broker.yaml index 532c287..c3dc36d 100644 --- a/broker.yaml +++ b/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 \ No newline at end of file diff --git a/client.py b/client.py index d80477f..471e64f 100644 --- a/client.py +++ b/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)), ] diff --git a/main.py b/main.py index 9d125a3..28449e2 100644 --- a/main.py +++ b/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([ - ('a/b', QOS_1), - ]) + #await C.subscribe([ + # ('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))) - await C.unsubscribe(['a/b']) - await C.disconnect() + dictout = decoder(packet.payload.data) + print("%d: %s => %s" % (count, packet.variable_header.topic_name, dictout)) + 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() diff --git a/run.sh b/run.sh index 92aa61b..97b5044 100755 --- a/run.sh +++ b/run.sh @@ -1,4 +1,9 @@ #!/bin/bash export PYWAGGLE_LOG_DIR=test-run -python main.py \ No newline at end of file +python main.py & +sleep 1 +python client.py 2&>/dev/null > /dev/null +sleep 5 +jobs -p +kill $(jobs -p) \ No newline at end of file diff --git a/subscribe.yaml b/subscribe.yaml index a888d24..0d577ca 100644 --- a/subscribe.yaml +++ b/subscribe.yaml @@ -1,3 +1,10 @@ sensors: pico1: - - "distance" \ No newline at end of file + - "distance" + - "uptime" + pico2: + cameras: + - "1" + - "2" + pi1: + - "humidity" \ No newline at end of file diff --git a/test-run/data.ndjson b/test-run/data.ndjson index 39e2f0b..6ad8df1 100644 --- a/test-run/data.ndjson +++ b/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}