A python concurrency agnostic communication library.

Pass a URL to the single point API function connio.connection_for_url() and it will return a communication object with a common generic API.

Helpful when dealing with instrumentation which work over serial line or TCP (and in future USB) with simple REQ-REP communication protocols (example: SCPI).

The request for a communication object is forwarded to the corresponding serialio or sockio libraries depending on the URL you give.

Written in asyncio with support for different concurrency models:

• asyncio
• classic blocking API
• future based API
• python 2 compatible blocking API (for those pour souls stuck with python 2)

From within your favorite python environment:

pip install connio

import asyncio
from connio import connection_for_url

async def main():

    # A local async serial line
    sl = connection_for_url("serial:///dev/ttyS0", parity="E")
    print(await sl.write_readline(b"*IDN?\n"))

    # An async serial line over telnet server
    sl = connection_for_url("rfc2217://moxa.acme.org:7890", parity="E")
    print(await sl.write_readline(b"*IDN?\n"))

    # An async TCP connection
    tcp = connection_for_url("tcp://gepace.acme.org:5025")
    print(await tcp.write_readline(b"*IDN?\n"))

    # An sync TCP connection
    tcp = connection_for_url("tcp://gepace.acme.org:5025", concurrency="sync")
    print(tcp.write_readline(b"*IDN?\n"))

asyncio.run(main())