### pyserial-asyncio Example

Source: https://github.com/johannjhang/aioserial.py/blob/main/README.md

This snippet shows the original example usage from pyserial-asyncio for establishing a serial connection and handling data.

```python
import asyncio
import serial_asyncio

class Output(asyncio.Protocol):
    def connection_made(self, transport):
        self.transport = transport
        print('port opened', transport)
        transport.serial.rts = False  # You can manipulate Serial object via transport
        transport.write(b'Hello, World!\n')  # Write serial data via transport

    def data_received(self, data):
        print('data received', repr(data))
        if b'\n' in data:
            self.transport.close()

    def connection_lost(self, exc):
        print('port closed')
        self.transport.loop.stop()

    def pause_writing(self):
        print('pause writing')
        print(self.transport.get_write_buffer_size())

    def resume_writing(self):
        print(self.transport.get_write_buffer_size())
        print('resume writing')

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()
```

--------------------------------

### aioserial Equivalence to pyserial-asyncio

Source: https://github.com/johannjhang/aioserial.py/blob/main/README.md

This snippet provides the equivalent aioserial implementation for the pyserial-asyncio example, demonstrating asynchronous reading, writing, and connection handling.

```python
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

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')))
```

--------------------------------

### Read from Serial Port Asynchronously

Source: https://github.com/johannjhang/aioserial.py/blob/main/README.md

A simple example demonstrating how to read data from a serial port asynchronously using aioserial and print it to the console. Ensure the correct port name is specified.

```python
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')))
```

--------------------------------

### Initialize AioSerial

Source: https://context7.com/johannjhang/aioserial.py/llms.txt

Demonstrates basic and full parameter initialization of AioSerial, which extends pySerial. Verifies subclassing relationship.

```python
import asyncio
import aioserial

# Basic usage — open COM1 at 9600 baud
ser = aioserial.AioSerial(port='COM1')

# Full parameter usage — open /dev/ttyUSB0 at 115200 baud
ser = aioserial.AioSerial(
    port='/dev/ttyUSB0',
    baudrate=115200,
    bytesize=aioserial.EIGHTBITS,
    parity=aioserial.PARITY_NONE,
    stopbits=aioserial.STOPBITS_ONE,
    timeout=2,                  # read timeout in seconds
    write_timeout=2,            # write timeout in seconds
    xonxoff=False,
    rtscts=False,
    dsrdtr=False,
    cancel_read_timeout=1,      # seconds to wait when cancelling a read
    cancel_write_timeout=1,     # seconds to wait when cancelling a write
)

# Verify it is a true serial.Serial subclass
import serial
assert isinstance(ser, serial.Serial)   # True
assert issubclass(aioserial.AioSerial, serial.Serial)  # True
assert aioserial.Serial is serial.Serial  # True
```

--------------------------------

### AioSerial Constructor

Source: https://github.com/johannjhang/aioserial.py/blob/main/README.md

Initializes an AioSerial instance. It accepts the same arguments as serial.Serial, plus optional asyncio loop and timeout parameters.

```APIDOC
## Constructor

### Description
Initializes an AioSerial instance. It accepts the same arguments as serial.Serial, plus optional asyncio loop and timeout parameters.

### Parameters
- `loop` (Optional[asyncio.AbstractEventLoop]) - The asyncio event loop to use.
- `cancel_read_timeout` (int) - Timeout for read operations in seconds.
- `cancel_write_timeout` (int) - Timeout for write operations in seconds.

### Example
```python
import aioserial

aioserial_instance = aioserial.AioSerial(
    port='COM1',
    baudrate=9600,
    loop=asyncio.get_event_loop(),
    cancel_read_timeout=1,
    cancel_write_timeout=1
)
```
```

--------------------------------

### AioSerial Constructor Signature

Source: https://github.com/johannjhang/aioserial.py/blob/main/README.md

Illustrates the constructor signature for aioserial.AioSerial, highlighting that it accepts arguments similar to serial.Serial, along with optional asyncio loop and timeout parameters.

```python
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)
```

--------------------------------

### Simultaneous Read and Write with `asyncio.gather`

Source: https://context7.com/johannjhang/aioserial.py/llms.txt

Demonstrates how to perform simultaneous read and write operations on the serial port using `asyncio.gather`.

```APIDOC
## Simultaneous Read and Write with `asyncio.gather`

### Description
Because reads and writes use separate executors and locks, they can run concurrently with `asyncio.gather`.

### Method
`await asyncio.gather(send(), receive())`

### Parameters
None

### Request Example
```python
import asyncio
import aioserial

async def echo_client():
    ser = aioserial.AioSerial(
        port='/dev/ttyUSB0',
        baudrate=115200,
        timeout=5,
    )

    async def send():
        await asyncio.sleep(0.1)  # brief delay so reader is ready
        await ser.write_async(b'Hello, World!\n')

    async def receive():
        data = await ser.read_until_async(expected=b'\n')
        print(f"Echo received: {data.decode(errors='ignore').strip()}")

    try:
        await asyncio.gather(send(), receive())
    finally:
        ser.close()

asyncio.run(echo_client())
```

### Response
#### Success Response (200)
- **Echo received** (str) - The echoed data received from the serial port.

#### Response Example
```
Echo received: Hello, World!
```
```

--------------------------------

### Concurrent Read and Write with asyncio.gather

Source: https://context7.com/johannjhang/aioserial.py/llms.txt

Demonstrates concurrent read and write operations on a serial port using `asyncio.gather`. Reads and writes use separate executors and locks, allowing them to run in parallel.

```python
import asyncio
import aioserial

async def echo_client():
    ser = aioserial.AioSerial(
        port='/dev/ttyUSB0',
        baudrate=115200,
        timeout=5,
    )

    async def send():
        await asyncio.sleep(0.1)  # brief delay so reader is ready
        await ser.write_async(b'Hello, World!\n')

    async def receive():
        data = await ser.read_until_async(expected=b'\n')
        print(f"Echo received: {data.decode(errors='ignore').strip()}")

    try:
        await asyncio.gather(send(), receive())
    finally:
        ser.close()

asyncio.run(echo_client())
```

--------------------------------

### Read Until Specific Data Asynchronously

Source: https://github.com/johannjhang/aioserial.py/blob/main/README.md

Demonstrates the `read_until_async` method for reading data until a specific byte sequence is encountered or a maximum size is reached.

```python
at_most_certain_size_of_bytes_read: bytes = \
    await aioserial_instance.read_until_async(
        expected: bytes = aioserial.LF, size: Optional[int] = None)
```

--------------------------------

### Write Data Asynchronously

Source: https://github.com/johannjhang/aioserial.py/blob/main/README.md

Shows the signature for `write_async`, used to write bytes-like data to the serial port.

```python
number_of_byte_like_data_written: int = \
    await aioserial_instance.write_async(bytes_like_data)
```

--------------------------------

### `readlines_async`

Source: https://context7.com/johannjhang/aioserial.py/llms.txt

Asynchronously reads multiple lines and returns them as a list of `bytes` objects. The optional `hint` parameter specifies an approximate total byte target; -1 (default) reads until EOF or timeout.

```APIDOC
## `readlines_async`

### Description
Asynchronously reads multiple lines and returns them as a list of `bytes` objects. The optional `hint` parameter specifies an approximate total byte target; -1 (default) reads until EOF or timeout.

### Method
`await ser.readlines_async(hint=-1)`

### Parameters
#### Path Parameters
None

#### Query Parameters
None

#### Request Body
None

### Request Example
```python
import asyncio
import aioserial

async def read_all_lines():
    ser = aioserial.AioSerial(port='/dev/ttyUSB0', baudrate=9600, timeout=3)
    try:
        lines = await ser.readlines_async()
        for i, line in enumerate(lines):
            print(f"[{i}] {line.decode(errors='ignore').rstrip()}")

        # Read approximately 128 bytes' worth of lines
        chunk = await ser.readlines_async(hint=128)
        print(f"Got {len(chunk)} lines in ~128 byte hint window")
    finally:
        ser.close()

asyncio.run(read_all_lines())
```

### Response
#### Success Response (200)
- **lines** (list[bytes]) - A list of lines read from the serial port.
- **chunk** (list[bytes]) - A list of lines read within the specified byte hint.

#### Response Example
```
[0] First line
[1] Second line
Got 5 lines in ~128 byte hint window
```
```

--------------------------------

### Asynchronously Read Multiple Lines with readlines_async

Source: https://context7.com/johannjhang/aioserial.py/llms.txt

Reads multiple lines asynchronously and returns them as a list of bytes. The 'hint' parameter suggests an approximate total byte count. Defaults to reading until EOF or timeout.

```python
import asyncio
import aioserial

async def read_all_lines():
    ser = aioserial.AioSerial(port='/dev/ttyUSB0', baudrate=9600, timeout=3)
    try:
        lines = await ser.readlines_async()
        for i, line in enumerate(lines):
            print(f"[{i}] {line.decode(errors='ignore').rstrip()}")

        # Read approximately 128 bytes' worth of lines
        chunk = await ser.readlines_async(hint=128)
        print(f"Got {len(chunk)} lines in ~128 byte hint window")
    finally:
        ser.close()

asyncio.run(read_all_lines())
```

--------------------------------

### Verify AioSerial Inheritance

Source: https://github.com/johannjhang/aioserial.py/blob/main/README.md

This snippet demonstrates that aioserial.AioSerial is an instance and subclass of serial.Serial, confirming compatibility with pySerial.

```python
>>> import aioserial
>>> import serial

>>> isinstance(aioserial.AioSerial(), serial.Serial)
True

>>> issubclass(aioserial.AioSerial, serial.Serial)
True

>>> aioserial.Serial is serial.Serial
True
```

--------------------------------

### write_async

Source: https://github.com/johannjhang/aioserial.py/blob/main/README.md

Asynchronously writes bytes-like data to the serial port.

```APIDOC
## write_async

### Description
Asynchronously writes bytes-like data to the serial port.

### Method
`await aioserial_instance.write_async(bytes_like_data)`

### Parameters
- `bytes_like_data` (bytes) - The data to write to the serial port.

### Response
- `number_of_byte_like_data_written` (int) - The number of bytes written.

### Example
```python
await aioserial_instance.write_async(b'Hello, world!\n')
```
```

--------------------------------

### Write Multiple Lines Asynchronously

Source: https://github.com/johannjhang/aioserial.py/blob/main/README.md

Illustrates the `writelines_async` method for writing a list of bytes-like data to the serial port.

```python
number_of_byte_like_data_in_the_given_list_written: int = \
    await aioserial_instance.writelines_async(list_of_bytes_like_data)
```

--------------------------------

### writelines_async

Source: https://github.com/johannjhang/aioserial.py/blob/main/README.md

Asynchronously writes a list of bytes-like data to the serial port.

```APIDOC
## writelines_async

### Description
Asynchronously writes a list of bytes-like data to the serial port.

### Method
`await aioserial_instance.writelines_async(list_of_bytes_like_data)`

### Parameters
- `list_of_bytes_like_data` (list[bytes]) - A list of bytes-like objects to write.

### Response
- `number_of_byte_like_data_in_the_given_list_written` (int) - The total number of bytes written from the list.

### Example
```python
await aioserial_instance.writelines_async([b'line1\n', b'line2\n'])
```
```

--------------------------------

### read_async

Source: https://github.com/johannjhang/aioserial.py/blob/main/README.md

Asynchronously reads up to a specified number of bytes from the serial port.

```APIDOC
## read_async

### Description
Asynchronously reads up to a specified number of bytes from the serial port.

### Method
`await aioserial_instance.read_async(size: int = 1)`

### Parameters
- `size` (int) - The maximum number of bytes to read. Defaults to 1.

### Response
- `bytes_read` (bytes) - The bytes read from the serial port.

### Example
```python
bytes_read = await aioserial_instance.read_async(size=10)
```
```

--------------------------------

### Read Multiple Lines Asynchronously

Source: https://github.com/johannjhang/aioserial.py/blob/main/README.md

Demonstrates the `readlines_async` method for reading multiple lines from the serial port, with an optional hint for the total size.

```python
lines_of_at_most_certain_size_of_bytes_read: bytes = \
    await aioserial_instance.readlines_async(hint: int = -1)
```

--------------------------------

### Read Line Asynchronously

Source: https://github.com/johannjhang/aioserial.py/blob/main/README.md

Shows the signature for `readline_async`, which reads a single line from the serial port, up to a specified maximum size.

```python
a_line_of_at_most_certain_size_of_bytes_read: bytes = \
    await aioserial_instance.readline_async(size: int = -1)
```

--------------------------------

### readlines_async

Source: https://github.com/johannjhang/aioserial.py/blob/main/README.md

Asynchronously reads multiple lines from the serial port, up to a specified hint size.

```APIDOC
## readlines_async

### Description
Asynchronously reads multiple lines from the serial port, up to a specified hint size.

### Method
`await aioserial_instance.readlines_async(hint: int = -1)`

### Parameters
- `hint` (int) - A hint for the maximum number of bytes to read. Defaults to -1 (no limit).

### Response
- `lines_of_at_most_certain_size_of_bytes_read` (bytes) - A bytes object containing all the lines read.

### Example
```python
all_lines = await aioserial_instance.readlines_async()
```
```

--------------------------------

### AioSerial Constructor

Source: https://context7.com/johannjhang/aioserial.py/llms.txt

The AioSerial constructor extends the standard serial.Serial constructor with two additional parameters: cancel_read_timeout and cancel_write_timeout. It accepts all standard pySerial constructor arguments.

```APIDOC
## AioSerial Constructor

### Description
`AioSerial` extends `serial.Serial` with two extra parameters: `cancel_read_timeout` and `cancel_write_timeout` (both default to 1 second). All standard pySerial constructor arguments are accepted unchanged.

### Parameters
#### Path Parameters
None

#### Query Parameters
None

#### Request Body
None

### Constructor Arguments
- **port** (str) - Required - The serial port name (e.g., 'COM1' on Windows, '/dev/ttyUSB0' on Linux).
- **baudrate** (int) - Optional - The baud rate (default is usually 9600).
- **bytesize** (int) - Optional - Number of data bits (e.g., `aioserial.EIGHTBITS`).
- **parity** (str) - Optional - Parity setting (e.g., `aioserial.PARITY_NONE`).
- **stopbits** (int) - Optional - Number of stop bits (e.g., `aioserial.STOPBITS_ONE`).
- **timeout** (float) - Optional - Read timeout in seconds.
- **write_timeout** (float) - Optional - Write timeout in seconds.
- **xonxoff** (bool) - Optional - Enable XON/XOFF flow control.
- **rtscts** (bool) - Optional - Enable RTS/CTS hardware flow control.
- **dsrdtr** (bool) - Optional - Enable DSR/DTR hardware flow control.
- **cancel_read_timeout** (float) - Optional - Seconds to wait when cancelling a read operation (defaults to 1).
- **cancel_write_timeout** (float) - Optional - Seconds to wait when cancelling a write operation (defaults to 1).

### Request Example
```python
import asyncio
import aioserial

# Basic usage
ser = aioserial.AioSerial(port='COM1')

# Full parameter usage
ser = aioserial.AioSerial(
    port='/dev/ttyUSB0',
    baudrate=115200,
    bytesize=aioserial.EIGHTBITS,
    parity=aioserial.PARITY_NONE,
    stopbits=aioserial.STOPBITS_ONE,
    timeout=2,
    write_timeout=2,
    xonxoff=False,
    rtscts=False,
    dsrdtr=False,
    cancel_read_timeout=1,
    cancel_write_timeout=1,
)
```

### Response
#### Success Response (Instance of AioSerial)
- **Instance** (AioSerial) - An initialized AioSerial object, which is a subclass of `serial.Serial`.
```

--------------------------------

### Read Data Asynchronously

Source: https://github.com/johannjhang/aioserial.py/blob/main/README.md

Shows the signature for the `read_async` method, which reads a specified number of bytes from the serial port.

```python
bytes_read: bytes = \
    await aioserial_instance.read_async(size: int = 1)
```

--------------------------------

### read_until_async

Source: https://github.com/johannjhang/aioserial.py/blob/main/README.md

Asynchronously reads bytes until a specified delimiter is found or a maximum size is reached.

```APIDOC
## read_until_async

### Description
Asynchronously reads bytes until a specified delimiter is found or a maximum size is reached.

### Method
`await aioserial_instance.read_until_async(expected: bytes = aioserial.LF, size: Optional[int] = None)`

### Parameters
- `expected` (bytes) - The delimiter bytes to read until. Defaults to `aioserial.LF` (line feed).
- `size` (Optional[int]) - The maximum number of bytes to read. If None, there is no size limit.

### Response
- `at_most_certain_size_of_bytes_read` (bytes) - The bytes read from the serial port, including the delimiter if found.

### Example
```python
line = await aioserial_instance.read_until_async(expected=b'\r\n')
```
```

--------------------------------

### Read Into Buffer Asynchronously

Source: https://github.com/johannjhang/aioserial.py/blob/main/README.md

Illustrates the `readinto_async` method, which reads data directly into a mutable buffer like an array or bytearray.

```python
number_of_byte_read: int = \
    await aioserial_instance.readinto_async(b: Union[array.array, bytearray])
```

--------------------------------

### Asynchronously Write Multiple Lines with writelines_async

Source: https://context7.com/johannjhang/aioserial.py/llms.txt

Asynchronously writes a list of bytes-like objects to the serial port sequentially. Returns the total number of bytes written.

```python
import asyncio
import aioserial

async def write_multiple():
    ser = aioserial.AioSerial(port='/dev/ttyUSB0', baudrate=9600)
    try:
        lines = [b'line one\n', b'line two\n', bytearray(b'line three\n')]
        total: int = await ser.writelines_async(lines)
        print(f"Wrote {total} bytes total")  # Wrote 30 bytes total
    finally:
        ser.close()

asyncio.run(write_multiple())
```

--------------------------------

### readinto_async

Source: https://github.com/johannjhang/aioserial.py/blob/main/README.md

Asynchronously reads bytes into a pre-allocated buffer.

```APIDOC
## readinto_async

### Description
Asynchronously reads bytes into a pre-allocated buffer.

### Method
`await aioserial_instance.readinto_async(b: Union[array.array, bytearray])`

### Parameters
- `b` (Union[array.array, bytearray]) - The buffer to read bytes into.

### Response
- `number_of_byte_read` (int) - The number of bytes read into the buffer.

### Example
```python
import array

buffer = array.array('B', [0] * 100)
bytes_read = await aioserial_instance.readinto_async(buffer)
```
```

--------------------------------

### read_async

Source: https://context7.com/johannjhang/aioserial.py/llms.txt

Asynchronously reads up to a specified size of bytes from the serial port. This method wraps the blocking `serial.Serial.read` call in a thread executor and uses an asyncio lock to ensure concurrent read operations are serialized.

```APIDOC
## read_async

### Description
Asynchronously reads up to `size` bytes from the serial port. Wraps `serial.Serial.read` in a thread executor and acquires an async read lock, so concurrent callers are serialized automatically.

### Method
`await ser.read_async(size)`

### Parameters
#### Path Parameters
None

#### Query Parameters
None

#### Request Body
None

### Parameters
- **size** (int) - Optional - The maximum number of bytes to read. If not specified, it reads one byte by default.

### Request Example
```python
import asyncio
import aioserial

async def read_bytes():
    ser = aioserial.AioSerial(port='/dev/ttyUSB0', baudrate=115200, timeout=5)
    try:
        # Read exactly 4 bytes
        data: bytes = await ser.read_async(size=4)
        print(f"Received {len(data)} bytes: {data.hex()}")

        # Read 1 byte (default)
        single: bytes = await ser.read_async()
        print(f"Single byte: {single!r}")
    except asyncio.CancelledError:
        print("Read was cancelled; serial port cleaned up automatically")
        raise
    finally:
        ser.close()

asyncio.run(read_bytes())
```

### Response
#### Success Response (bytes)
- **data** (bytes) - The bytes read from the serial port. Returns an empty bytes object if a timeout occurs before any data is received.
```

--------------------------------

### readline_async

Source: https://github.com/johannjhang/aioserial.py/blob/main/README.md

Asynchronously reads a single line from the serial port, up to a specified maximum size.

```APIDOC
## readline_async

### Description
Asynchronously reads a single line from the serial port, up to a specified maximum size.

### Method
`await aioserial_instance.readline_async(size: int = -1)`

### Parameters
- `size` (int) - The maximum number of bytes to read. Defaults to -1 (no limit).

### Response
- `a_line_of_at_most_certain_size_of_bytes_read` (bytes) - The line read from the serial port, including the newline character if found.

### Example
```python
line = await aioserial_instance.readline_async()
```
```

--------------------------------

### Cancellation Safety with aioserial

Source: https://context7.com/johannjhang/aioserial.py/llms.txt

Illustrates how aioserial handles task cancellation safely, ensuring the serial port remains usable even when read operations are cancelled externally. The `cancel_read_timeout` parameter controls how long cancellation attempts to complete.

```python
import asyncio
import aioserial

async def cancellable_read():
    ser = aioserial.AioSerial(
        port='/dev/ttyUSB0',
        baudrate=115200,
        cancel_read_timeout=2,   # allow up to 2s for cancel_read to complete
    )
    task = asyncio.create_task(ser.read_async(size=64))
    await asyncio.sleep(0.5)  # simulate cancellation after 500 ms
    task.cancel()
    try:
        await task
    except asyncio.CancelledError:
        print("Task cancelled cleanly; serial port is still usable")
    finally:
        ser.close()

asyncio.run(cancellable_read())
```

--------------------------------

### Asynchronously Write Data with write_async

Source: https://context7.com/johannjhang/aioserial.py/llms.txt

Asynchronously writes bytes, bytearray, or memoryview to the serial port. Returns the number of bytes written. Concurrent writes are handled by an internal async lock.

```python
import asyncio
import aioserial

async def write_data():
    ser = aioserial.AioSerial(port='/dev/ttyUSB0', baudrate=115200)
    try:
        # Write bytes literal
        n: int = await ser.write_async(b'Hello, World!\n')
        print(f"Wrote {n} bytes")  # Wrote 14 bytes

        # Write bytearray
        payload = bytearray([0x01, 0x02, 0x03, 0xFF])
        n2: int = await ser.write_async(payload)
        print(f"Wrote {n2} bytes of binary payload")

        # Concurrent read and write
        read_task = asyncio.create_task(ser.read_async(size=14))
        await ser.write_async(b'Ping\n')
        response = await read_task
        print(f"Response: {response.decode(errors='ignore')}")
    except asyncio.CancelledError:
        print("Write cancelled; port cleanup handled automatically")
        raise
    finally:
        ser.close()

asyncio.run(write_data())
```

--------------------------------

### Asynchronously Read a Single Line with readline_async

Source: https://context7.com/johannjhang/aioserial.py/llms.txt

Reads a single line terminated by '\n' asynchronously. Use the 'size' parameter to limit the number of bytes read. Defaults to no limit.

```python
import asyncio
import aioserial

async def read_lines_loop():
    ser = aioserial.AioSerial(port='/dev/ttyUSB0', baudrate=9600, timeout=5)
    try:
        for _ in range(5):
            line: bytes = await ser.readline_async()
            print(f">> {line.decode(errors='ignore').rstrip()}")

        # With a byte cap
        partial: bytes = await ser.readline_async(size=32)
        print(f"Capped line ({len(partial)} bytes): {partial!r}")
    finally:
        ser.close()

asyncio.run(read_lines_loop())
```

--------------------------------

### Cancellation Safety

Source: https://context7.com/johannjhang/aioserial.py/llms.txt

Illustrates how aioserial handles cancellation to ensure the serial port remains in a clean state.

```APIDOC
## Cancellation Safety

### Description
aioserial shields the `cancel_read` / `cancel_write` cleanup from cancellation, ensuring the port is left in a clean state even when a task is cancelled externally.

### Method
`task.cancel()` followed by `await task` within a try-except `asyncio.CancelledError` block.

### Parameters
None

### Request Example
```python
import asyncio
import aioserial

async def cancellable_read():
    ser = aioserial.AioSerial(
        port='/dev/ttyUSB0',
        baudrate=115200,
        cancel_read_timeout=2,   # allow up to 2s for cancel_read to complete
    )
    task = asyncio.create_task(ser.read_async(size=64))
    await asyncio.sleep(0.5)  # simulate cancellation after 500 ms
    task.cancel()
    try:
        await task
    except asyncio.CancelledError:
        print("Task cancelled cleanly; serial port is still usable")
    finally:
        ser.close()

asyncio.run(cancellable_read())
```

### Response
#### Success Response (200)
- **Output message** (str) - Indicates that the task was cancelled cleanly and the serial port is still usable.

#### Response Example
```
Task cancelled cleanly; serial port is still usable
```
```

--------------------------------

### `write_async`

Source: https://context7.com/johannjhang/aioserial.py/llms.txt

Asynchronously writes a `bytes`, `bytearray`, or `memoryview` object to the serial port. Returns the number of bytes written. Concurrent writes are serialized via an internal async lock.

```APIDOC
## `write_async`

### Description
Asynchronously writes a `bytes`, `bytearray`, or `memoryview` object to the serial port. Returns the number of bytes written. Concurrent writes are serialized via an internal async lock.

### Method
`await ser.write_async(data)`

### Parameters
#### Path Parameters
None

#### Query Parameters
None

#### Request Body
- **data** (bytes | bytearray | memoryview) - The data to write to the serial port.

### Request Example
```python
import asyncio
import aioserial

async def write_data():
    ser = aioserial.AioSerial(port='/dev/ttyUSB0', baudrate=115200)
    try:
        # Write bytes literal
        n: int = await ser.write_async(b'Hello, World!\n')
        print(f"Wrote {n} bytes")  # Wrote 14 bytes

        # Write bytearray
        payload = bytearray([0x01, 0x02, 0x03, 0xFF])
        n2: int = await ser.write_async(payload)
        print(f"Wrote {n2} bytes of binary payload")

        # Concurrent read and write
        read_task = asyncio.create_task(ser.read_async(size=14))
        await ser.write_async(b'Ping\n')
        response = await read_task
        print(f"Response: {response.decode(errors='ignore')}")
    except asyncio.CancelledError:
        print("Write cancelled; port cleanup handled automatically")
        raise
    finally:
        ser.close()

asyncio.run(write_data())
```

### Response
#### Success Response (200)
- **n** (int) - The number of bytes written to the serial port.
- **n2** (int) - The number of bytes written to the serial port.
- **response** (bytes) - The response received from the serial port after writing.

#### Response Example
```
Wrote 14 bytes
Wrote 4 bytes of binary payload
Response: Hello, World!\n
```
```

--------------------------------

### Asynchronous Read Into Buffer with AioSerial

Source: https://context7.com/johannjhang/aioserial.py/llms.txt

Reads bytes directly into a bytearray or array.array buffer asynchronously. Returns the number of bytes read. Ensures serial port cleanup.

```python
import asyncio
import array
import aioserial

async def read_into_buffer():
    ser = aioserial.AioSerial(port='/dev/ttyUSB0', baudrate=115200, timeout=5)
    try:
        # Read into a bytearray
        buf = bytearray(16)
        n: int = await ser.readinto_async(buf)
        print(f"Read {n} bytes into buffer: {buf[:n].hex()}")

        # Read into a typed array
        typed_buf = array.array('B', [0] * 8)
        n2: int = await ser.readinto_async(typed_buf)
        print(f"Read {n2} bytes into typed array: {list(typed_buf[:n2])}")
    finally:
        ser.close()

asyncio.run(read_into_buffer())
```

--------------------------------

### read_until_async

Source: https://context7.com/johannjhang/aioserial.py/llms.txt

Reads bytes from the serial port until a specified terminator sequence is encountered or a maximum byte count is reached. It defaults to reading until a newline character.

```APIDOC
## read_until_async

### Description
Reads bytes until a specified terminator sequence is found or an optional maximum byte count is reached. Defaults to reading until a line-feed (`\n`) byte.

### Method
`await ser.read_until_async(expected, size)`

### Parameters
#### Path Parameters
None

#### Query Parameters
None

#### Request Body
None

### Parameters
- **expected** (bytes) - Optional - The sequence of bytes to read until. Defaults to `b'\n'` (newline).
- **size** (int) - Optional - The maximum number of bytes to read. If not specified, there is no size limit other than available buffer space and timeouts.

### Request Example
```python
import asyncio
import aioserial

async def read_until_marker():
    ser = aioserial.AioSerial(port='/dev/ttyUSB0', baudrate=9600, timeout=10)
    try:
        # Read until newline (default)
        line: bytes = await ser.read_until_async()
        print(f"Line: {line.decode(errors='ignore').strip()}")

        # Read until a custom terminator, capped at 64 bytes
        data: bytes = await ser.read_until_async(expected=b'\r\n', size=64)
        print(f"Data: {data!r}")
    finally:
        ser.close()

asyncio.run(read_until_marker())
```

### Response
#### Success Response (bytes)
- **data** (bytes) - The bytes read from the serial port up to the terminator or size limit. Returns an empty bytes object if a timeout occurs before the terminator is found or any data is received.
```

--------------------------------

### readinto_async

Source: https://context7.com/johannjhang/aioserial.py/llms.txt

Asynchronously reads bytes directly into a pre-allocated buffer, such as a `bytearray` or `array.array`. It returns the number of bytes that were actually read into the buffer.

```APIDOC
## readinto_async

### Description
Asynchronously reads bytes directly into an existing `bytearray` or `array.array` buffer. Returns the number of bytes actually read.

### Method
`await ser.readinto_async(buffer)`

### Parameters
#### Path Parameters
None

#### Query Parameters
None

#### Request Body
None

### Parameters
- **buffer** (bytearray or array.array) - Required - The buffer object to read data into.

### Request Example
```python
import asyncio
import array
import aioserial

async def read_into_buffer():
    ser = aioserial.AioSerial(port='/dev/ttyUSB0', baudrate=115200, timeout=5)
    try:
        # Read into a bytearray
        buf = bytearray(16)
        n: int = await ser.readinto_async(buf)
        print(f"Read {n} bytes into buffer: {buf[:n].hex()}")

        # Read into a typed array
        typed_buf = array.array('B', [0] * 8)
        n2: int = await ser.readinto_async(typed_buf)
        print(f"Read {n2} bytes into typed array: {list(typed_buf[:n2])}")
    finally:
        ser.close()

asyncio.run(read_into_buffer())
```

### Response
#### Success Response (int)
- **bytes_read** (int) - The number of bytes that were read into the buffer. Returns 0 if a timeout occurs before any data is read.
```

--------------------------------

### Asynchronous Read Bytes with AioSerial

Source: https://context7.com/johannjhang/aioserial.py/llms.txt

Reads a specified number of bytes asynchronously from the serial port. Handles asyncio.CancelledError and ensures serial port cleanup.

```python
import asyncio
import aioserial

async def read_bytes():
    ser = aioserial.AioSerial(port='/dev/ttyUSB0', baudrate=115200, timeout=5)
    try:
        # Read exactly 4 bytes
        data: bytes = await ser.read_async(size=4)
        print(f"Received {len(data)} bytes: {data.hex()}")
        # Expected output: Received 4 bytes: deadbeef

        # Read 1 byte (default)
        single: bytes = await ser.read_async()
        print(f"Single byte: {single!r}")
    except asyncio.CancelledError:
        print("Read was cancelled; serial port cleaned up automatically")
        raise
    finally:
        ser.close()

asyncio.run(read_bytes())
```

--------------------------------

### Asynchronous Read Until Terminator with AioSerial

Source: https://context7.com/johannjhang/aioserial.py/llms.txt

Reads bytes until a terminator sequence or a maximum size is reached. Defaults to reading until a newline character. Ensures serial port cleanup.

```python
import asyncio
import aioserial

async def read_until_marker():
    ser = aioserial.AioSerial(port='/dev/ttyUSB0', baudrate=9600, timeout=10)
    try:
        # Read until newline (default)
        line: bytes = await ser.read_until_async()
        print(f"Line: {line.decode(errors='ignore').strip()}")

        # Read until a custom terminator, capped at 64 bytes
        data: bytes = await ser.read_until_async(expected=b'\r\n', size=64)
        print(f"Data: {data!r}")
    finally:
        ser.close()

asyncio.run(read_until_marker())
```

=== COMPLETE CONTENT === This response contains all available snippets from this library. No additional content exists. Do not make further requests.