oscpy

OSCPy

La CI se realiza mediante comprobaciones de Github; consulte la confirmación actual para conocer el estado de la compilación.

Una implementación moderna de OSC para python2/3.

¿Qué es OSC?

OpenSoundControl es un protocolo de red basado en UDP, que está diseñado para el envío rápido de mensajes urgentes. Como su nombre indica, fue diseñado como un reemplazo de MIDI, pero se aplica bien a otras situaciones. El protocolo es fácil de usar, las direcciones OSC parecen URL http y aceptan varios tipos básicos, como string, float, int, etc. Puedes considerarlo básicamente como un http POST, con menos gastos generales.

Puede obtener más información sobre OSC en OpenSoundControl.org

Objetivos

• Compatibilidad con python2.7/3.6+ (se puede relajar más en el lado de python3 si es necesario, pero no se admitirá nada anterior a 2.7)
• rápido
• fácil de usar
• robusto (devuelve errores significativos en caso de mensajes con formato incorrecto, siempre hace lo correcto en los mensajes correctos y, de forma predeterminada, intercepta y registra las excepciones generadas por las devoluciones de llamada)
• separación de preocupaciones (análisis de mensajes versus comunicación)
• compatibilidad sincronizada y asíncrona (threads, asyncio, trio…)
• código limpio y fácil de leer

Características

• serializar y analizar tipos de datos OSC/Mensajes/Paquetes
• un servidor udp basado en subprocesos para abrir sockets (INET o UNIX) y vincular devoluciones de llamada en direcciones osc en ellos
• un cliente sencillo

Instalar

pip install oscpy

Uso

Servidor (hilo)

 from oscpy . server import OSCThreadServer
from time import sleep

def callback ( * values ):
    print ( "got values: {}" . format ( values ))

osc = OSCThreadServer ()  # See sources for all the arguments

# You can also use an *nix socket path here
sock = osc . listen ( address = '0.0.0.0' , port = 8000 , default = True )
osc . bind ( b'/address' , callback )
sleep ( 1000 )
osc . stop ()  # Stop the default socket

osc . stop_all ()  # Stop all sockets

# Here the server is still alive, one might call osc.listen() again

osc . terminate_server ()  # Request the handler thread to stop looping

osc . join_server ()  # Wait for the handler thread to finish pending tasks and exit

o puedes usar la API del decorador.

Servidor (hilo)

 from oscpy . server import OSCThreadServer
from time import sleep

osc = OSCThreadServer ()
sock = osc . listen ( address = '0.0.0.0' , port = 8000 , default = True )

@ osc . address ( b'/address' )
def callback ( * values ):
    print ( "got values: {}" . format ( values ))

sleep ( 1000 )
osc . stop ()

Los servidores también son clientes, en el sentido de que pueden enviar mensajes y responder a mensajes de otros servidores.

 from oscpy . server import OSCThreadServer
from time import sleep

osc_1 = OSCThreadServer ()
osc_1 . listen ( default = True )

@ osc_1 . address ( b'/ping' )
def ping ( * values ):
    print ( "ping called" )
    if True in values :
        cont . append ( True )
    else :
        osc_1 . answer ( b'/pong' )

osc_2 = OSCThreadServer ()
osc_2 . listen ( default = True )

@ osc_2 . address ( b'/pong' )
def pong ( * values ):
    print ( "pong called" )
    osc_2 . answer ( b'/ping' , [ True ])

osc_2 . send_message ( b'/ping' , [], * osc_1 . getaddress ())

timeout = time () + 1
while not cont :
    if time () > timeout :
        raise OSError ( 'timeout while waiting for success message.' )

Servidor (asíncrono) (¡TODO!)

 from oscpy . server import OSCThreadServer

with OSCAsyncServer ( port = 8000 ) as OSC :
    for address , values in OSC . listen ():
       if address == b'/example' :
            print ( "got {} on /example" . format ( values ))
       else :
            print ( "unknown address {}" . format ( address ))

Cliente

 from oscpy . client import OSCClient

address = "127.0.0.1"
port = 8000

osc = OSCClient ( address , port )
for i in range ( 10 ):
    osc . send_message ( b'/ping' , [ i ])

Unicódigo

De forma predeterminada, el servidor y el cliente toman bytes (cadenas codificadas), no cadenas Unicode, tanto para direcciones osc como para cadenas osc. Sin embargo, puede pasar un parámetro encoding para que sus cadenas codifiquen y decodifiquen automáticamente, de modo que sus devoluciones de llamada obtengan cadenas Unicode (unicode en python2, str en python3).

 osc = OSCThreadServer ( encoding = 'utf8' )
osc . listen ( default = True )

values = []

@ osc . address ( u'/encoded' )
def encoded ( * val ):
    for v in val :
        assert not isinstance ( v , bytes )
    values . append ( val )

send_message (
    u'/encoded' ,
    [ u'hello world' , u'ééééé ààààà' ],
    * osc . getaddress (), encoding = 'utf8' )

(Se agregaron literales u aquí para mayor claridad).

CLI

OSCPy proporciona una utilidad "oscli" para ayudar con la depuración:

• oscli dump para escuchar mensajes y volcarlos
• oscli send para enviar mensajes o paquetes a un servidor

Consulte oscli -h para obtener más información.

PROBLEMAS

• None los valores no está permitido en la serialización.
• Los sockets de tipo Unix no deben existir cuando escuchas() en ellos

HACER

• soporte real para etiquetas de tiempo (actualmente solo admite, opcionalmente, eliminar paquetes tardíos, sin retrasar aquellos con etiquetas de tiempo en el futuro)
• soporte para tipos de argumentos adicionales
• una implementación de servidor orientada a asincio
• ejemplos y documentación

Contribuyendo

Consulte nuestra guía de contribuciones y siéntase libre de mejorar OSCPy.

Licencia

OSCPy se publica bajo los términos de la licencia MIT. Consulte el archivo LICENCIA.txt.