import paho.mqtt.client as paho import tkinter as tk from tkinter import ttk import configparser import threading import requests import json import sys import os ZIGBEE_TRANSFORMATIONS = { "Power": "on", "ActivePower": "power", "RMSVoltage": "voltage", "RMSCurrent": "current", "RMSPower": "power" } TASMOTA_TRANSFORMATIONS = { "Power": "power", "Voltage": "voltage", "Current": "current", "POWER": "on", "Current": "current" } class App(tk.Tk): def __init__(self, config_path, *args, **kwargs): super().__init__(*args, **kwargs) self.title("Power Buttons") self.protocol("WM_DELETE_WINDOW", self._on_closing) self.config = configparser.ConfigParser() self.config.read(config_path) self.mqttc = paho.Client("PowerButtonsBUI", clean_session = True) self.mqttc.on_connect = self._on_connect_cb self.mqttc.on_message = self._on_message_cb self.mqttc.username_pw_set(self.config["zigbee"]["mqtt_username"], password = self.config["zigbee"]["mqtt_password"]) self.mqttc.connect(self.config["zigbee"]["mqtt_host"], 1883, 60) tk.Button(self, text = "sneed", command = lambda: print("sneed")).pack() self.img_green = tk.PhotoImage(file = os.path.join(os.path.dirname(__file__), "Assets", "green.png")) self.img_yellow = tk.PhotoImage(file = os.path.join(os.path.dirname(__file__), "Assets", "yellow.png")) self.img_red = tk.PhotoImage(file = os.path.join(os.path.dirname(__file__), "Assets", "red.png")) self.devices = {} self.device_widgets = {} self.device_types = {} self.tasmota_ips = {} for friendlyname in self.config["zigbee"]["friendlynames"].split(","): self.devices[friendlyname] = { field: None for field in ZIGBEE_TRANSFORMATIONS.values() } type_ = "Zigbee MQTT" self.device_widgets[friendlyname] = DeviceButtonWidget(self, friendlyname, type_) self.device_widgets[friendlyname].pack(fill = tk.BOTH, expand = True) self.device_types[friendlyname] = type_ for ip in self.config["tasmota"]["plugs"].split(","): friendlyname = send_raw_tasmota_http(ip, self.config["tasmota"]["password"], "DeviceName")["DeviceName"] self.devices[friendlyname] = { field: None for field in TASMOTA_TRANSFORMATIONS.values() } type_ = "Tasmota HTTP" self.device_widgets[friendlyname] = DeviceButtonWidget(self, friendlyname, type_) self.device_widgets[friendlyname].pack(fill = tk.BOTH, expand = True) self.device_types[friendlyname] = type_ self.tasmota_ips[friendlyname] = ip self.after(30 * 1000, self._after) self.mqtt_thread = threading.Thread(target = self._mqtt_thread_func) self.mqtt_thread.start() def _after(self): for ip in self.config["tasmota"]["plugs"].split(","): data = query_tasmota_power(ip, self.config["tasmota"]["password"]) fields = tasmota_query_to_fields(data) friendlyname = list(self.tasmota_ips.keys())[list(self.tasmota_ips.values()).index(ip)] self.devices[friendlyname] = fields self.device_widgets[friendlyname].update() self.after(30 * 1000, self._after) def _on_connect_cb(self, mqtt, userdata, flags, rc): print("Connected to broker") self.mqttc.subscribe(self.config["zigbee"]["mqtt_topic"]) def _on_message_cb(self, mqtt, userdata, msg): print('Topic: {0} | Message: {1}'.format(msg.topic, msg.payload)) msg_j = json.loads(msg.payload.decode()) msg_j = msg_j["ZbReceived"][list(msg_j["ZbReceived"].keys())[0]] if msg_j["Name"] in self.devices.keys(): for k, v in msg_j.items(): if k in ZIGBEE_TRANSFORMATIONS.keys(): self.devices[msg_j["Name"]][ZIGBEE_TRANSFORMATIONS[k]] = v # print(self.devices) self.device_widgets[msg_j["Name"]].update() def _mqtt_thread_func(self): self.mqttc.loop_forever() def _on_closing(self): self.mqttc.disconnect() self.destroy() class DeviceButtonWidget(tk.Frame): def __init__(self, parent:App, devicename, method): tk.Frame.__init__(self, parent) self.devicename = devicename self.method = method self.parent = parent self.btn_img = ttk.Label(self, image = self.parent.img_yellow) self.btn_img.pack(side = tk.LEFT, padx = 5, pady = 5, fill = tk.BOTH, expand = True) self.lbl_name = tk.Label(self, text = devicename, font = "-weight bold") self.lbl_name.pack(side = tk.LEFT, fill = tk.BOTH, expand = True) self.lbl_details = tk.Label(self, text = "?W ?A ?V") self.lbl_details.pack(side = tk.LEFT, padx = 5, pady = 5, fill = tk.BOTH, expand = True) self.lbl_method = tk.Label(self, text = method) self.lbl_method.pack(side = tk.RIGHT, padx = 5, pady = 5, fill = tk.BOTH, expand = True) self.btn_off = ttk.Button(self, text = "Off", command = lambda: self.set_power("off")) self.btn_off.pack(side = tk.RIGHT, padx = 5, pady = 5, fill = tk.X, expand = True) self.btn_on = ttk.Button(self, text = "On", command = lambda: self.set_power("on")) self.btn_on.pack(side = tk.RIGHT, padx = 5, pady = 5, fill = tk.X, expand = True) def set_power(self, set_to): if set_to == "on": payload = "1" elif set_to == "off": payload = "0" if self.method == "Tasmota HTTP": send_raw_tasmota_http(self.parent.tasmota_ips[self.devicename], self.parent.config["tasmota"]["password"], "Power %s" % payload) self.after(200, self.update_one_http) elif self.method == "Zigbee MQTT": self.parent.mqttc.publish( self.parent.config["zigbee"]["cmd_mqtt_topic"], json.dumps({"device": self.devicename, "send": {"power": int(payload)}}), qos = 1 ) def update_one_http(self): self.parent.devices[self.devicename] = tasmota_query_to_fields(query_tasmota_power( self.parent.tasmota_ips[self.devicename], self.parent.config["tasmota"]["password"] )) self.update() def update(self): t = "" device_info = self.parent.devices[self.devicename] if device_info["on"]: self.btn_img.configure(image = self.parent.img_green) elif device_info["on"] == False: self.btn_img.configure(image = self.parent.img_red) if device_info["power"] is not None: t += "%iW " % device_info["power"] else: t += "?W " if device_info["current"] is not None: t += "%.2fA " % device_info["current"] else: t += "?A " if device_info["voltage"] is not None: t += "%iV" % device_info["voltage"] else: t += "?V" self.lbl_details.configure(text = t) def tasmota_query_to_fields(query_results): def transform_power(p): if p == "ON": return True if p == "OFF": return False return p return {TASMOTA_TRANSFORMATIONS[k]: transform_power(v) for k, v in query_results.items() if k in TASMOTA_TRANSFORMATIONS.keys()} def query_tasmota_power(host, password): return send_raw_tasmota_http( host, password, "Status 8" )["StatusSNS"]["ENERGY"] | send_raw_tasmota_http( host, password, "Power" ) def send_raw_tasmota_http(host, password, command): req = requests.get("http://%s/cm" % host, params = { "cmnd": str(command), "user": "admin", "password": password }) return req.json() if __name__ == "__main__": app = App(sys.argv[1]) app.mainloop() # print(tasmota_query_to_fields(query_tasmota_power("geoffery.plug", "securebackdoor")))