Download oscpy - download do código-fonte oscpy

oscpy

Outras categorias

1.0.0

Baixar

OSCPy

CI é feito por verificações do Github, consulte o commit atual para ver o status da construção.

Uma implementação moderna de OSC para python2/3.

O que é OSC.

OpenSoundControl é um protocolo de rede baseado em UDP, projetado para envio rápido de mensagens urgentes, como o nome sugere, foi projetado como um substituto para MIDI, mas se aplica bem a outras situações. O protocolo é simples de usar, os endereços OSC se parecem com URLs http e aceitam vários tipos básicos, como string, float, int, etc. Você pode pensar nele basicamente como um HTTP POST, com menos sobrecarga.

Você pode aprender mais sobre OSC em OpenSoundControl.org

Metas

Compatibilidade com python2.7/3.6+ (pode ser mais relaxado no lado python3 se necessário, mas nada antes do 2.7 será suportado)
rápido
fácil de usar
robusto (retorna erros significativos no caso de mensagens malformadas, sempre faz a coisa certa nas mensagens corretas e, por padrão, intercepta + registra as exceções levantadas pelos retornos de chamada)
separação de preocupações (análise de mensagem vs comunicação)
compatibilidade sincronizada e assíncrona (threads, asyncio, trio…)
código limpo e fácil de ler

Características

serializar e analisar tipos de dados/mensagens/pacotes OSC
um servidor udp baseado em thread para abrir soquetes (INET ou UNIX) e vincular retornos de chamada em endereços osc neles
um cliente simples

Instalar

pip install oscpy

Uso

Servidor (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 você pode usar a API do decorador.

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

Os servidores também são clientes, no sentido de que podem enviar mensagens e responder mensagens de outros 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 (assí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 ])

Unicode

Por padrão, o servidor e o cliente usam bytes (strings codificadas), não strings unicode, para endereços osc, bem como strings osc. No entanto, você pode passar um parâmetro encoding para que suas strings sejam automaticamente codificadas e decodificadas por eles, para que seus retornos de chamada obtenham strings unicode (unicode em python2, str em 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 literais adicionados aqui para maior clareza).

CLI

OSCPy fornece um utilitário "oscli" para ajudar na depuração:

oscli dump para ouvir mensagens e despejá-las
oscli send para enviar mensagens ou pacotes para um servidor

Consulte oscli -h para obter mais informações.

PEGADA

None valor não é permitido na serialização
Soquetes do tipo Unix ainda não devem existir quando você escuta() neles

PENDÊNCIA

suporte real para timetag (atualmente suporta apenas a eliminação opcional de pacotes atrasados, não atrasando aqueles com timetags no futuro)
suporte para tipos de argumentos adicionais
uma implementação de servidor orientada para assíncio
exemplos e documentação