From dd7790dab8d3fbea8f2b58eb4d5aaffc36b3cb09 Mon Sep 17 00:00:00 2001
From: jwansek <eddie.atten.ea29@gmail.com>
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 = "<docker>"
-)
-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"] == "<docker>":
- 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 = "<docker>"
+)
+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"] == "<docker>":
+ 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