oscpy Unduh - unduhan kode sumber oscpy

oscpy

Kategori lainnya

1.0.0

Unduh

OSCPy

CI dilakukan oleh Github Checks, lihat komit saat ini untuk status build.

Implementasi modern OSC untuk python2/3.

Apa itu OSC.

OpenSoundControl adalah protokol jaringan berbasis UDP, yang dirancang untuk pengiriman pesan sensitif terhadap waktu dengan cepat, seperti namanya, ini dirancang sebagai pengganti MIDI, tetapi dapat diterapkan dengan baik pada situasi lain. Protokolnya mudah digunakan, alamat OSC terlihat seperti URL http, dan menerima berbagai tipe dasar, seperti string, float, int, dll. Pada dasarnya Anda dapat menganggapnya sebagai http POST, dengan overhead yang lebih sedikit.

Anda dapat mempelajari lebih lanjut tentang OSC di OpenSoundControl.org

Sasaran

kompatibilitas python2.7/3.6+ (dapat lebih santai di sisi python3 jika diperlukan, tetapi tidak ada yang sebelum 2.7 yang akan didukung)
cepat
mudah digunakan
kuat (mengembalikan kesalahan yang berarti jika ada pesan yang salah format, selalu melakukan hal yang benar pada pesan yang benar, dan secara default mencegat+mencatat pengecualian yang ditimbulkan oleh panggilan balik)
pemisahan kekhawatiran (penguraian pesan vs komunikasi)
kompatibilitas sinkronisasi dan async (utas, asyncio, trio…)
kode yang bersih dan mudah dibaca

Fitur

membuat serial dan mengurai tipe data/Pesan/Paket OSC
server udp berbasis thread untuk membuka soket (INET atau UNIX) dan mengikat panggilan balik pada alamat osc pada soket tersebut
klien sederhana

Memasang

pip install oscpy

Penggunaan

Server (utas)

 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

atau Anda dapat menggunakan API dekorator.

Server (utas)

 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 juga bersifat client, dalam artian dapat mengirim pesan dan menjawab pesan dari server lain

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

Klien

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

Unikode

Secara default, server dan klien mengambil byte (string yang disandikan), bukan string unicode, untuk alamat osc dan juga string osc. Namun, Anda dapat meneruskan parameter encoding agar string Anda secara otomatis dikodekan dan didekodekan oleh parameter tersebut, sehingga panggilan balik Anda akan mendapatkan string unicode (unicode di python2, str di 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 literal ditambahkan di sini untuk kejelasan).

CLI

OSCPy menyediakan utilitas "oscli", untuk membantu proses debug:

oscli dump untuk mendengarkan pesan dan membuangnya
oscli send untuk mengirim pesan atau bundel ke server

Lihat oscli -h untuk informasi lebih lanjut.

GOTCHAS

None nilai yang tidak diperbolehkan dalam serialisasi
Soket tipe Unix tidak boleh sudah ada saat Anda mendengarkan() pada soket tersebut

TODO

dukungan nyata untuk penanda waktu (saat ini hanya mendukung pelepasan bundel yang terlambat secara opsional, tidak menunda bundel yang memiliki penanda waktu di masa mendatang)
dukungan untuk tipe argumen tambahan
implementasi server berorientasi asyncio
contoh & dokumentasi