oscpy Télécharger - oscpy Téléchargement du code source

oscpy

Autres catégories

1.0.0

Télécharger

OSCPy

CI est effectué par Github Checks, voir la validation actuelle pour l'état de la construction.

Une implémentation moderne d'OSC pour python2/3.

Qu'est-ce que l'OSC.

OpenSoundControl est un protocole réseau basé sur UDP, conçu pour l'envoi rapide de messages sensibles au temps. Comme son nom l'indique, il a été conçu pour remplacer le MIDI, mais s'applique bien à d'autres situations. Le protocole est simple à utiliser, les adresses OSC ressemblent à des URL http et acceptent divers types de base, tels que string, float, int, etc. Vous pouvez le considérer essentiellement comme un http POST, avec moins de surcharge.

Vous pouvez en savoir plus sur OSC sur OpenSoundControl.org

Objectifs

Compatibilité python2.7/3.6+ (peut être davantage assouplie du côté python3 si nécessaire, mais rien avant la version 2.7 ne sera pris en charge)
rapide
facile à utiliser
robuste (renvoie des erreurs significatives en cas de messages mal formés, fait toujours ce qu'il faut sur les messages corrects et, par défaut, intercepte + enregistre les exceptions déclenchées par les rappels)
séparation des préoccupations (analyse des messages vs communication)
compatibilité sync et async (threads, asyncio, trio…)
code propre et facile à lire

Caractéristiques

sérialiser et analyser les types de données OSC/Messages/Bundles
un serveur udp basé sur des threads pour ouvrir des sockets (INET ou UNIX) et y lier des rappels sur des adresses osc
un simple client

Installer

pip install oscpy

Usage

Serveur (thread)

 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

ou vous pouvez utiliser l'API du décorateur.

Serveur (thread)

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

Les serveurs sont également clients, dans le sens où ils peuvent envoyer des messages et répondre aux messages d'autres serveurs.

 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.' )

Serveur (asynchrone) (À FAIRE !)

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

Client

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

Unicode

Par défaut, le serveur et le client prennent des octets (chaînes codées), et non des chaînes Unicode, pour les adresses osc ainsi que pour les chaînes osc. Cependant, vous pouvez transmettre un paramètre encoding pour que vos chaînes soient automatiquement codées et décodées par elles, afin que vos rappels obtiennent des chaînes 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' )

( u littéraux ajoutés ici pour plus de clarté).

CLI

OSCPy fournit un utilitaire « oscli » pour faciliter le débogage :

oscli dump pour écouter les messages et les vider
oscli send pour envoyer des messages ou des bundles à un serveur

Voir oscli -h pour plus d'informations.

OBTENUS

None valeur n'est autorisée dans la sérialisation
Les sockets de type Unix ne doivent pas déjà exister lorsque vous écoutez() dessus

FAIRE

prise en charge réelle de l'horodatage (actuellement, ne prend en charge que la suppression facultative des lots tardifs, sans retarder ceux avec des horodatages à l'avenir)
prise en charge de types d'arguments supplémentaires
une implémentation de serveur orientée asyncio
exemples et documentation