"""Build OSC messages for client applications.""" from pythonosc import osc_message from pythonosc.parsing import osc_types from typing import List, Tuple, Union, Any class BuildError(Exception): """Error raised when an incomplete message is trying to be built.""" class OscMessageBuilder(object): """Builds arbitrary OscMessage instances.""" ARG_TYPE_FLOAT = "f" ARG_TYPE_DOUBLE = "d" ARG_TYPE_INT = "i" ARG_TYPE_STRING = "s" ARG_TYPE_BLOB = "b" ARG_TYPE_RGBA = "r" ARG_TYPE_MIDI = "m" ARG_TYPE_TRUE = "T" ARG_TYPE_FALSE = "F" ARG_TYPE_ARRAY_START = "[" ARG_TYPE_ARRAY_STOP = "]" _SUPPORTED_ARG_TYPES = ( ARG_TYPE_FLOAT, ARG_TYPE_DOUBLE, ARG_TYPE_INT, ARG_TYPE_BLOB, ARG_TYPE_STRING, ARG_TYPE_RGBA, ARG_TYPE_MIDI, ARG_TYPE_TRUE, ARG_TYPE_FALSE) def __init__(self, address: str=None) -> None: """Initialize a new builder for a message. Args: - address: The osc address to send this message to. """ self._address = address self._args = [] @property def address(self) -> str: """Returns the OSC address this message will be sent to.""" return self._address @address.setter def address(self, value: str) -> None: """Sets the OSC address this message will be sent to.""" self._address = value @property def args(self) -> List[Tuple[str, Union[str, bytes, bool, int, float, tuple, list]]]: # TODO: Make 'tuple' more specific for it is a MIDI packet """Returns the (type, value) arguments list of this message.""" return self._args def _valid_type(self, arg_type: str) -> bool: if arg_type in self._SUPPORTED_ARG_TYPES: return True elif isinstance(arg_type, list): for sub_type in arg_type: if not self._valid_type(sub_type): return False return True return False def add_arg(self, arg_value: Union[str, bytes, bool, int, float, tuple, list], arg_type: str=None) -> None: # TODO: Make 'tuple' more specific for it is a MIDI packet """Add a typed argument to this message. Args: - arg_value: The corresponding value for the argument. - arg_type: A value in ARG_TYPE_* defined in this class, if none then the type will be guessed. Raises: - ValueError: if the type is not supported. """ if arg_type and not self._valid_type(arg_type): raise ValueError( 'arg_type must be one of {}, or an array of valid types' .format(self._SUPPORTED_ARG_TYPES)) if not arg_type: arg_type = self._get_arg_type(arg_value) if isinstance(arg_type, list): self._args.append((self.ARG_TYPE_ARRAY_START, None)) for v, t in zip(arg_value, arg_type): self.add_arg(v, t) self._args.append((self.ARG_TYPE_ARRAY_STOP, None)) else: self._args.append((arg_type, arg_value)) def _get_arg_type(self, arg_value: Union[str, bytes, bool, int, float, tuple, list]) -> str: # TODO: Make 'tuple' more specific for it is a MIDI packet """Guess the type of a value. Args: - arg_value: The value to guess the type of. Raises: - ValueError: if the type is not supported. """ if isinstance(arg_value, str): arg_type = self.ARG_TYPE_STRING elif isinstance(arg_value, bytes): arg_type = self.ARG_TYPE_BLOB elif arg_value is True: arg_type = self.ARG_TYPE_TRUE elif arg_value is False: arg_type = self.ARG_TYPE_FALSE elif isinstance(arg_value, int): arg_type = self.ARG_TYPE_INT elif isinstance(arg_value, float): arg_type = self.ARG_TYPE_FLOAT elif isinstance(arg_value, tuple) and len(arg_value) == 4: arg_type = self.ARG_TYPE_MIDI elif isinstance(arg_value, list): arg_type = [self._get_arg_type(v) for v in arg_value] else: raise ValueError('Infered arg_value type is not supported') return arg_type def build(self) -> osc_message.OscMessage: """Builds an OscMessage from the current state of this builder. Raises: - BuildError: if the message could not be build or if the address was empty. Returns: - an osc_message.OscMessage instance. """ if not self._address: raise BuildError('OSC addresses cannot be empty') dgram = b'' try: # Write the address. dgram += osc_types.write_string(self._address) if not self._args: dgram += osc_types.write_string(',') return osc_message.OscMessage(dgram) # Write the parameters. arg_types = "".join([arg[0] for arg in self._args]) dgram += osc_types.write_string(',' + arg_types) for arg_type, value in self._args: if arg_type == self.ARG_TYPE_STRING: dgram += osc_types.write_string(value) elif arg_type == self.ARG_TYPE_INT: dgram += osc_types.write_int(value) elif arg_type == self.ARG_TYPE_FLOAT: dgram += osc_types.write_float(value) elif arg_type == self.ARG_TYPE_DOUBLE: dgram += osc_types.write_double(value) elif arg_type == self.ARG_TYPE_BLOB: dgram += osc_types.write_blob(value) elif arg_type == self.ARG_TYPE_RGBA: dgram += osc_types.write_rgba(value) elif arg_type == self.ARG_TYPE_MIDI: dgram += osc_types.write_midi(value) elif arg_type in (self.ARG_TYPE_TRUE, self.ARG_TYPE_FALSE, self.ARG_TYPE_ARRAY_START, self.ARG_TYPE_ARRAY_STOP): continue else: raise BuildError('Incorrect parameter type found {}'.format( arg_type)) return osc_message.OscMessage(dgram) except osc_types.BuildError as be: raise BuildError('Could not build the message: {}'.format(be))