Concurrency agnostic socket API
Project description
connio
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)
Installation
From within your favorite python environment:
pip install connio
Usage
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())
Project details
Release history Release notifications | RSS feed
Download files
Download the file for your platform. If you're not sure which to choose, learn more about installing packages.
Source Distribution
connio-0.2.0.tar.gz
(2.8 kB
view hashes)