54 lines
1.8 KiB
Python
54 lines
1.8 KiB
Python
import threading
|
|
import paho.mqtt.client as mqtt
|
|
from logger import Logger
|
|
import json
|
|
import datetime
|
|
from time import mktime
|
|
|
|
|
|
CA_CERTS = "VeriSign-Class 3-Public-Primary-Certification-Authority-G5.pem"
|
|
CERT_FILE = "aws-certificate.pem.crt"
|
|
KEY_FILE = "aws-private.pem.key"
|
|
MQTT_HOST = "a86hx9xnv9yty.iot.us-west-2.amazonaws.com"
|
|
MQTT_PORT = 8883
|
|
CLIENT_ID = "d0bf9c206e14x"
|
|
|
|
class MyEncoder(json.JSONEncoder):
|
|
|
|
def default(self, obj):
|
|
if isinstance(obj, datetime.datetime):
|
|
return str(obj)
|
|
|
|
return json.JSONEncoder.default(self, obj)
|
|
|
|
class AwsPublish(threading.Thread):
|
|
def __init__(self, queue):
|
|
threading.Thread.__init__(self)
|
|
self.queue = queue
|
|
self.setDaemon(True)
|
|
|
|
def run(self):
|
|
client = mqtt.Client()
|
|
client.tls_set(ca_certs=CA_CERTS, certfile=CERT_FILE, keyfile=KEY_FILE)
|
|
client.connect(MQTT_HOST, MQTT_PORT, 60)
|
|
|
|
client.loop_start()
|
|
|
|
while True:
|
|
try:
|
|
msg = self.queue.get()
|
|
dataBlock = msg['data']
|
|
if 'decodedTelegram' in dataBlock:
|
|
del dataBlock['decodedTelegram']
|
|
if 'telegram' in dataBlock:
|
|
del dataBlock['telegram']
|
|
metadataBlock = msg['metadata']
|
|
if 'Slave' in metadataBlock:
|
|
if metadataBlock['Slave'] == 'Thermometer' and metadataBlock['device'] == 'ModbusHub':
|
|
metadataBlock['name'] = 'FridgeThermometer'
|
|
topic = "IoT/ParsedData/%s" % metadataBlock['name']
|
|
client.publish(topic, json.dumps(msg, cls=MyEncoder))
|
|
Logger.log("RePublisher has sent data")
|
|
except Exception, e:
|
|
Logger.log("Unexcepted exception %s in RePublisher: %s" % (e.__class__.__name__, str(e)))
|