Factorization

* Added packet.py which can serialize OSC packets

client.py

@@ -3,6 +3,9 @@
 import asyncio
 import struct
 
+from pytyuosc import packet
+#import packet
+
 _DEBUG = True
 _TASK = None
 _Q = asyncio.Queue()
@@ -31,38 +34,13 @@ async def __serve() -> None:
         port = val['port']
         addr = val['addr']
         data = val['data']
-        # OSC type tag string
-        type_tag = ','
-        # OSC data to append to the final packet
-        osc_data = bytes()
-        # check if data is correct
-        for d in data:
-            d_type = type(d)
-            # process the value
-            if d_type is int:
-                type_tag += 'i'
-                osc_data += struct.pack('>i', d)
-            elif d_type is str:
-                type_tag += 's'
-                osc_data += __string_to_osc(d)
-            # unsupported
-            else:
-                if _DEBUG:
-                    print('[!] OSC Client: unsupported data type was provided! The packet is discarded.')
-                    print('      * host: %s' % host)
-                    print('      * port: %s' % port)
-                    print('      * address: %s' % addr)
-                    print('      * data: %s' % data)
-                    print('         (bad type is %s)' % d_type)
-                type_tag = None
-                break
-        # no type tag
-        if type_tag is None:
+        
+        try:
+            data_to_send = packet.serialize(addr, data, _DEBUG)
+            if data_to_send is None:
+                raise Exception()
+        except:
             continue
-        # convert type tag to OSC-string
-        type_tag = __string_to_osc(type_tag)
-        # create OSC packet
-        packet = __string_to_osc(addr) + type_tag + osc_data
 
         # send the packet
         trans, prot = await asyncio.get_running_loop().create_datagram_endpoint(
@@ -71,7 +49,7 @@ async def __serve() -> None:
             allow_broadcast=True
         )
         try:
-            trans.sendto(packet)
+            trans.sendto(data_to_send)
             if _DEBUG: print('[I] OSC Client: sent data to %s:%s' % (host, port))
         except:
             if _DEBUG:
@@ -118,3 +96,16 @@ async def send(host: str, port: str, addr: str, values: list) -> None:
         'data': values
     }
     await _Q.put(d)
+
+#
+# TEST
+#
+async def __test__() -> None:
+    print('Going to send data to 255.255.255.255:8000')
+    await start()
+    await send('255.255.255.255', 8000, '/pytyuosc/osc/testing/is/in/progress', ['a', 'b', 1, 2, 3.4, 5.6])
+    await stop()
+    print('Data must have been sent')
+
+if __name__ == '__main__':
+    asyncio.run(__test__())

packet.py

@@ -0,0 +1,50 @@
+''' OSC packet serializer/parser '''
+
+import struct
+import traceback
+
+def __string_to_osc(s: str) -> bytes:
+    ''' Convert python string to OSC-string '''
+    data = s.encode('ascii')
+    to_append = 4 - (len(data) % 4)
+    data += b'\0' * to_append
+    return data
+
+def serialize(addr: str, data: list | None = None, log_error : bool = True) -> bytearray | None:
+    ''' Serializes OSC data into bytearray. Refer to README.md for more info. '''
+    try:
+        # no args
+        if data is None:
+            data = []
+        type_tag = ','
+        osc_data = bytes()
+        for d in data:
+            d_type = type(d)
+            # int
+            if d_type is int:
+                type_tag += 'i'
+                osc_data += struct.pack('>i', d)
+            # float
+            elif d_type is float:
+                type_tag += 'f'
+                osc_data += struct.pack('>f', d)
+            # string
+            elif d_type is str:
+                type_tag += 's'
+                osc_data += __string_to_osc(d)
+            # unsuported
+            else:
+                if log_error:
+                    print('[!] OSC packet serialize: unsupported data type was provided in \'data\'!')
+                    print('      * address: %s' % addr)
+                    print('      * data: %s' % data)
+                    print('         (bad type is %s)' % d_type)
+                return None
+        # convert type tag to OSC-string
+        type_tag = __string_to_osc(type_tag)
+        # create OSC packet
+        return bytearray(__string_to_osc(addr) + type_tag + osc_data)
+    except:
+        if log_error:
+            traceback.print_exc()
+        return None

parser.py

@@ -95,4 +95,4 @@ def parse(body: bytearray) -> dict | None:
         'addr': addr[0],
         'type_tag': type_tag[0],
         'args': args
-    }
+    }