python-osc/pythonosc/osc_message_builder.py at main · ZipFile/python-osc · GitHub

History

215 lines (188 loc) · 7.7 KB

Raw

100

101

102

103

104

105

106

107

108

109

110

111

112

113

114

115

116

117

118

119

120

121

122

123

124

125

126

127

128

129

130

131

132

133

134

135

136

137

138

139

140

141

142

143

144

145

146

147

148

149

150

151

152

153

154

155

156

157

158

159

160

161

162

163

164

165

166

167

168

169

170

171

172

173

174

175

176

177

178

179

180

181

182

183

184

185

186

187

188

189

190

191

192

193

194

195

196

197

198

199

200

201

202

203

204

205

206

207

208

209

210

211

212

213

214

215

"""Build OSC messages for client applications."""

from typing import Any, Iterable, List, Optional, Tuple, Union

from pythonosc import osc_message

from pythonosc.parsing import osc_types

# Represents a single OSC argument value.

# Can be a primitive type, a MIDI packet, or a list/tuple for nested OSC arrays.

ArgValue = Union[

str, bytes, bool, int, float, osc_types.MidiPacket, List[Any], Tuple[Any, ...], None

]

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_INT64 = "h"

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_NIL = "N"

ARG_TYPE_ARRAY_START = "["

ARG_TYPE_ARRAY_STOP = "]"

_SUPPORTED_ARG_TYPES = (

ARG_TYPE_FLOAT,

ARG_TYPE_DOUBLE,

ARG_TYPE_INT,

ARG_TYPE_INT64,

ARG_TYPE_BLOB,

ARG_TYPE_STRING,

ARG_TYPE_RGBA,

ARG_TYPE_MIDI,

ARG_TYPE_TRUE,

ARG_TYPE_FALSE,

ARG_TYPE_NIL,

)

def __init__(self, address: Optional[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 = [] # type: List[Tuple[str, Union[ArgValue, None]]]

@property

def address(self) -> Optional[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[ArgValue, None]]]:

"""Returns the (type, value) arguments list of this message."""

return self._args

def _valid_type(self, arg_type: Union[str, List[Any]]) -> bool:

if isinstance(arg_type, str):

return arg_type in self._SUPPORTED_ARG_TYPES

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: ArgValue, arg_type: Optional[Union[str, List[Any]]] = None

) -> None:

"""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(

f"arg_type must be one of {self._SUPPORTED_ARG_TYPES}, or an array of valid 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): # type: ignore[arg-type]

self.add_arg(v, t)

self._args.append((self.ARG_TYPE_ARRAY_STOP, None))

else:

self._args.append((arg_type, arg_value))

# The return type here is actually Union[str, List[<self>]], however there

# is no annotation for a recursive type like this.

def _get_arg_type(self, arg_value: ArgValue) -> Union[str, Any]:

"""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 # type: Union[str, Any]

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):

if arg_value.bit_length() > 31:

arg_type = self.ARG_TYPE_INT64

else:

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]

elif arg_value is None:

arg_type = self.ARG_TYPE_NIL

else:

raise ValueError("Inferred 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(f",{arg_types}")

for arg_type, value in self._args:

if arg_type == self.ARG_TYPE_STRING:

dgram += osc_types.write_string(value) # type: ignore[arg-type]

elif arg_type == self.ARG_TYPE_INT:

dgram += osc_types.write_int(value) # type: ignore[arg-type]

elif arg_type == self.ARG_TYPE_INT64:

dgram += osc_types.write_int64(value) # type: ignore[arg-type]

elif arg_type == self.ARG_TYPE_FLOAT:

dgram += osc_types.write_float(value) # type: ignore[arg-type]

elif arg_type == self.ARG_TYPE_DOUBLE:

dgram += osc_types.write_double(value) # type: ignore[arg-type]

elif arg_type == self.ARG_TYPE_BLOB:

dgram += osc_types.write_blob(value) # type: ignore[arg-type]

elif arg_type == self.ARG_TYPE_RGBA:

dgram += osc_types.write_rgba(value) # type: ignore[arg-type]

elif arg_type == self.ARG_TYPE_MIDI:

dgram += osc_types.write_midi(value) # type: ignore[arg-type]

elif arg_type in (

self.ARG_TYPE_TRUE,

self.ARG_TYPE_FALSE,

self.ARG_TYPE_ARRAY_START,

self.ARG_TYPE_ARRAY_STOP,

self.ARG_TYPE_NIL,

continue

else:

raise BuildError(f"Incorrect parameter type found {arg_type}")

return osc_message.OscMessage(dgram)

except osc_types.BuildError as be:

raise BuildError(f"Could not build the message: {be}")

def build_msg(

address: str, value: Union[ArgValue, Iterable[ArgValue]] = ""

) -> osc_message.OscMessage:

builder = OscMessageBuilder(address=address)

values: Iterable[Any]

if value == "":

values = []

elif not isinstance(value, Iterable) or isinstance(value, (str, bytes)):

values = [value]

else:

values = value

for val in values:

builder.add_arg(val)

return builder.build()

python-osc/pythonosc/osc_message_builder.py at main · ZipFile/python-osc · GitHub

Expand file tree

Breadcrumbs

osc_message_builder.py

Latest commit

History

Breadcrumbs

osc_message_builder.py

File metadata and controls