From dd7790dab8d3fbea8f2b58eb4d5aaffc36b3cb09 Mon Sep 17 00:00:00 2001 From: jwansek Date: Sun, 16 Feb 2025 14:16:43 +0000 Subject: Renamed the scripts to adhere to python conventions --- tasmota-http-cli.py | 54 ------------------------- tasmota-mqtt-client.py | 104 ------------------------------------------------- tasmotaHTTPClient.py | 54 +++++++++++++++++++++++++ tasmotaMQTTClient.py | 104 +++++++++++++++++++++++++++++++++++++++++++++++++ 4 files changed, 158 insertions(+), 158 deletions(-) delete mode 100644 tasmota-http-cli.py delete mode 100644 tasmota-mqtt-client.py create mode 100644 tasmotaHTTPClient.py create mode 100644 tasmotaMQTTClient.py diff --git a/tasmota-http-cli.py b/tasmota-http-cli.py deleted file mode 100644 index c151f3f..0000000 --- a/tasmota-http-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-mqtt-client.py b/tasmota-mqtt-client.py deleted file mode 100644 index e8a43b8..0000000 --- a/tasmota-mqtt-client.py +++ /dev/null @@ -1,104 +0,0 @@ -import paho.mqtt.client as paho -import threading -import argparse -import getpass -import docker_net -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") - topic = "tele/TasmotaPlug/%s/+" % self.friendlyname - if self.message is None: - print("Waiting for '%s'" % topic) - self.mqtt_c.subscribe(topic) - - else: - self.mqtt_c.publish("cmnd/TasmotaPlug/%s/Power" % self.friendlyname, payload = self.message) - print("Sent message '%s' to topic 'cmnd/TasmotaPlug/%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("/")[2] == self.friendlyname: - if msg.topic.split("/")[3] == "SENSOR": - self.switch_energy = json.loads(msg.payload.decode())["ENERGY"] - elif msg.topic.split("/")[3] == "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", - default = "" -) -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["mqtt_host"] == "": - args["mqtt_host"] = docker_net.get_mqtt_addr()[0] - 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)) diff --git a/tasmotaHTTPClient.py b/tasmotaHTTPClient.py new file mode 100644 index 0000000..c151f3f --- /dev/null +++ b/tasmotaHTTPClient.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/tasmotaMQTTClient.py b/tasmotaMQTTClient.py new file mode 100644 index 0000000..e8a43b8 --- /dev/null +++ b/tasmotaMQTTClient.py @@ -0,0 +1,104 @@ +import paho.mqtt.client as paho +import threading +import argparse +import getpass +import docker_net +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") + topic = "tele/TasmotaPlug/%s/+" % self.friendlyname + if self.message is None: + print("Waiting for '%s'" % topic) + self.mqtt_c.subscribe(topic) + + else: + self.mqtt_c.publish("cmnd/TasmotaPlug/%s/Power" % self.friendlyname, payload = self.message) + print("Sent message '%s' to topic 'cmnd/TasmotaPlug/%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("/")[2] == self.friendlyname: + if msg.topic.split("/")[3] == "SENSOR": + self.switch_energy = json.loads(msg.payload.decode())["ENERGY"] + elif msg.topic.split("/")[3] == "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", + default = "" +) +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["mqtt_host"] == "": + args["mqtt_host"] = docker_net.get_mqtt_addr()[0] + 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