hass_button_emulator.py

@@ -4,7 +4,9 @@ import time
 import paho.mqtt.client as mqtt
 import json
 
-import mqtt_client
+MQTT_SERVER_ADDRESS = "localhost"
+MQTT_SERVER_PORT = 1883
+MQTT_TOPIC = "homeassistant/device/mixer"
 
 
 def on_connect(client, userdata, flags, reason_code, properties):
@@ -20,13 +22,13 @@ client = mqtt.Client(callback_api_version=mqtt.CallbackAPIVersion.VERSION2)
 client.on_connect = on_connect
 client.on_message = on_message
 
-client.connect(mqtt_client.MQTT_SERVER_ADDRESS, mqtt_client.MQTT_SERVER_PORT, 60)
+client.connect(MQTT_SERVER_ADDRESS, MQTT_SERVER_PORT, 60)
 client.loop_start()
 
-j = json.dumps({"command": "scene_recall", "args": {"scene_id": 0}})
+j = json.dumps({"command": "scene_recall", "args": {"scene_id": 42}})
 # j = json.dumps({'command': 'shutdown'})
 
-client.publish(mqtt_client.MQTT_TOPIC, payload=j, qos=0, retain=False)
+client.publish(MQTT_TOPIC, payload=j, qos=0, retain=False)
 
 time.sleep(2)
 

mixer.py

@@ -6,14 +6,12 @@ import mido
 
 from enum import Enum
 
-# For Allen & Heath Qu MIDI protocol documentation see:
-# https://www.allen-heath.com/content/uploads/2023/06/Qu_MIDI_Protocol_V1.9.pdf
+MIXER_PORT = 51325
+
 
 class Mixer:
-    def __init__(self, ip):
-        self.MIXER_PORT = 51325
-        self.sock = socket.create_connection((ip, self.MIXER_PORT))
-        self.mido_parser = mido.Parser()
+    def __init__(self, ip, port):
+        self.sock = socket.create_connection((ip, port))
 
     ALLEN_HEATH_ID = [0x00, 0x00, 0x1A]
     QU_MIXER = [0x50, 0x11]
@@ -31,16 +29,18 @@ class Mixer:
         GET_METER_DATA_RESPONSE = 0x13
 
     def recv_sys_ex(self, response_msg_filter: SysExMessageId = None):
+        p = mido.Parser()
+
         while True:
             data = self.sock.recv(1024)
 
             if not data:
                 break
 
-            self.mido_parser.feed(data)
+            p.feed(data)
 
-            if self.mido_parser.pending():
-                msg = self.mido_parser.get_message()
+            if p.pending():
+                msg = p.get_message()
 
                 print(vars(msg))
 
@@ -207,7 +207,7 @@ def main():
     args = parser.parse_args()
 
     print(f"IP: {args.ip}")
-    mixer = Mixer(args.ip)
+    mixer = Mixer(args.ip, MIXER_PORT)
 
     if args.command:
         print(f"Command: {args.command}")