SolarManager/wsMQTTbridge.py

import websocket
import time
import json
import dataclasses
import gc, logging as log
import threading
import paho.mqtt.client as mqtt
unacked_publish = set()

mqttClient=mqtt.Client(mqtt.CallbackAPIVersion.VERSION2)
#mqttClient.on_publish = on_publish
mqttClient.user_data_set(unacked_publish)

def publish(dict,topic="weatherStation"):
    if dataclasses.is_dataclass(dict):
        for field in dataclasses.fields(dict):
            key = field.name
            value = getattr(dict, field.name)
            if hasattr(value, '__len__') and (not isinstance(value, str)):
                if dataclasses.is_dataclass(value): #make subtopics of dataclasses
                    publish(value,topic+"/"+key)
                else:
                    #try:    #make subtopics of dataclass arrays
                    if len(value):
                        if dataclasses.is_dataclass(value[0]):
                            i=0
                            for d in value:
                                publish(d,topic+"/"+key+str(i))
                                i=i+1
                    #except:
                        #else:
                            #mqttClient.publish(topic+"/"+key,json.dumps(value, cls=EnhancedJSONEncoder),0,True)
                    #else:   #output empty array anyway to keep data consistent
                        #mqttClient.publish(topic+"/"+key,json.dumps(value, cls=EnhancedJSONEncoder),0,True)
            else:
                try:
                    flt = float(value)
                    mqttClient.publish(topic+"/"+key,round(flt,2),0,True)
                except ValueError:
                    mqttClient.publish(topic+"/"+key,value,0,True)
    else:
        for key, value in dict.items():
            if hasattr(value, '__len__') and (not isinstance(value, str)):
                if dataclasses.is_dataclass(value): #make subtopics of dataclasses
                    publish(value,topic+"/"+key)
                else:
                    #try:    #make subtopics of dataclass arrays
                    if len(value):
                        if dataclasses.is_dataclass(value[0]):
                            i=0
                            for d in value:
                                publish(d,topic+"/"+key+str(i))
                                i=i+1
                    #except:
                        #else:
                            #mqttClient.publish(topic+"/"+key,json.dumps(value, cls=EnhancedJSONEncoder),0,True)
                    #else:   #output empty array anyway to keep data consistent
                        #mqttClient.publish(topic+"/"+key,json.dumps(value, cls=EnhancedJSONEncoder),0,True)
            else:
                try:
                    flt = float(value)
                    mqttClient.publish(topic+"/"+key,round(flt,2),0,True)
                except ValueError:
                    mqttClient.publish(topic+"/"+key,value,0,True)
                except:
                	log.error(f"value is no number")
                
    #print(message)

def startMqttClient():
   mqttClient.connect("localhost", 1883, 60)
   mqttClient.loop_start()
        #await asyncio.Future()  # run forever
def closeMqttClient():
    mqttClient.disconnect()



def on_message(ws, message, msgCounter=[0]):
	msgCounter[0] += 1
	data = json.loads(message)
	if msgCounter[0] >1:
		if "tempAmb" in data:
			data.pop("tempAmb")
		if "hum" in data:
			data.pop("hum")
		if "qff" in data:
			data.pop("qff")
		if "dewpt" in data:
			data.pop("dewpt")
		if "iaq" in data:
			data.pop("iaq")
		if "avgWindspeed" in data:
			data.pop("avgWindspeed")
		if "avgWindDeg" in data:
			data.pop("avgWindDeg")
		if msgCounter[0] > 20:
			msgCounter[0] = 0
	
	publish(data,topic="weatherStation")

def on_error(ws, error):
    #ws.web_socket_open = False
    #ws.logged_in = False
    #print("WebSocket Error ")
    #print("Reconnect to the endpoint after 3 seconds... ")
    #time.sleep(3)
    #try:
    #    ws.run_forever()
    #except Exception as error:
    print(f"Encountered error: {error}")

def on_close(ws, close_status_code, close_msg):
    ws.web_socket_open = False
    ws.logged_in = False
    print("WebSocket Closed")


def on_open(ws):
    print("Connection opened")
    ws.send("Hello, Server!")


if __name__ == "__main__":
    startMqttClient()
    while True:
        try:
            ws = websocket.WebSocketApp("ws://192.168.179.42/ws",
                            on_message=on_message,
                            on_error=on_error,
                            on_close=on_close)
            ws.on_open = on_open
            ws.run_forever(skip_utf8_validation=True,ping_interval=10,ping_timeout=8)
        except Exception as e:
            gc.collect()
            print("Websocket connection Error  : {0}".format(e))                    
        print("Reconnecting websocket  after 10 sec")
        time.sleep(10)
    #ws.run_forever()