56 lines
1.3 KiB
Python
56 lines
1.3 KiB
Python
'''
|
|
Created on 09.06.2015
|
|
|
|
@author: wn
|
|
'''
|
|
|
|
import threading
|
|
import Queue
|
|
from logger import Logger
|
|
import paho.mqtt.client as mqtt
|
|
import json
|
|
import datetime
|
|
|
|
__queue = None
|
|
|
|
def on_message(client, userdata, msg):
|
|
j = json.loads(msg.payload)
|
|
now = datetime.datetime.now()
|
|
midnight = now.replace(now.year, now.month, now.day, 0,0,0,0)
|
|
seconds = (now - midnight).seconds
|
|
j['metadata']['timestamp'] = datetime.datetime.now()
|
|
j['metadata']['seconds'] = seconds
|
|
j['metadata']['day'] = midnight
|
|
|
|
try:
|
|
__queue.put_nowait(j)
|
|
except Queue.Full:
|
|
Logger.log("Message %s dropped" % (j))
|
|
|
|
|
|
|
|
|
|
class MqttReceiver(threading.Thread):
|
|
singleton = None
|
|
|
|
@classmethod
|
|
def create(cls, queue):
|
|
if cls.singleton is None:
|
|
cls.singleton = MqttReceiver(queue)
|
|
__queue = queue
|
|
return cls.singleton
|
|
|
|
def __init__(self, queue):
|
|
threading.Thread.__init__(self)
|
|
self.queue = queue
|
|
self.setDaemon(True)
|
|
|
|
def run(self):
|
|
client = mqtt.Client()
|
|
client.on_message = on_message
|
|
client.connect("mqttbroker", 1883, 60)
|
|
client.subscribe("IoT/Measurement/#")
|
|
client.subscribe("IoT/WiFiPowerMeter/Measurement")
|
|
|
|
client.loop_forever()
|