aiolo

liblo的Python绑定，用于POSIX系统上的开放音控（OSC）协议实现。

安装

安装liblo

OS X: brew install liblo

Ubuntu: apt-get install liblo7 liblo-dev

然后

pip install aiolo

示例

Python中许多美妙之处之一是支持操作符重载。aiolo热情拥抱这一点，为像Message、Bundle、Route和Sub这样的对象提供直观的编程体验。

简单的回声服务器

import asyncio

from aiolo import Address, Midi, Server


async def main():

    server = Server(port=12001)
    server.start()

    # Create endpoints

    # /foo accepts an int, a float, and a MIDI packet
    foo = server.route('/foo', [int, float, Midi])
    ex = server.route('/exit')

    address = Address(port=12001)

    for i in range(5):
        address.send(foo, i, float(i), Midi(i, i, i, i))

    # Notify subscriptions to exit in 1 sec
    address.delay(1, ex)

    # Subscribe to messages for any of the routes
    subs = foo.sub() | ex.sub()

    async for route, data in subs:
        print(f'echo_server: {str(route.path)} received {data}')
        if route == ex:
            await subs.unsub()

    server.stop()


if __name__ == '__main__':
    asyncio.get_event_loop().run_until_complete(main())

多播

import asyncio
import random

from aiolo import MultiCast, MultiCastAddress, Route, Server


async def main():
    # Create endpoints for receiving data
    foo = Route('/foo', str)
    ex = Route('/exit')

    # Create a multicast group
    multicast = MultiCast('224.0.1.1', port=15432)

    # Create a cluster of servers in the same multicast group
    cluster = []
    for i in range(10):
        server = Server(multicast=multicast)
        # Have them all handle the same route
        server.route(foo)
        server.route(ex)
        server.start()
        cluster.append(server)

    address = MultiCastAddress(server=random.choice(cluster))

    # Send a single message from any one server to the entire cluster.
    # The message will be received by each server.
    address.send(foo, 'hello cluster')

    # Notify subscriptions to exit in 1 sec
    address.delay(1, ex)

    # Listen for incoming strings at /foo on any server in the cluster
    subs = foo.sub() | ex.sub()
    async for route, data in subs:
        print(f'{route} got data: {data}')
        if route == ex:
            await subs.unsub()

    for server in cluster:
        server.stop()


if __name__ == '__main__':
    asyncio.get_event_loop().run_until_complete(main())

有关更多信息，请参阅示例和测试。

支持的平台

Travis CI测试，以下配置

• Ubuntu 18.04 Bionic Beaver + liblo 0.29 + [CPython3.6, CPython3.7, CPython3.8, PyPy7.3.0 (3.6.9)]
• OS X + liblo 0.29 + [CPython3.6, CPython3.7, CPython3.8, PyPy7.3.0 (3.6.9)]

贡献

欢迎提交拉取请求，请报告您遇到的任何问题。

变更日志

4.1.1 (2020-07-22)

• 通过传递zip_safe=False防止egg安装错误

4.1.0

• 纠正了一些__hash__问题。

4.0.0

• 使用基于Python的OSC地址模式匹配而不是liblo，支持转义特殊字符
• 确保ThreadedServer.start()等待线程初始化
• 修复了订阅者可能收不到挂起数据的问题
• 修复了在AioServer.stop()上未调用loop.remove_reader()的问题