Add 'm' 4-byte MIDI type tag

Author: levaitamas

pythonosc/osc_message.py

@@ -47,6 +47,8 @@ def _parse_datagram(self):
           val, index = osc_types.get_blob(self._dgram, index)
         elif param == "r":  # RGBA.
           val, index = osc_types.get_rgba(self._dgram, index)
+        elif param == "m":  # MIDI.
+          val, index = osc_types.get_midi(self._dgram, index)
         elif param == "t":  # osc time tag:
             val, index = osc_types.get_ttag(self._dgram, index)
         elif param == "T":  # True.

pythonosc/osc_message_builder.py

@@ -16,14 +16,16 @@ class OscMessageBuilder(object):
   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_INT, ARG_TYPE_BLOB, ARG_TYPE_STRING, ARG_TYPE_RGBA, ARG_TYPE_TRUE, ARG_TYPE_FALSE)
+      ARG_TYPE_FLOAT, 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=None):
     """Initialize a new builder for a message.
@@ -103,6 +105,8 @@ def _get_arg_type(self, arg_value):
       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:
@@ -143,6 +147,8 @@ def build(self):
           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,

pythonosc/parsing/osc_types.py

@@ -319,3 +319,44 @@ def get_rgba(dgram, start_index):
         start_index + _INT_DGRAM_LEN)
   except (struct.error, TypeError) as e:
     raise ParseError('Could not parse datagram %s' % e)
+
+
+def write_midi(val):
+  """Returns the datagram for the given MIDI message parameter value
+
+     A valid MIDI message: (port id, status byte, data1, data2).
+
+  Raises:
+    - BuildError if the MIDI message could not be converted.
+
+  """
+  try:
+    assert 4 == len(val)
+    value = sum((value & 0xFF) << 8 * (3-pos) for pos, value in enumerate(val))
+    return struct.pack('>I', value)
+  except (struct.error, AssertionError) as e:
+    raise BuildError('Wrong argument value passed: {}'.format(e))
+
+
+def get_midi(dgram, start_index):
+  """Get a MIDI message (port id, status byte, data1, data2) from the datagram.
+
+  Args:
+    dgram: A datagram packet.
+    start_index: An index where the MIDI message starts in the datagram.
+
+  Returns:
+    A tuple containing the MIDI message and the new end index.
+
+  Raises:
+    ParseError if the datagram could not be parsed.
+  """
+  try:
+    if len(dgram[start_index:]) < _INT_DGRAM_LEN:
+      raise ParseError('Datagram is too short')
+    val = struct.unpack('>I',
+                        dgram[start_index:start_index + _INT_DGRAM_LEN])[0]
+    midi_msg = tuple((val & 0xFF << 8 * i) >> 8 * i for i in range(3,-1, -1))
+    return (midi_msg, start_index + _INT_DGRAM_LEN)
+  except (struct.error, TypeError) as e:
+    raise ParseError('Could not parse datagram %s' % e)

pythonosc/test/parsing/test_osc_types.py

@@ -123,6 +123,45 @@ def test_datagram_too_short(self):
     self.assertRaises(osc_types.ParseError, osc_types.get_rgba, dgram, 2)
 
 
+class TestMidi(unittest.TestCase):  # TODO
+
+  def test_get_midi(self):
+    cases = {
+        b"\x00\x00\x00\x00": ((0,0,0,0), 4),
+        b"\x00\x00\x00\x02": ((0,0,0,1), 4),
+        b"\x00\x00\x00\x02": ((0,0,0,2), 4),
+        b"\x00\x00\x00\x03": ((0,0,0,3), 4),
+
+        b"\x00\x00\x01\x00": ((0,0,1,0), 4),
+        b"\x00\x01\x00\x00": ((0,1,0,0), 4),
+        b"\x01\x00\x00\x00": ((1,0,0,0), 4),
+
+        b"\x00\x00\x00\x01GARBAGE": ((0,0,0,1), 4),
+    }
+
+    for dgram, expected in cases.items():
+        self.assertEqual(
+            expected, osc_types.get_midi(dgram, 0))
+
+  def test_get_midi_raises_on_type_error(self):
+    cases = [b'', True]
+
+    for case in cases:
+      self.assertRaises(osc_types.ParseError, osc_types.get_midi, case, 0)
+
+  def test_get_midi_raises_on_wrong_start_index(self):
+    self.assertRaises(
+        osc_types.ParseError, osc_types.get_midi, b'\x00\x00\x00\x11', 1)
+
+  def test_get_midi_raises_on_wrong_start_index_negative(self):
+    self.assertRaises(
+        osc_types.ParseError, osc_types.get_midi, b'\x00\x00\x00\x00', -1)
+
+  def test_datagram_too_short(self):
+    dgram = b'\x00' * 3
+    self.assertRaises(osc_types.ParseError, osc_types.get_midi, dgram, 2)
+
+
 class TestDate(unittest.TestCase):
     def test_get_ttag(self):
         cases = {

pythonosc/test/test_osc_message_builder.py

@@ -42,13 +42,16 @@ def test_all_param_types(self):
     builder.add_arg(b"\x01\x02\x03", builder.ARG_TYPE_BLOB)
     builder.add_arg([1, ["abc"]], [builder.ARG_TYPE_INT, [builder.ARG_TYPE_STRING]])
     builder.add_arg(4278255360, builder.ARG_TYPE_RGBA)
-    self.assertEqual(len("fisTFb[i[s]]")*2+1, len(builder.args))
+    builder.add_arg((1,145, 36, 125), builder.ARG_TYPE_MIDI)
+    self.assertEqual(len("fisTFb[i[s]]")*2+2, len(builder.args))
     self.assertEqual("/SYNC", builder.address)
     builder.address = '/SEEK'
     msg = builder.build()
     self.assertEqual("/SEEK", msg.address)
     self.assertSequenceEqual(
-        [4.0, 2, "value", True, False, b"\x01\x02\x03", [1, ["abc"]]] * 2 + [4278255360], msg.params)
+        [4.0, 2, "value", True, False, b"\x01\x02\x03", [1, ["abc"]]] * 2 +
+        [4278255360, (1,145, 36, 125)],
+        msg.params)
 
   def test_long_list(self):
     huge_list = list(range(512))