oscpy下载 - oscpy源码下载

oscpy

其他类别

1.0.0

下载

奥氏体吡啶

CI 由 Github Checks 完成，查看当前提交的构建状态。

python2/3 OSC 的现代实现。

什么是OSC。

OpenSoundControl 是一种基于 UDP 的网络协议，专为快速调度时间敏感消息而设计，顾名思义，它被设计为 MIDI 的替代品，但也适用于其他情况。该协议使用简单，OSC地址看起来像http URL，并接受各种基本类型，例如字符串、浮点数、整数等。您可以将其基本上视为http POST，开销较小。

您可以在 OpenSoundControl.org 上了解有关 OSC 的更多信息

目标

python2.7/3.6+兼容性（如果需要的话，python3端可以放宽一些，但2.7之前不支持）
快速地
便于使用
健壮（在出现格式错误的消息时返回有意义的错误，始终对正确的消息执行正确的操作，并且默认情况下拦截+记录回调引发的异常）
关注点分离（消息解析与通信）
同步和异步兼容性（线程、异步、三重奏……）
干净且易于阅读的代码

特征

序列化和解析 OSC 数据类型/消息/包
基于线程的 udp 服务器，用于打开套接字（INET 或 UNIX）并在其上的 osc 地址上绑定回调
一个简单的客户端

安装

pip install oscpy

用法

服务器（线程）

 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

或者您可以使用装饰器 API。

服务器（线程）

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

服务器也是客户端，从某种意义上说，它们可以发送消息并回答来自其他服务器的消息

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

服务器（异步）（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 ))

客户

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

统一码

默认情况下，服务器和客户端对于 osc 地址和 osc 字符串采用字节（编码字符串），而不是 unicode 字符串。但是，您可以传递一个encoding参数，让字符串自动编码和解码，这样您的回调将获得 unicode 字符串（python2 中为 unicode，python3 中为 str）。

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