From 8ff27687d163423726f50f2a94fd59ac93cbb0b6 Mon Sep 17 00:00:00 2001 From: jwansek Date: Sat, 16 Dec 2023 22:44:42 +0000 Subject: Added MQTT CLI --- requirements.txt | 2 + tasmota-cli.py | 54 --------------------------- tasmota-http-cli.py | 54 +++++++++++++++++++++++++++ tasmota-mqtt-client.py | 99 ++++++++++++++++++++++++++++++++++++++++++++++++++ 4 files changed, 155 insertions(+), 54 deletions(-) delete mode 100644 tasmota-cli.py create mode 100644 tasmota-http-cli.py create mode 100644 tasmota-mqtt-client.py diff --git a/requirements.txt b/requirements.txt index c292ad5..f6950d8 100644 --- a/requirements.txt +++ b/requirements.txt @@ -1,2 +1,4 @@ tasmotadevicecontroller aiohttp +paho-mqtt + diff --git a/tasmota-cli.py b/tasmota-cli.py deleted file mode 100644 index c151f3f..0000000 --- a/tasmota-cli.py +++ /dev/null @@ -1,54 +0,0 @@ -import asyncio -import tasmotadevicecontroller -import argparse -import getpass -import json -import time - -async def main(host, username, password, toggle): - if username is None: - device = await tasmotadevicecontroller.TasmotaDevice.connect(host) - else: - device = await tasmotadevicecontroller.TasmotaDevice.connect(url = host, username = username, password = password) - - if toggle: - await device.setPower(tasmotadevicecontroller.tasmota_types.PowerType.TOGGLE) - time.sleep(2) - - friendlyname = await device.getFriendlyName() - power = await device.getPower() - status8 = await device.sendRawRequest("status 8") - watts = status8["StatusSNS"]["ENERGY"] - print(json.dumps(watts, indent = 4)) - - if power: - print("'%s' is currently ON" % friendlyname) - else: - print("'%s' is currently OFF" % friendlyname) - -parser = argparse.ArgumentParser() -parser.add_argument( - "-d", "--device-host", - type = str, - help = "Tasmota host port", - required = True -) -parser.add_argument( - "-u", "--user", - type = str, - help = "Username to login with" -) -parser.add_argument( - "-t", "--toggle", - action = "store_true", - help = "Toggle current power status" -) - -if __name__ == "__main__": - args = vars(parser.parse_args()) - if args["user"] is not None: - args["password"] = getpass.getpass("Input password for %s@%s: " % (args["user"], args["device_host"])) - else: - args["password"] = None - loop = asyncio.get_event_loop() - loop.run_until_complete(main(args["device_host"], args["user"], args["password"], args["toggle"])) diff --git a/tasmota-http-cli.py b/tasmota-http-cli.py new file mode 100644 index 0000000..c151f3f --- /dev/null +++ b/tasmota-http-cli.py @@ -0,0 +1,54 @@ +import asyncio +import tasmotadevicecontroller +import argparse +import getpass +import json +import time + +async def main(host, username, password, toggle): + if username is None: + device = await tasmotadevicecontroller.TasmotaDevice.connect(host) + else: + device = await tasmotadevicecontroller.TasmotaDevice.connect(url = host, username = username, password = password) + + if toggle: + await device.setPower(tasmotadevicecontroller.tasmota_types.PowerType.TOGGLE) + time.sleep(2) + + friendlyname = await device.getFriendlyName() + power = await device.getPower() + status8 = await device.sendRawRequest("status 8") + watts = status8["StatusSNS"]["ENERGY"] + print(json.dumps(watts, indent = 4)) + + if power: + print("'%s' is currently ON" % friendlyname) + else: + print("'%s' is currently OFF" % friendlyname) + +parser = argparse.ArgumentParser() +parser.add_argument( + "-d", "--device-host", + type = str, + help = "Tasmota host port", + required = True +) +parser.add_argument( + "-u", "--user", + type = str, + help = "Username to login with" +) +parser.add_argument( + "-t", "--toggle", + action = "store_true", + help = "Toggle current power status" +) + +if __name__ == "__main__": + args = vars(parser.parse_args()) + if args["user"] is not None: + args["password"] = getpass.getpass("Input password for %s@%s: " % (args["user"], args["device_host"])) + else: + args["password"] = None + loop = asyncio.get_event_loop() + loop.run_until_complete(main(args["device_host"], args["user"], args["password"], args["toggle"])) diff --git a/tasmota-mqtt-client.py b/tasmota-mqtt-client.py new file mode 100644 index 0000000..849ae9d --- /dev/null +++ b/tasmota-mqtt-client.py @@ -0,0 +1,99 @@ +import paho.mqtt.client as paho +import threading +import argparse +import getpass +import time +import json + +SENT_EVENT = threading.Event() + +class MQTTClient: + + switch_energy = None + switch_power = None + + def __init__(self, host, friendlyname, username, password, message = None): + self.host = host + self.friendlyname = friendlyname + self.username = username + self.password = password + self.message = message + + # print("Instantiated. Message:", message) + + self.mqtt_c = paho.Client("tasmota-cli", clean_session = True) + + self.mqtt_c.on_connect = self._on_connect_cb + if message is None: + self.mqtt_c.on_message = self._on_message_cb + + self.mqtt_c.username_pw_set(username = username, password = password) + self.mqtt_c.connect(self.host, 1883, 60) + self.mqtt_c.loop_forever() + + def _on_connect_cb(self, mqtt, userdata, flags, rc): + # print("Connected to broker") + if self.message is None: + self.mqtt_c.subscribe("tele/%s/+" % self.friendlyname) + + else: + self.mqtt_c.publish("cmnd/%s/Power" % self.friendlyname, payload = self.message) + print("Sent message '%s' to topic 'cmnd/%s/Power'" % (self.message, self.friendlyname)) + SENT_EVENT.set() + self.mqtt_c.disconnect() + + def _on_message_cb(self, mqtt, userdata, msg): + # print('Topic: {0} | Message: {1}'.format(msg.topic, msg.payload)) + + if msg.topic.split("/")[1] == self.friendlyname: + if msg.topic.split("/")[2] == "SENSOR": + self.switch_energy = json.loads(msg.payload.decode())["ENERGY"] + elif msg.topic.split("/")[2] == "STATE": + self.switch_power = json.loads(msg.payload.decode())["POWER"] + + if self.switch_power is not None and self.switch_energy is not None: + self.mqtt_c.disconnect() + + + +parser = argparse.ArgumentParser() +parser.add_argument( + "-m", "--mqtt-host", + type = str, + help = "MQTT Server", + required = True +) +parser.add_argument( + "-u", "--user", + type = str, + help = "Username to login with", + required = True +) +parser.add_argument( + "-t", "--toggle", + action = "store_true", + help = "Toggle current power status" +) +parser.add_argument( + "-n", "--friendlyname", + help = "Device friendly name", + type = str, + required = True +) + +if __name__ == "__main__": + args = vars(parser.parse_args()) + if args["user"] is not None: + args["password"] = getpass.getpass("Input password for %s@%s: " % (args["user"], args["mqtt_host"])) + else: + args["password"] = None + + if args["toggle"]: + client = MQTTClient(args["mqtt_host"], args["friendlyname"], args["user"], args["password"], "TOGGLE") + # print("Waiting for event...") + SENT_EVENT.wait() + # print("Done waiting.") + + client = MQTTClient(args["mqtt_host"], args["friendlyname"], args["user"], args["password"]) + print(json.dumps(client.switch_energy, indent = 4)) + print("'%s' is currently %s" % (client.friendlyname, client.switch_power)) -- cgit v1.2.3