Universal Reliable Serial Transport (URST) protocol implementation for Python. See URST Specification for details.
URST is a lightweight, acknowledgment-based communication protocol designed for reliable data transmission over serial connections. This Python implementation provides:
- Reliable delivery with automatic retransmission
- Error detection via CRC-16
- Frame delimiting through COBS encoding
- Message fragmentation for large payloads
- Simple API for easy integration
uv add install urst-python...or...
pip install urst-pythonimport urst
# Create handler
handler = urst.create_handler('/dev/ttyUSB0', baudrate=115200)
# Send data
success = handler.send(b"Hello, World!")
if success:
print("Message sent successfully!")
# Receive data (with timeout)
data = handler.receive(timeout_ms=5000)
if data:
print(f"Received: {data}")
# Cleanup
handler.close()import urst
# Automatic cleanup with context manager
with urst.create_handler('/dev/ttyUSB0', baudrate=115200) as handler:
# Send data
handler.send(b"Hello, World!")
# Receive response
response = handler.receive(timeout_ms=3000)
if response:
print(f"Got response: {response}")
# Handler is automatically closed hereimport urst
# Send a single message (opens and closes connection automatically)
success = urst.send_message('/dev/ttyUSB0', b"Quick message", baudrate=115200)
# Receive a single message
message = urst.receive_message('/dev/ttyUSB0', timeout=5.0, baudrate=115200)- Full URST v0.3.2 specification compliance
- Four-layer architecture (Codec, Transport, Protocol, Handler)
- Automatic fragmentation for messages > 200 bytes with reassembly
- CRC-16-CCITT error detection with automatic retransmission
- COBS encoding for robust frame delimiting
- Stop-and-wait flow control with configurable retries
- Sequence number management with duplicate detection
- Clean Pythonic API with simple
send()andreceive()methods - Context manager support for automatic resource cleanup
- Comprehensive error handling with detailed exception hierarchy
- Full type hints for better IDE support and code quality
- Performance optimized with lookup table CRC and efficient buffers
- Cross-platform support (Windows, Linux, macOS)
- Flexible configuration for timeouts, retries, and buffer sizes
import urst
with urst.create_handler('/dev/ttyUSB0', max_payload_size=100) as handler:
# Large messages are automatically fragmented
large_data = b"x" * 1000 # 1KB message
success = handler.send(large_data) # Automatically split into ~10 fragments
# Fragments are automatically reassembled on receive
received = handler.receive(timeout_ms=10000)
assert received == large_data # Original message reconstructedimport urst
from urst.exceptions import URSTTimeoutError, URSTSerialError, URSTCRCError
try:
with urst.create_handler('/dev/ttyUSB0') as handler:
handler.send(b"Test message")
response = handler.receive(timeout_ms=1000)
except URSTTimeoutError as e:
print(f"Operation timed out: {e}")
except URSTSerialError as e:
print(f"Serial port error: {e}")
except URSTCRCError as e:
print(f"Data corruption detected: {e}")
except urst.URSTError as e:
print(f"General URST error: {e}")import urst
# Create handler with custom settings
handler = urst.create_handler(
port='/dev/ttyUSB0',
baudrate=9600,
timeout=2.0, # 2 second serial timeout
rx_buffer_size=2048, # Larger receive buffer
max_payload_size=150, # Smaller fragmentation threshold
fragment_timeout_ms=15000, # 15 second fragment timeout
max_retries=5, # More retry attempts
)
# Modify settings at runtime
handler.set_fragment_timeout_ms(30000)
handler.set_max_retries(3)URSTHandler: Main class for application use with high-level APIURSTProtocol: Protocol layer (ACK/NAK, retransmission)URSTTransport: Transport layer (frame types, sequence numbers)URSTCodec: Codec layer (COBS encoding, CRC calculation)
- **
create_handler(port, **kwargs)**: Create configured handler - **
send_message(port, data, **kwargs)**: One-shot send operation - **
receive_message(port, timeout, **kwargs)**: One-shot receive operation
URSTError (base)
├── URSTTimeoutError
├── URSTCRCError
├── URSTFramingError
├── URSTSerialError
├── URSTFragmentationError
└── URSTProtocolError
| Parameter | Default | Description |
|---|---|---|
baudrate |
115200 | Serial communication speed |
timeout |
1.0 | Serial read/write timeout (seconds) |
rx_buffer_size |
512 | Receive buffer size (bytes) |
max_payload_size |
200 | Fragmentation threshold (bytes) |
fragment_timeout_ms |
5000 | Fragment reassembly timeout (ms) |
default_timeout_ms |
1000 | Protocol ACK timeout (ms) |
max_retries |
3 | Maximum retransmission attempts |
The examples/ directory contains comprehensive usage examples:
basic_usage.py: Simple send/receive operations and error handlingadvanced_usage.py: Fragmentation, monitoring, and performance testing
Run examples:
python examples/basic_usage.py /dev/ttyUSB0 115200
python examples/advanced_usage.py /dev/ttyUSB0 115200- Python 3.8+
- pyserial 3.5+
This project uses UV for dependency management:
# Clone repository
git clone https://github.com/simonl65/URST-Python.git
cd URST-Python
# Install dependencies
uv sync --dev
# Run tests
uv run pytest
# Run linting
uv run ruff check
uv run mypy src/This code is released under the Sustainable Use License detailed in LICENSE.md.
In essence the license is one of Fair-code. It effectively prohibits you from making profit from this code if I haven't given you a specific license to do so. If I give you such a separate license then I'm happy for us both to profit off this code.
- URST Specification - Protocol specification
- URST MicroPython - MicroPython implementation