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:927e2513e3ae37dc145eb6678c5aa1aff0b9eb4f
Branches Tags
cccb:main

No commits in common. "2ae803ebfece22f9f0b6be0628dbfaa14007a333" and "927e2513e3ae37dc145eb6678c5aa1aff0b9eb4f" have entirely different histories.
2ae803ebfe ... 927e2513e3

3 changed files with 1 additions and 91 deletions
Download patch file Download diff file Expand all files Collapse all files

36
hass_button_emulator.py
View file

@ -1,36 +0,0 @@
#!/bin/python
import time
import paho.mqtt.client as mqtt
import json
MQTT_SERVER_ADDRESS = "localhost"
MQTT_SERVER_PORT = 1883
MQTT_TOPIC = "homeassistant/device/mixer"
def on_connect(client, userdata, flags, reason_code, properties):
print(f"Connected: {reason_code}")
client.subscribe("#")
def on_message(client, userdata, msg):
print(f"{msg.topic}: {msg.payload.decode()}")
client = mqtt.Client(callback_api_version=mqtt.CallbackAPIVersion.VERSION2)
client.on_connect = on_connect
client.on_message = on_message
client.connect(MQTT_SERVER_ADDRESS, MQTT_SERVER_PORT, 60)
client.loop_start()
j = json.dumps({"command": "scene_recall", "args": {"scene_id": 42}})
# j = json.dumps({'command': 'shutdown'})
client.publish(MQTT_TOPIC, payload=j, qos=0, retain=False)
time.sleep(2)
client.loop_stop()
client.disconnect()

2
mixer.py
View file

@ -157,7 +157,7 @@ class Mixer:
self.sock.sendall(msg_bytes)
def scene_recall(self, scene_id):
print(f"scene_recall: scene_id={scene_id}")
print(f"scene_recall: scene_id={scene_id}", scene_id)
self.set_bank_1()
msb = mido.Message("program_change", channel=0, program=scene_id)

54
mqtt_client.py
View file

@ -1,54 +0,0 @@
#!/bin/python
import argparse
import json
import paho.mqtt.client as mqtt
import mixer
from mixer import Mixer
MQTT_SERVER_ADDRESS = "localhost"
MQTT_SERVER_PORT = 1883
MQTT_TOPIC = "homeassistant/device/mixer"
m: Mixer = None
def on_connect(client, userdata, flags, reason_code, properties):
print(f"Connected: {reason_code}")
client.subscribe("#")
def on_message(client, userdata, msg):
payload = msg.payload.decode()
print(f"{msg.topic}: {payload}")
j = json.loads(payload)
match j["command"]:
case "scene_recall":
m.scene_recall(j["args"]["scene_id"])
case "shutdown":
m.shutdown()
def main():
parser = argparse.ArgumentParser(description="Allen & Heath Qu MQTT Remote Control")
parser.add_argument("ip", help="IP of the mixer")
args = parser.parse_args()
print(f"IP: {args.ip}")
global m
m = mixer.Mixer(args.ip, mixer.MIXER_PORT)
client = mqtt.Client(callback_api_version=mqtt.CallbackAPIVersion.VERSION2)
client.on_connect = on_connect
client.on_message = on_message
client.connect(MQTT_SERVER_ADDRESS, MQTT_SERVER_PORT, 60)
client.loop_forever()
if __name__ == "__main__":
main()