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

add optional filter for recv() function

Browse source
This commit is contained in:
coon 2025-09-23 21:48:28 +02:00
parent f639bb9d1b
commit 4a31f50b50
1 changed files with 11 additions and 3 deletions
Download patch file Download diff file Expand all files Collapse all files

14
mixer.py
View file

@ -27,7 +27,7 @@ class Mixer:
GET_METER_DATA_REQUEST = 0x12 GET_METER_DATA_REQUEST = 0x12
GET_METER_DATA_RESPONSE = 0x13 GET_METER_DATA_RESPONSE = 0x13
def recv(self): def recv(self, response_msg_filter: SysExMessageId = None):
p = mido.Parser() p = mido.Parser()
while True: while True:
@ -41,12 +41,20 @@ class Mixer:
if p.pending(): if p.pending():
msg = p.get_message() msg = p.get_message()
print(msg.type)
print(vars(msg)) print(vars(msg))
if hasattr(msg, "data"): if hasattr(msg, "data"):
print("Hex: ", " ".join(f"{b:02X}" for b in msg.data)) print("Hex: ", " ".join(f"{b:02X}" for b in msg.data))
if response_msg_filter is not None:
if msg.type != 'sysex':
continue
msg_id = int(msg.data[8])
if msg_id != response_msg_filter.value:
continue
return msg return msg
def get_system_state(self): def get_system_state(self):
@ -109,7 +117,7 @@ class Mixer:
print("Sent:", " ".join(f"{b:02X}" for b in msg_bytes)) print("Sent:", " ".join(f"{b:02X}" for b in msg_bytes))
self.sock.sendall(msg_bytes) self.sock.sendall(msg_bytes)
response = self.recv() response = self.recv(self.SysExMessageId.GET_NAME_FROM_QU_RESPONSE)
sysex_header = response.data[:8] sysex_header = response.data[:8]