aqara-mqtt/src/mqtt.py

import paho.mqtt.client as mqtt
import os
import logging
from queue import Queue
from threading import Thread
import json

_LOGGER = logging.getLogger(__name__)

class Mqtt:
	username = ""
	password = ""
	server = "localhost"
	port = 1883
	prefix = "home"

	_client = None
	_sids = None
	_queue = None
	_threads = None

	def __init__(self, config):
		if (config == None):
			raise "Config is null"

		#load sids dictionary
		self._sids = config.get("sids", None)
		if (self._sids == None):
			self._sids = dict({})

		#load mqtt settings
		mqttConfig = config.get("mqtt", None)
		if (mqttConfig == None):
			raise "Config mqtt section is null"

		self.username = mqttConfig.get("username", "")
		self.password = mqttConfig.get("password", "")
		self.server = mqttConfig.get("server", "localhost")
		self.port = mqttConfig.get("port", 1883)
		self.prefix = mqttConfig.get("prefix", "home")
		self._queue = Queue()
		self._threads = []

	def connect(self):
		_LOGGER.info("Connecting to MQTT server " + self.server + ":" + str(self.port) + " with username (" + self.username + ":" + self.password + ")")
		self._client = mqtt.Client()
		if (self.username != "" and self.password != ""):
			self._client.username_pw_set(self.username, self.password)
		self._client.on_message = self._mqtt_process_message
		self._client.on_connect = self._mqtt_on_connect
		self._client.connect(self.server, self.port, 60)

		#run message processing loop
		t1 = Thread(target=self._mqtt_loop)
		t1.start()
		self._threads.append(t1)

	def subscribe(self, model="+", name="+", prop="+", command=None):
		topic = self.prefix + "/" + model + "/" + name + "/" + prop
		if command is not None:
			topic +=  "/" + command
		_LOGGER.info("Subscibing to " + topic + ".")
		self._client.subscribe(topic)

	def publish(self, model, sid, data, retain=True):
		sidprops = self._sids.get(sid, None)
		if (sidprops != None):
			model = sidprops.get("model",model)
			sid = sidprops.get("name",sid)

		# _LOGGER.info("data is " + format(data))
		PATH_FMT = self.prefix + "/{model}/{sid}/{prop}"
		for key, value in data.items():
			# fix for latest motion value
			if (model == "motion" and key == "no_motion"):
				key="status"
				value="no_motion"
			if (model == "magnet" and key == "no_close"):
				key="status"
				value="open"

			# do not retain event-based sensors (like switches and cubes).
			if (model in ["switch", "cube"]):
				retain = False

			# fix for rgb format
			if (key == "rgb" and self._is_int(value)):
				value = self._color_xiaomi_to_rgb(value)

			topic = PATH_FMT.format(model=model, sid=sid, prop=key)
			_LOGGER.info("Publishing message to topic " + topic + ": " + str(value) + ".")
			self._client.publish(topic, payload=value, qos=0, retain=retain)

	def _mqtt_on_connect(self, client, userdata, rc, unk):
		_LOGGER.info("Connected to mqtt server.")

	def _mqtt_process_message(self, client, userdata, msg):
		_LOGGER.info("Processing message in " + str(msg.topic) + ": " + str(msg.payload) + ".")
		parts = msg.topic.split("/")
		if len(parts) < 4:
			# should we return an error message ?
			return

		model = parts[1]
		query_sid = parts[2] #sid or name part
		param = parts[3] #param part
		method = None
		if len(parts) > 4:
			method = parts[4]
		else:
			method = parts[3]

		name = "" # we will find it next
		sid = query_sid
		isFound = False
		for current_sid in self._sids:
			if (current_sid == None):
				continue
			sidprops = self._sids.get(current_sid, None)
			if sidprops == None:
				continue
			sidname = sidprops.get("name", current_sid)
			sidmodel = sidprops.get("model", "")
			if (sidname == query_sid and sidmodel == model):
				sid = current_sid
				name = sidname
				isFound = True
				break
			else:
				_LOGGER.debug(sidmodel + "-" + sidname + " is not " + model + "-" + query_sid + ".")
				continue

		if isFound == False:
			# should we return an error message ?
			return

		if method == "set":
			# use single value set method

			value = (msg.payload).decode('utf-8')
			if self._is_int(value):
				value = int(value)

			# fix for rgb format
			if (param == "rgb" and "," in str(value)):
				value = self._color_rgb_to_xiaomi(value)

			# prepare values dict
			data = {'sid': sid, 'model': model, 'name': name,
					'values': {param: value}}
			# put in process queuee
			self._queue.put(data)
		
		elif method == "write":
			# use raw write method to the sensor, we expect a jsonified dict here.
			values = json.loads((msg.payload).decode('utf-8'))
			data = {'sid': sid, 'model': model, 'name': name,
					'values': values}
			# put in process queuee
			self._queue.put(data)

	def _mqtt_loop(self):
		_LOGGER.info("Starting mqtt loop.")
		self._client.loop_forever()

	def _color_xiaomi_to_rgb(self, xiaomi_color):
		intval = int(xiaomi_color)
		blue =  (intval) & 255
		green = (intval >> 8) & 255
		red =  (intval >> 16) & 255
		bright = (intval >> 24) & 255
		value = str(red)+","+str(green)+","+str(blue)+","+str(bright)
		return value

	def _color_rgb_to_xiaomi(self, rgb_string):
		arr = rgb_string.split(",")
		r = int(arr[0])
		g = int(arr[1])
		b = int(arr[2])
		if len(arr)>3:
			bright = int(arr[3])
		else:
			bright = 255
		value = int('%02x%02x%02x%02x' % (bright, r, g, b), 16)
		return value

	def _is_int(self, x):
		try:
			tmp = int(x)
			return True
		except Exception as e:
			return False