oscpy 0.6.0

OSCPy

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之前的版本将不支持）
• 快速
• 易于使用
• 健壮（在消息格式不正确时返回有意义的错误，在正确消息上始终做正确的事，并默认拦截+记录回调抛出的异常）
• 关注点分离（消息解析与通信）
• 同步和异步兼容性（线程、asyncio、trio…）
• 代码简洁易读

功能

• 序列化和解析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.')

服务器（异步）(待办！)

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

Unicode

默认情况下，服务器和客户端将字节（编码后的字符串）作为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')

（此处添加了u字面量以提高清晰度）。

命令行界面

OSCPy提供了一个"oscli"实用工具，用于帮助调试

• oscli dump用于监听消息并将它们转储
• oscli send用于向服务器发送消息或捆绑包

有关更多信息，请参阅oscli -h

注意事项

• 序列化中不允许使用None值
• Unix类型的套接字在你监听它们时必须已存在

待办事项

• 对时间戳的实际支持（目前仅支持可选地丢弃过时的捆绑包，而不是延迟具有时间戳的捆绑包）
• 支持其他参数类型
• 异步IO导向的服务器实现
• 示例和文档

贡献

查看我们的贡献指南，并自由地改进OSCPy。

许可证

OSCPy根据MIT许可证条款发布。请参阅LICENSE.txt文件。