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

mixer.py: ran ruff

Browse source
This commit is contained in:
coon 2025-09-24 21:38:34 +02:00
parent cce93ce2cb
commit dbdf3a226f
2 changed files with 50 additions and 41 deletions
Download patch file Download diff file Expand all files Collapse all files

76
mixer.py
View file

@ -8,6 +8,7 @@ from enum import Enum
MIXER_PORT = 51325
class Mixer:
def __init__(self, ip, port):
self.sock = socket.create_connection((ip, port))
@ -47,7 +48,7 @@ class Mixer:
print("Hex: ", " ".join(f"{b:02X}" for b in msg.data))
if response_msg_filter is not None:
if msg.type != 'sysex':
if msg.type != "sysex":
continue
msg_id = int(msg.data[8])
@ -63,7 +64,7 @@ class Mixer:
data = self.SYSEX_ALL_CALL + [msg_id, i_pad_flag]
msg = mido.Message('sysex', data=data)
msg = mido.Message("sysex", data=data)
msg_bytes = bytes(msg.bytes())
self.sock.sendall(msg_bytes)
@ -109,9 +110,9 @@ class Mixer:
data = self.SYSEX_HEADER + [int(0), msg_id, int(channel_no)]
if name_to_set:
data += list(name_to_set.encode('utf-8'))
data += list(name_to_set.encode("utf-8"))
msg = mido.Message('sysex', data=data)
msg = mido.Message("sysex", data=data)
msg_bytes = bytes(msg.bytes())
print("Sent:", " ".join(f"{b:02X}" for b in msg_bytes))
@ -131,61 +132,77 @@ class Mixer:
print(f"Channel Name: {channel_name}")
def nrpn_parameter_control(self, midi_ch, mixer_ch, id, va, vx):
a = mido.Message('control_change', channel=midi_ch, control=0x63, value=mixer_ch)
b = mido.Message('control_change', channel=midi_ch, control=0x62, value=id)
c = mido.Message('control_change', channel=midi_ch, control=0x06, value=va)
d = mido.Message('control_change', channel=midi_ch, control=0x26, value=vx)
a = mido.Message(
"control_change", channel=midi_ch, control=0x63, value=mixer_ch
)
b = mido.Message("control_change", channel=midi_ch, control=0x62, value=id)
c = mido.Message("control_change", channel=midi_ch, control=0x06, value=va)
d = mido.Message("control_change", channel=midi_ch, control=0x26, value=vx)
msg_bytes = bytes(a.bytes() + b.bytes() + c.bytes() + d.bytes())
print(' '.join(f"{b:02X}" for b in msg_bytes))
print(" ".join(f"{b:02X}" for b in msg_bytes))
self.sock.sendall(msg_bytes)
def shutdown(self):
self.nrpn_parameter_control(midi_ch=0, mixer_ch=0, id=0x5F, va=0x00, vx=0x00)
def set_bank_1(self):
msb = mido.Message('control_change', channel=0, control=0x00, value=0x00)
lsb = mido.Message('control_change', channel=0, control=0x20, value=0x00)
msb = mido.Message("control_change", channel=0, control=0x00, value=0x00)
lsb = mido.Message("control_change", channel=0, control=0x20, value=0x00)
msg_bytes = bytes(msb.bytes() + lsb.bytes())
print(' '.join(f"{b:02X}" for b in msg_bytes))
print(" ".join(f"{b:02X}" for b in msg_bytes))
self.sock.sendall(msg_bytes)
def scene_recall(self, 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)
msb = mido.Message("program_change", channel=0, program=scene_id)
msg_bytes = bytes(msb.bytes())
print(' '.join(f"{b:02X}" for b in msg_bytes))
print(" ".join(f"{b:02X}" for b in msg_bytes))
self.sock.sendall(msg_bytes)
def watch(self):
while True:
self.recv_sys_ex()
def main():
parser = argparse.ArgumentParser(description="Allen & Heath Qu Remote Control")
parser.add_argument("ip", help="IP of the mixer")
subparsers = parser.add_subparsers(dest="command", required=True, help="Available commands")
subparsers = parser.add_subparsers(
dest="command", required=True, help="Available commands"
)
channel_naming_parser = subparsers.add_parser("get_name_from_qu", help="Channel naming")
channel_naming_parser.add_argument("channel_id", type=int, choices=range(0, 16), help="Number of channel to get or set its name")
channel_naming_parser = subparsers.add_parser(
"get_name_from_qu", help="Channel naming"
)
channel_naming_parser.add_argument(
"channel_id",
type=int,
choices=range(0, 16),
help="Number of channel to get or set its name",
)
channel_naming_parser.add_argument("-n", "--name", help="Channel name to set")
subparsers.add_parser("get_system_state", help="Get system information")
subparsers.add_parser("shutdown", help="Shut down the mixer")
scene_parser = subparsers.add_parser("scene_recall", help="Recall a specific scene")
scene_parser.add_argument("scene_number", type=int, choices=range(0, 100), help="Scene number to recall")
scene_parser.add_argument(
"scene_number", type=int, choices=range(0, 100), help="Scene number to recall"
)
subparsers.add_parser("scene_recall_default", help="Set the default scene 0")
subparsers.add_parser("watch", help="Just receive data from mixer and print it to console")
subparsers.add_parser(
"watch", help="Just receive data from mixer and print it to console"
)
args = parser.parse_args()
@ -197,12 +214,19 @@ def main():
print(f"Args: {vars(args)}")
match args.command:
case 'get_name_from_qu': mixer.get_name_from_qu(args.channel_id, args.name)
case 'get_system_state': mixer.get_system_state()
case 'shutdown': mixer.shutdown()
case 'scene_recall': mixer.scene_recall(args.scene_number)
case 'scene_recall_default': mixer.scene_recall(0)
case 'watch': mixer.watch()
case "get_name_from_qu":
mixer.get_name_from_qu(args.channel_id, args.name)
case "get_system_state":
mixer.get_system_state()
case "shutdown":
mixer.shutdown()
case "scene_recall":
mixer.scene_recall(args.scene_number)
case "scene_recall_default":
mixer.scene_recall(0)
case "watch":
mixer.watch()
if __name__ == '__main__':
if __name__ == "__main__":
main()

15
requirements.txt
View file

@ -1,15 +0,0 @@
blessed==1.21.0
bpython==0.25
certifi==2025.8.3
charset-normalizer==3.4.3
curtsies==0.4.3
cwcwidth==0.1.10
greenlet==3.2.4
idna==3.10
mido==1.3.3
packaging==25.0
Pygments==2.19.2
pyxdg==0.28
requests==2.32.5
urllib3==2.5.0
wcwidth==0.2.13