# Serio: Modern Asynchronous Serial Communication Library




**Serio** is a modern asynchronous serial port library for Python 3.11+ that provides efficient, native async I/O without thread pool overhead. It offers both high-level streams API and low-level transport API for serial communication.
## Features
- ✅ **True async I/O** (no thread pool overhead)
- ✅ **Python 3.11+ native support**
- ✅ **Clean, modern API** with asyncio integration
- ✅ **POSIX and Windows support**
- ✅ **High-level Streams API** (StreamReader/StreamWriter)
- ✅ **Low-level Transport API** for custom protocols
- ✅ **Platform-optimized async I/O** (polling on Windows, file descriptors on POSIX)
- ✅ **Proper flow control** with buffer management
- ✅ Comprehensive exception handling
## Installation
```bash
pip install serio
```
## Getting Started
### High-Level API Example
```python
import asyncio
from serio import open_serial_connection
async def main():
    reader, writer = await open_serial_connection(
        port='/dev/ttyUSB0',
        baudrate=115200
    )
    
    writer.write(b'AT\r\n')
    response = await reader.readuntil(b'\r\n')
    print(f"Received: {response.decode()}")
asyncio.run(main())
```
### Low-Level API Example
```python
import asyncio
from serio import create_serial_connection
class MyProtocol(asyncio.Protocol):
    def connection_made(self, transport):
        print("Connection established")
        
    def data_received(self, data):
        print(f"Received: {data}")
        
    def connection_lost(self, exc):
        print("Connection closed")
async def main():
    loop = asyncio.get_running_loop()
    transport, protocol = await create_serial_connection(
        loop,
        MyProtocol,
        port='/dev/ttyUSB0',
        baudrate=115200
    )
    
    transport.write(b'AT\r\n')
    await asyncio.sleep(1)
    transport.close()
asyncio.run(main())
```
### Context Manager Example
```python
from serio import SerialStream
async def main():
    async with SerialStream(port='/dev/ttyUSB0', baudrate=115200) as (reader, writer):
        writer.write(b'AT\r\n')
        response = await reader.readuntil(b'\r\n')
        print(f"Received: {response.decode()}")
asyncio.run(main())
```
## API Reference
### High-Level API
- `open_serial_connection(**kwargs) -> (StreamReader, StreamWriter)`
- `create_serial_connection(loop, protocol_factory, **kwargs) -> (SerialTransport, Protocol)`
- `SerialStream(**kwargs)` - Context manager for serial connections
### Low-Level API
- `SerialTransport(loop, protocol, serial_instance)` - Asynchronous serial transport
- `SerialError` - Base exception class
- `SerialConnectionError` - Raised when connection fails
- `SerialConfigError` - Raised for invalid configuration
- `PlatformNotSupportedError` - Raised for unsupported platforms
## Configuration Parameters
| Parameter | Description | Default |
|-----------|-------------|---------|
| `port` | Serial port name (e.g., `/dev/ttyUSB0` or `COM3`) | None |
| `baudrate` | Baud rate | 9600 |
| `bytesize` | Data bits | `serial.EIGHTBITS` |
| `parity` | Parity checking | `serial.PARITY_NONE` |
| `stopbits` | Stop bits | `serial.STOPBITS_ONE` |
| `timeout` | Read timeout (seconds) | None |
| `xonxoff` | Software flow control | False |
| `rtscts` | Hardware (RTS/CTS) flow control | False |
| `write_timeout` | Write timeout (seconds) | None |
| `dsrdtr` | Hardware (DSR/DTR) flow control | False |
| `inter_byte_timeout` | Inter-character timeout | None |
| `exclusive` | Exclusive access mode | None |
## Platform Support
- **POSIX**: Uses file descriptors for true async I/O
- **Windows**: Uses efficient polling (5ms intervals)
- **Unsupported**: Other platforms will raise `PlatformNotSupportedError`
## License
This project is licensed under the MIT License - see the [LICENSE](LICENSE) file for details.
## Contributing
Contributions are welcome! Please feel free to submit pull requests or open issues.
---
**Author**: Semenets V. Pavel  
**Version**: 0.1.2  
**Email**: [p.semenets@gmail.com](mailto:p.semenets@gmail.com)
            
         
        Raw data
        
            {
    "_id": null,
    "home_page": null,
    "name": "serio",
    "maintainer": null,
    "docs_url": null,
    "requires_python": ">=3.11",
    "maintainer_email": null,
    "keywords": "serial, asyncio, async, rs232, rs485, com-port, uart, serial-port, asynchronous, pyserial",
    "author": "Semenets V. Pavel",
    "author_email": "p.semenets@gmail.com",
    "download_url": "https://files.pythonhosted.org/packages/2a/10/fb0d39da8afd40d55d1d92929a1bce5bb8c1c2f9b4cdf6c9b6d5a607e790/serio-0.1.2.tar.gz",
    "platform": null,
    "description": "# Serio: Modern Asynchronous Serial Communication Library\n\n\n\n\n\n\n**Serio** is a modern asynchronous serial port library for Python 3.11+ that provides efficient, native async I/O without thread pool overhead. It offers both high-level streams API and low-level transport API for serial communication.\n\n## Features\n\n- \u2705 **True async I/O** (no thread pool overhead)\n- \u2705 **Python 3.11+ native support**\n- \u2705 **Clean, modern API** with asyncio integration\n- \u2705 **POSIX and Windows support**\n- \u2705 **High-level Streams API** (StreamReader/StreamWriter)\n- \u2705 **Low-level Transport API** for custom protocols\n- \u2705 **Platform-optimized async I/O** (polling on Windows, file descriptors on POSIX)\n- \u2705 **Proper flow control** with buffer management\n- \u2705 Comprehensive exception handling\n\n## Installation\n\n```bash\npip install serio\n```\n\n## Getting Started\n\n### High-Level API Example\n\n```python\nimport asyncio\nfrom serio import open_serial_connection\n\nasync def main():\n    reader, writer = await open_serial_connection(\n        port='/dev/ttyUSB0',\n        baudrate=115200\n    )\n    \n    writer.write(b'AT\\r\\n')\n    response = await reader.readuntil(b'\\r\\n')\n    print(f\"Received: {response.decode()}\")\n\nasyncio.run(main())\n```\n\n### Low-Level API Example\n\n```python\nimport asyncio\nfrom serio import create_serial_connection\n\nclass MyProtocol(asyncio.Protocol):\n    def connection_made(self, transport):\n        print(\"Connection established\")\n        \n    def data_received(self, data):\n        print(f\"Received: {data}\")\n        \n    def connection_lost(self, exc):\n        print(\"Connection closed\")\n\nasync def main():\n    loop = asyncio.get_running_loop()\n    transport, protocol = await create_serial_connection(\n        loop,\n        MyProtocol,\n        port='/dev/ttyUSB0',\n        baudrate=115200\n    )\n    \n    transport.write(b'AT\\r\\n')\n    await asyncio.sleep(1)\n    transport.close()\n\nasyncio.run(main())\n```\n\n### Context Manager Example\n\n```python\nfrom serio import SerialStream\n\nasync def main():\n    async with SerialStream(port='/dev/ttyUSB0', baudrate=115200) as (reader, writer):\n        writer.write(b'AT\\r\\n')\n        response = await reader.readuntil(b'\\r\\n')\n        print(f\"Received: {response.decode()}\")\n\nasyncio.run(main())\n```\n\n## API Reference\n\n### High-Level API\n\n- `open_serial_connection(**kwargs) -> (StreamReader, StreamWriter)`\n- `create_serial_connection(loop, protocol_factory, **kwargs) -> (SerialTransport, Protocol)`\n- `SerialStream(**kwargs)` - Context manager for serial connections\n\n### Low-Level API\n\n- `SerialTransport(loop, protocol, serial_instance)` - Asynchronous serial transport\n- `SerialError` - Base exception class\n- `SerialConnectionError` - Raised when connection fails\n- `SerialConfigError` - Raised for invalid configuration\n- `PlatformNotSupportedError` - Raised for unsupported platforms\n\n## Configuration Parameters\n\n| Parameter | Description | Default |\n|-----------|-------------|---------|\n| `port` | Serial port name (e.g., `/dev/ttyUSB0` or `COM3`) | None |\n| `baudrate` | Baud rate | 9600 |\n| `bytesize` | Data bits | `serial.EIGHTBITS` |\n| `parity` | Parity checking | `serial.PARITY_NONE` |\n| `stopbits` | Stop bits | `serial.STOPBITS_ONE` |\n| `timeout` | Read timeout (seconds) | None |\n| `xonxoff` | Software flow control | False |\n| `rtscts` | Hardware (RTS/CTS) flow control | False |\n| `write_timeout` | Write timeout (seconds) | None |\n| `dsrdtr` | Hardware (DSR/DTR) flow control | False |\n| `inter_byte_timeout` | Inter-character timeout | None |\n| `exclusive` | Exclusive access mode | None |\n\n\n## Platform Support\n\n- **POSIX**: Uses file descriptors for true async I/O\n- **Windows**: Uses efficient polling (5ms intervals)\n- **Unsupported**: Other platforms will raise `PlatformNotSupportedError`\n\n## License\n\nThis project is licensed under the MIT License - see the [LICENSE](LICENSE) file for details.\n\n## Contributing\n\nContributions are welcome! Please feel free to submit pull requests or open issues.\n\n---\n\n**Author**: Semenets V. Pavel  \n**Version**: 0.1.2  \n**Email**: [p.semenets@gmail.com](mailto:p.semenets@gmail.com)\n\n",
    "bugtrack_url": null,
    "license": "MIT",
    "summary": "Modern asynchronous serial port library for Python 3.11+",
    "version": "0.1.2",
    "project_urls": {
        "Documentation": "https://github.com/xDarkmanx/serio#readme",
        "Homepage": "https://github.com/xDarkmanx/serio",
        "Repository": "https://github.com/xDarkmanx/serio"
    },
    "split_keywords": [
        "serial",
        " asyncio",
        " async",
        " rs232",
        " rs485",
        " com-port",
        " uart",
        " serial-port",
        " asynchronous",
        " pyserial"
    ],
    "urls": [
        {
            "comment_text": null,
            "digests": {
                "blake2b_256": "d9550c5fc391d8a6422e1d35fd0dc4d50acec14e1c1f2f8fc54e72debd20778e",
                "md5": "024aab7e1432373ea89864c8dc62c3c5",
                "sha256": "8f3f2e34d9f2bb5415e3bf4dcca2698576c3829b3af39544e01974200845586b"
            },
            "downloads": -1,
            "filename": "serio-0.1.2-py3-none-any.whl",
            "has_sig": false,
            "md5_digest": "024aab7e1432373ea89864c8dc62c3c5",
            "packagetype": "bdist_wheel",
            "python_version": "py3",
            "requires_python": ">=3.11",
            "size": 10678,
            "upload_time": "2025-11-02T05:25:52",
            "upload_time_iso_8601": "2025-11-02T05:25:52.399587Z",
            "url": "https://files.pythonhosted.org/packages/d9/55/0c5fc391d8a6422e1d35fd0dc4d50acec14e1c1f2f8fc54e72debd20778e/serio-0.1.2-py3-none-any.whl",
            "yanked": false,
            "yanked_reason": null
        },
        {
            "comment_text": null,
            "digests": {
                "blake2b_256": "2a10fb0d39da8afd40d55d1d92929a1bce5bb8c1c2f9b4cdf6c9b6d5a607e790",
                "md5": "e6e83536af11c372587f02a61ba98ae1",
                "sha256": "92c2efdda65e44ae0fb440e18546f94de39b39c547f0b07536ba47cee9658145"
            },
            "downloads": -1,
            "filename": "serio-0.1.2.tar.gz",
            "has_sig": false,
            "md5_digest": "e6e83536af11c372587f02a61ba98ae1",
            "packagetype": "sdist",
            "python_version": "source",
            "requires_python": ">=3.11",
            "size": 9973,
            "upload_time": "2025-11-02T05:25:53",
            "upload_time_iso_8601": "2025-11-02T05:25:53.601385Z",
            "url": "https://files.pythonhosted.org/packages/2a/10/fb0d39da8afd40d55d1d92929a1bce5bb8c1c2f9b4cdf6c9b6d5a607e790/serio-0.1.2.tar.gz",
            "yanked": false,
            "yanked_reason": null
        }
    ],
    "upload_time": "2025-11-02 05:25:53",
    "github": true,
    "gitlab": false,
    "bitbucket": false,
    "codeberg": false,
    "github_user": "xDarkmanx",
    "github_project": "serio#readme",
    "travis_ci": false,
    "coveralls": false,
    "github_actions": true,
    "lcname": "serio"
}