From 6944ba51287ebb81bdb4ebc8f5a1082ebe2a4260 Mon Sep 17 00:00:00 2001 From: onyx-and-iris <75868496+onyx-and-iris@users.noreply.github.com> Date: Mon, 7 Nov 2022 15:22:59 +0000 Subject: [PATCH] minor refactors --- xair_api/shared.py | 2 +- xair_api/xair.py | 26 +++++++++++--------------- 2 files changed, 12 insertions(+), 16 deletions(-) diff --git a/xair_api/shared.py b/xair_api/shared.py index 32d8256..0aa67b6 100644 --- a/xair_api/shared.py +++ b/xair_api/shared.py @@ -5,7 +5,7 @@ from .meta import geq_prop from .util import _get_fader_val, _set_fader_val, lin_get, lin_set, log_get, log_set """ -Classes shared by /ch, /rtn, /rt/aux, /bus, /fxsend, /lr +Classes shared by /ch, /rtn, /rtn/aux, /bus, /fxsend, /lr """ diff --git a/xair_api/xair.py b/xair_api/xair.py index d6c6f3a..0c859c0 100644 --- a/xair_api/xair.py +++ b/xair_api/xair.py @@ -3,7 +3,7 @@ import logging import threading import time from pathlib import Path -from typing import Optional +from typing import Optional, Union try: import tomllib @@ -31,15 +31,12 @@ class OSCClientServer(BlockingOSCUDPServer): super().__init__(("", 0), dispatcher) self.xr_address = address - def send_message(self, address: str, value: str): + def send_message(self, address: str, vals: Optional[Union[str, list]]): builder = OscMessageBuilder(address=address) - if value is None: - values = list() - elif isinstance(value, list): - values = value - else: - values = [value] - for val in values: + vals = vals if vals is not None else [] + if not isinstance(vals, list): + vals = [vals] + for val in vals: builder.add_arg(val) msg = builder.build() self.socket.sendto(msg.dgram, self.xr_address) @@ -80,14 +77,13 @@ class XAirRemote(abc.ABC): def validate_connection(self): self.send("/xinfo") time.sleep(self._CONNECT_TIMEOUT) - if len(self.info_response) > 0: - print( - f"Successfully connected to {self.info_response[2]} at {self.info_response[0]}." - ) - else: - print( + if not self.info_response: + raise XAirRemoteError( "Error: Failed to setup OSC connection to mixer. Please check for correct ip address." ) + print( + f"Successfully connected to {self.info_response[2]} at {self.info_response[0]}." + ) def run_server(self): self.server.serve_forever()