cccb/mixer-rpc
cccb/mixer-rpc
0
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

Compare commits

base: cccb:2ae803ebfece22f9f0b6be0628dbfaa14007a333
Branches Tags
cccb:main
..
compare: cccb:cd65e57b67125d3afe79784edefd6afadd4b14ed
Branches Tags
cccb:main

4 commits
2ae803ebfe ... cd65e57b67

Author SHA1 Message Date
coon
cd65e57b67 add comment with link to A&H midi spec 2025-09-25 22:00:40 +02:00
coon
238aada4a1 use MQTT credentials from mqtt_client.py 2025-09-25 21:17:51 +02:00
coon
0f3ab79bdb make mido parser a class var 2025-09-25 21:04:43 +02:00
coon
3e123bc751 move MIXER_PORT into Mixer class 2025-09-25 20:59:28 +02:00
2 changed files with 14 additions and 16 deletions
Download patch file Download diff file Expand all files Collapse all files

10
hass_button_emulator.py
View file

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

20
mixer.py
View file

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