Skip to main content

asyncio-friendly Python bindings for liblo

Project description


asyncio-friendly Python bindings for liblo, an implementation of the Open Sound Control (OSC) protocol for POSIX systems.



Install liblo:

OS X: brew install liblo

Ubuntu: apt-get install liblo7 liblo-dev


pip install aiolo


One of the many beautiful things in Python is support for operator overloading. aiolo embraces this enthusiastically to offer the would-be OSC hacker an intuitive programming experience for objects such as Message, Bundle, Route, and Sub.

Simple echo server

import asyncio

from aiolo import Address, Midi, Server

async def main():

    server = Server(port=12001)

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


if __name__ == '__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('', 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

    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:

if __name__ == '__main__':

For additional usage see the examples and tests.

Supported platforms

Travis CI tests with the following configurations:

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


Pull requests are welcome, please file any issues you encounter.


4.1.1 (2020-07-22)

  • Prevent egg installation errors by passing zip_safe=False


  • Rectify some __hash__ issues.


  • Use Python-based OSC address pattern matching rather than liblo's, supports escaped special characters
  • Ensure ThreadedServer.start() waits for thread to be initialized
  • Fix bug where subscribers might not receive pending data
  • Fix bug where loop.remove_reader() was not being called on AioServer.stop()

Project details

Download files

Download the file for your platform. If you're not sure which to choose, learn more about installing packages.

Source Distribution

aiolo-4.1.1.tar.gz (711.1 kB view hashes)

Uploaded source

Supported by

AWS AWS Cloud computing and Security Sponsor Datadog Datadog Monitoring Fastly Fastly CDN Google Google Download Analytics Microsoft Microsoft PSF Sponsor Pingdom Pingdom Monitoring Sentry Sentry Error logging StatusPage StatusPage Status page