oscpy Download - oscpy Quellcode herunterladen

oscpy

Andere Kategorien

1.0.0

Herunterladen

OSCPy

CI wird von Github Checks durchgeführt. Den Build-Status finden Sie im aktuellen Commit.

Eine moderne Implementierung von OSC für Python2/3.

Was ist OSC?

OpenSoundControl ist ein UDP-basiertes Netzwerkprotokoll, das für den schnellen Versand zeitkritischer Nachrichten konzipiert ist. Wie der Name schon sagt, wurde es als Ersatz für MIDI entwickelt, lässt sich aber auch gut für andere Situationen verwenden. Das Protokoll ist einfach zu verwenden, OSC-Adressen sehen aus wie http-URLs und akzeptieren verschiedene Grundtypen wie String, Float, Int usw. Sie können es sich im Grunde als http-POST mit weniger Overhead vorstellen.

Mehr über OSC erfahren Sie auf OpenSoundControl.org

Ziele

Kompatibilität mit Python2.7/3.6+ (kann auf der Python3-Seite bei Bedarf weiter gelockert werden, aber nichts vor 2.7 wird unterstützt)
schnell
einfach zu bedienen
Robust (gibt bei fehlerhaften Nachrichten aussagekräftige Fehler zurück, macht bei korrekten Nachrichten immer das Richtige und fängt standardmäßig die durch Rückrufe ausgelösten Ausnahmen ab und protokolliert sie)
Trennung von Anliegen (Nachrichtenparsing vs. Kommunikation)
Sync- und Async-Kompatibilität (Threads, Asyncio, Trio…)
sauberer und leicht lesbarer Code

Merkmale

OSC-Datentypen/Nachrichten/Bundles serialisieren und analysieren
Ein Thread-basierter UDP-Server zum Öffnen von Sockets (INET oder UNIX) und zum Binden von Rückrufen an OSC-Adressen darauf
ein einfacher Kunde

Installieren

pip install oscpy

Verwendung

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

Oder Sie können die Decorator-API verwenden.

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

Server sind auch Clients in dem Sinne, dass sie Nachrichten senden und auf Nachrichten von anderen Servern antworten können

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

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

Kunde

 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

Standardmäßig verwenden Server und Client Bytes (codierte Zeichenfolgen) und keine Unicode-Zeichenfolgen sowohl für OSC-Adressen als auch für OSC-Zeichenfolgen. Sie können jedoch einen encoding übergeben, damit Ihre Zeichenfolgen automatisch von ihnen kodiert und dekodiert werden, sodass Ihre Rückrufe Unicode-Zeichenfolgen erhalten (Unicode in Python2, Str in 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 -Literale wurden hier der Übersichtlichkeit halber hinzugefügt).

CLI

OSCPy stellt ein „oscli“-Dienstprogramm zur Verfügung, das beim Debuggen hilft:

oscli dump um auf Nachrichten zu warten und diese auszugeben
oscli send , um Nachrichten oder Bundles an einen Server zu senden

Weitere Informationen finden Sie oscli -h .

Fallstricke

None Werte sind bei der Serialisierung nicht zulässig
Unix-Sockets dürfen nicht bereits vorhanden sein, wenn Sie listen() darauf verwenden

TODO

echte Unterstützung für Timetag (unterstützt derzeit nur das optionale Löschen verspäteter Bundles, nicht die Verzögerung derjenigen mit Timetags in der Zukunft)
Unterstützung für zusätzliche Argumenttypen
eine asynchron-orientierte Serverimplementierung
Beispiele und Dokumentation