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.aio provides a asyncio.Transport: SerialTransport.

A factory function (asyncio.coroutine) is provided:

serial.aio.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.aio.create_serial_connection(loop, Output, '/dev/ttyUSB0', baudrate=115200)
loop.run_until_complete(coro)
loop.run_forever()
loop.close()