nikita/lib-py-osc
1
0
Fork 0
Code Issues 1 Pull Requests Actions Packages Projects Releases Wiki Activity

Client improvements

Browse Source 
* Added broadcast support
* Added _DEBUG switch to switch logs state
This commit is contained in:
Nikita (from test rpi)
2025-11-10 08:04:59 -05:00
parent 7807c6b2ad
commit 4cfe79f474
1 changed files with 19 additions and 13 deletions
Download Patch File Download Diff File Expand all files Collapse all files

32
client.py
View File

@@ -3,6 +3,7 @@
import asyncio
import struct
_DEBUG = True
_TASK = None
_Q = asyncio.Queue()
@@ -16,7 +17,8 @@ def __string_to_osc(s: str) -> bytes:
async def __serve() -> None:
''' Task that manages OSC sending '''
# log
print('[I] OSC: task is running')
if _DEBUG:
print('[I] OSC Client: task is running')
# task loop
while True:
# get value from queue
@@ -45,12 +47,13 @@ async def __serve() -> None:
osc_data += __string_to_osc(d)
# unsupported
else:
print('[!] OSC: 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)
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
@@ -64,20 +67,23 @@ async def __serve() -> None:
# send the packet
trans, prot = await asyncio.get_running_loop().create_datagram_endpoint(
asyncio.DatagramProtocol,
remote_addr=(host, port)
remote_addr=(host, port),
allow_broadcast=True
)
try:
trans.sendto(packet)
print('[I] OSC: sent data to %s:%s' % (host, port))
if _DEBUG: print('[I] OSC Client: sent data to %s:%s' % (host, port))
except:
print('[!] OSC: failed to send to %s:%s!' % (host, port))
print(' * address: %s' % addr)
print(' * data: %s' % data)
if _DEBUG:
print('[!] OSC Client: failed to send to %s:%s!' % (host, port))
print(' * address: %s' % addr)
print(' * data: %s' % data)
finally:
trans.close()
# log
print('[I] OSC: task is stopped')
if _DEBUG:
print('[I] OSC Client: task is stopped')
async def start() -> None:
''' Starts OSC task '''