pySerial-asyncio API

asyncio

Warning

This implementation is currently in an experimental state. Use at your own risk.

Experimental asyncio support is available for Python 3.4 and newer. The module serial_asyncio provides a asyncio.Transport: SerialTransport.

A factory function (asyncio.coroutine) is provided:

serial_asyncio.create_serial_connection(loop, protocol_factory, *args, **kwargs)
Parameters:
  • loop – The event handler
  • protocol_factory – Factory function for a asyncio.Protocol
  • args – Passed to the serial.Serial init function
  • kwargs – Passed to the serial.Serial init function
Platform:

Posix

Get a connection making coroutine.

Example:

class Output(asyncio.Protocol):
    def connection_made(self, transport):
        self.transport = transport
        print('port opened', transport)
        transport.serial.rts = False
        transport.write(b'hello world\n')

    def data_received(self, data):
        print('data received', repr(data))
        self.transport.close()

    def connection_lost(self, exc):
        print('port closed')
        asyncio.get_event_loop().stop()

loop = asyncio.get_event_loop()
coro = serial_asyncio.create_serial_connection(loop, Output, '/dev/ttyUSB0', baudrate=115200)
loop.run_until_complete(coro)
loop.run_forever()
loop.close()