DOCSEARCHbyAlgolia
Submit Docs

aioserial

repository·main·Indexed 40 days ago

https://github.com/johannjhang/aioserial.py

An asynchronous serial port library for Python (v1.3.1) that provides an asyncio-compatible interface by combining asyncio with pySerial. It offers a higher-level API than pyserial-asyncio, allowing developers to use await on read and write methods, and provides support for Windows.

Tokens
2.5K
Snippets
8
Records
9
Agent score
0%

What's inside aioserial

  1. Replace pyserial-asyncio with aioserial

    main

    If you are currently using pyserial-asyncio with its Protocol-based approach, you can switch to aioserial for a higher-level, more direct API. Instead of implementing asyncio.Protocol, you can simply use await on read and write methods. This is particularly useful if you need Windows support, which pyserial-asyncio lacks.

    import asyncio
import aioserial

async def read_and_print(aioserial_instance: aioserial.AioSerial):
    while True:
        data: bytes = await aioserial_instance.read_async()
        print(data.decode(errors='ignore'), end='', flush=True)
        if b'\n' in data:
            aioserial_instance.close()
            break

# Example: Concurrent read and write
aioserial_instance: aioserial.AioSerial = aioserial.AioSerial(port='/dev/ttyUSB0', baudrate=115200)
asyncio.run(asyncio.gather(
    read_and_print(aioserial_instance), 
    aioserial_instance.write_async(b'Hello, World!\n')
))

  2. Quick start: Simple serial port reader

    main

    To read data from a serial port asynchronously, instantiate aioserial.AioSerial with the desired port and use the read_async() method within an asyncio loop.

    import asyncio
import aioserial

async def read_and_print(aioserial_instance: aioserial.AioSerial):
    while True:
        print((await aioserial_instance.read_async()).decode(errors='ignore'), end='', flush=True)

asyncio.run(read_and_print(aioserial.AioSerial(port='COM1')))

  3. Async write methods in AioSerial

    main

    Use these methods to asynchronously send data to the serial port.

    # write_async: writes bytes-like data
number_of_byte_like_data_written: int = await aioserial_instance.write_async(bytes_like_data)

# writelines_async: writes a list of bytes-like data
number_of_byte_like_data_in_the_given_list_written: int = await aioserial_instance.writelines_async(list_of_bytes_like_data)

  4. Async read methods in AioSerial

    main

    Use these methods to asynchronously retrieve data from the serial port.

    # read_async: reads a specific number of bytes
bytes_read: bytes = await aioserial_instance.read_async(size: int = 1)

# read_until_async: reads until a specific byte is found
at_most_certain_size_of_bytes_read: bytes = await aioserial_instance.read_until_async(
    expected: bytes = aioserial.LF, 
    size: Optional[int] = None
)

# readinto_async: reads data into an existing buffer
number_of_byte_read: int = await aioserial_instance.readinto_async(b: Union[array.array, bytearray])

# readline_async: reads a single line
a_line_of_at_most_certain_size_of_bytes_read: bytes = await aioserial_instance.readline_async(size: int = -1)

# readlines_async: reads multiple lines
lines_of_at_most_certain_size_of_bytes_read: bytes = await aioserial_instance.readlines_async(hint: int = -1)

  5. Initialize AioSerial

    main

    The AioSerial class provides an asynchronous interface for serial port communication. It inherits from serial.Serial, so it accepts all standard pyserial configuration parameters. You can optionally specify an asyncio.AbstractEventLoop via the loop parameter.

    from aioserial import AioSerial
import serial

# Basic initialization
ser = AioSerial(port='/dev/ttyUSB0', baudrate=115200)

# Advanced initialization with specific timeouts and loop control
ser = AioSerial(
    port='/dev/ttyUSB0', 
    baudrate=9600, 
    timeout=1, 
    cancel_read_timeout=2,
    cancel_write_timeout=2
)

  6. Perform asynchronous reads with AioSerial

    main

    Use the *_async methods to read data from the serial port without blocking the asyncio event loop. These methods use internal locks to ensure thread safety and handle task cancellation gracefully by attempting to cancel the underlying blocking operation.

    import asyncio
from aioserial import AioSerial

async def main():
    ser = AioSerial(port='/dev/ttyUSB0', baudrate=9600)
    
    # Read a specific number of bytes
    data = await ser.read_async(size=10)
    
    # Read until a specific character (e.g., newline)
    line = await ser.readline_async()
    
    # Read until a specific byte sequence
    until_data = await ser.read_until_async(expected=b'\r\n')
    
    # Read into an existing buffer
    buf = bytearray(10)
    await ser.readinto_async(buf)

asyncio.run(main())

  7. Perform asynchronous writes with AioSerial

    main

    Use the *_async methods to write data to the serial port asynchronously. These methods ensure that write operations do not block the event loop.

    import asyncio
from aioserial import AioSerial

async def main():
    ser = AioSerial(port='/dev/ttyUSB0', baudrate=9600)
    
    # Write bytes
    bytes_written = await ser.write_async(b'hello\n')
    
    # Write multiple lines
    await ser.writelines_async([b'line1\n', b'line2\n'])

asyncio.run(main())

  8. AioSerial initialization parameters

    main

    The AioSerial constructor accepts the following parameters. Most are passed directly to the underlying serial.Serial class:

    ParameterTypeDefaultDescription
    portOptional[str]NoneThe port name (e.g., '/dev/ttyUSB0' or 'COM3').
    baudrateint9600The baud rate of the serial port.
    bytesizeintserial.EIGHTBITSThe number of data bits per character.
    paritystrserial.PARITY_NONEThe parity check mode.
    stopbitsUnion[float, int]serial.STOPBITS_ONEThe number of stop bits.
    timeoutOptional[Union[float, int]]NoneRead timeout in seconds.
    xonxoffboolFalseSoftware flow control.
    rtsctsboolFalseHardware flow control (RTS/CTS).
    write_timeoutOptional[Union[float, int]]NoneWrite timeout in seconds.
    dsrdtrboolFalseHardware flow control (DSR/DTR).
    inter_byte_timeoutOptional[Union[float, int]]NoneInter-byte timeout in seconds.
    exclusiveOptional[bool]NoneWhether to request exclusive access to the port.
    loopOptional[asyncio.AbstractEventLoop]NoneThe event loop to use.
    cancel_read_timeoutint1Timeout (seconds) for the internal cancel_read operation.
    cancel_write_timeoutint1Timeout (seconds) for the internal cancel_write operation.
    **kwargsdictN/AAdditional arguments passed to serial.Serial.

  9. AioSerial class

    main

    The AioSerial class is a subclass of serial.Serial. It supports all arguments accepted by the standard pySerial constructor, plus several asyncio-specific parameters. Because it is a subclass, isinstance(aioserial.AioSerial(), serial.Serial) returns True.

    # Constructor signature
aioserial_instance: aioserial.AioSerial = aioserial.AioSerial(
    # ... same with what can be passed to serial.Serial ...
    loop: Optional[asyncio.AbstractEventLoop] = None,
    cancel_read_timeout: int = 1,
    cancel_write_timeout: int = 1
)