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,55 +4,62 @@ import argparse
 import json
 
 import paho.mqtt.client as mqtt
-import mixer
 from mixer import Mixer
 
 
-MQTT_SERVER_PORT = 1883
-MQTT_DISCOVERY_TOPIC = "homeassistant/button/mixer/config"
-MQTT_COMMAND_TOPIC = "livingroom/voc/allen_heath_qu16_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(self, client):
+        j = json.dumps(
+            {
+                "name": "Reset Settings",
+                "command_topic": self.MQTT_COMMAND_TOPIC,
+                "command_template": '{"command": "scene_recall", "args": {"scene_id": 0}}',
+                "device": {
+                    "name": "Mischpult",
+                    "model": "Qu-16",
+                    "manufacturer": "Allen & Heath",
+                    "suggested_area": "livingroom",
+                    "identifiers": "Mischpult",
+                },
+                "unique_id": "mischpult_reset_settings",
+            }
+        )
 
-def mqtt_autodiscovery(client):
-    j = json.dumps(
-        {
-            "name": "Reset Settings",
-            "command_topic": MQTT_COMMAND_TOPIC,
-            "command_template": '{"command": "scene_recall", "args": {"scene_id": 0}}',
-            "device": {
-                "name": "Mischpult",
-                "model": "Qu-16",
-                "manufacturer": "Allen & Heath",
-                "suggested_area": "livingroom",
-                "identifiers": "Mischpult",
-            },
-            "unique_id": "mischpult_reset_settings",
-        }
-    )
+        self, client.publish(self.MQTT_DISCOVERY_TOPIC, payload=j, qos=0, retain=False)
 
-    client.publish(MQTT_DISCOVERY_TOPIC, payload=j, qos=0, retain=False)
+    def on_connect(self, client, userdata, flags, reason_code, properties):
+        print(f"Connected: {reason_code}")
 
+        client.subscribe(self.MQTT_COMMAND_TOPIC)
+        self.mqtt_autodiscovery(client)
 
-def on_connect(client, userdata, flags, reason_code, properties):
-    print(f"Connected: {reason_code}")
-    client.subscribe(MQTT_COMMAND_TOPIC)
-    mqtt_autodiscovery(client)
+    def on_message(self, client, userdata, msg):
+        payload = msg.payload.decode()
 
+        print(f"{msg.topic}: {payload}")
 
-def on_message(client, userdata, msg):
-    payload = msg.payload.decode()
+        j = json.loads(payload)
 
-    print(f"{msg.topic}: {payload}")
+        match j["command"]:
+            case "scene_recall":
+                self.mixer.scene_recall(j["args"]["scene_id"])
+            case "shutdown":
+                self.mixer.shutdown()
 
-    j = json.loads(payload)
+    def run(self, hostname):
+        self.client.on_connect = self.on_connect
+        self.client.on_message = self.on_message
 
-    match j["command"]:
-        case "scene_recall":
-            m.scene_recall(j["args"]["scene_id"])
-        case "shutdown":
-            m.shutdown()
+        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__":