hass_button_emulator.py: ruff format

hass_button_emulator.py

@@ -22,13 +22,15 @@ def main():
     client = mqtt.Client(callback_api_version=mqtt.CallbackAPIVersion.VERSION2)
     client.on_connect = on_connect
 
-    client.connect(args.mqtt_hostname, mqtt_client.MQTT_SERVER_PORT, 60)
+    client.connect(args.mqtt_hostname, mqtt_client.MqttClient.MQTT_SERVER_PORT, 60)
     client.loop_start()
 
     j = json.dumps({"command": "scene_recall", "args": {"scene_id": 0}})
     # j = json.dumps({'command': 'shutdown'})
 
-    client.publish(mqtt_client.MQTT_TOPIC, payload=j, qos=0, retain=False)
+    client.publish(
+        mqtt_client.MqttClient.MQTT_COMMAND_TOPIC, payload=j, qos=0, retain=False
+    )
 
     time.sleep(2)
 

mqtt_client.py

@@ -4,22 +4,23 @@ import argparse
 import json
 
 import paho.mqtt.client as mqtt
-import mixer
 from mixer import Mixer
 
 
+class MqttClient:
     MQTT_SERVER_PORT = 1883
     MQTT_DISCOVERY_TOPIC = "homeassistant/button/mixer/config"
     MQTT_COMMAND_TOPIC = "livingroom/voc/allen_heath_qu16_mixer"
 
-m: Mixer = None
+    def __init__(self, mixer_ip):
+        self.mixer = Mixer(mixer_ip)
+        self.client = mqtt.Client(callback_api_version=mqtt.CallbackAPIVersion.VERSION2)
 
-
-def mqtt_autodiscovery(client):
+    def mqtt_autodiscovery(self, client):
         j = json.dumps(
             {
                 "name": "Reset Settings",
-            "command_topic": MQTT_COMMAND_TOPIC,
+                "command_topic": self.MQTT_COMMAND_TOPIC,
                 "command_template": '{"command": "scene_recall", "args": {"scene_id": 0}}',
                 "device": {
                     "name": "Mischpult",
@@ -32,16 +33,15 @@ def mqtt_autodiscovery(client):
             }
         )
 
-    client.publish(MQTT_DISCOVERY_TOPIC, payload=j, qos=0, retain=False)
+        self, client.publish(self.MQTT_DISCOVERY_TOPIC, payload=j, qos=0, retain=False)
 
-
-def on_connect(client, userdata, flags, reason_code, properties):
+    def on_connect(self, client, userdata, flags, reason_code, properties):
         print(f"Connected: {reason_code}")
-    client.subscribe(MQTT_COMMAND_TOPIC)
-    mqtt_autodiscovery(client)
 
+        client.subscribe(self.MQTT_COMMAND_TOPIC)
+        self.mqtt_autodiscovery(client)
 
-def on_message(client, userdata, msg):
+    def on_message(self, client, userdata, msg):
         payload = msg.payload.decode()
 
         print(f"{msg.topic}: {payload}")
@@ -50,9 +50,16 @@ def on_message(client, userdata, msg):
 
         match j["command"]:
             case "scene_recall":
-            m.scene_recall(j["args"]["scene_id"])
+                self.mixer.scene_recall(j["args"]["scene_id"])
             case "shutdown":
-            m.shutdown()
+                self.mixer.shutdown()
+
+    def run(self, hostname):
+        self.client.on_connect = self.on_connect
+        self.client.on_message = self.on_message
+
+        self.client.connect(hostname, self.MQTT_SERVER_PORT, 60)
+        self.client.loop_forever()
 
 
 def main():
@@ -75,15 +82,8 @@ def main():
 
     print(f"Mixer IP: {args.mixer_ip}, MQTT broker hostname: {args.mqtt_hostname}")
 
-    global m
-    m = mixer.Mixer(args.mixer_ip)
-
-    client = mqtt.Client(callback_api_version=mqtt.CallbackAPIVersion.VERSION2)
-    client.on_connect = on_connect
-    client.on_message = on_message
-
-    client.connect(args.mqtt_hostname, MQTT_SERVER_PORT, 60)
-    client.loop_forever()
+    client = MqttClient(args.mixer_ip)
+    client.run(args.mqtt_hostname)
 
 
 if __name__ == "__main__":