PySerial Configuration Guide

import serial

ser = serial.Serial(
    port='/dev/ttyUSB0',           # Device path
    baudrate=9600,                 # Communication speed
    bytesize=serial.EIGHTBITS,     # Data bits (5, 6, 7, 8)
    parity=serial.PARITY_NONE,     # Parity checking
    stopbits=serial.STOPBITS_ONE,  # Stop bits (1, 1.5, 2)
    timeout=1,                     # Read timeout in seconds
    xonxoff=False,                 # Software flow control
    rtscts=False,                  # Hardware (RTS/CTS) flow control
    write_timeout=None,            # Write timeout in seconds
    dsrdtr=False,                  # DSR/DTR flow control
    inter_byte_timeout=None,       # Inter-byte timeout
    exclusive=None                 # Exclusive access (Linux)
)

# Standard baud rates (widely supported)
STANDARD_BAUDS = [
    1200, 2400, 4800, 9600, 19200, 38400, 57600, 115200
]

# High-speed rates (platform/hardware dependent)
HIGH_SPEED_BAUDS = [
    230400, 460800, 921600, 1000000, 1500000, 2000000
]

# Test baud rate compatibility
def test_baud_rate(port, baudrate):
    """Test if baud rate works with device"""
    try:
        test_ser = serial.Serial(port, baudrate, timeout=1)
        
        # Send test command
        test_ser.write(b'AT\r\n')
        response = test_ser.readline()
        
        test_ser.close()
        
        return len(response) > 0
    except:
        return False

# Find optimal baud rate
def find_optimal_baudrate(port, test_bauds=None):
    """Find highest working baud rate"""
    if test_bauds is None:
        test_bauds = STANDARD_BAUDS + HIGH_SPEED_BAUDS
    
    working_bauds = []
    
    for baud in sorted(test_bauds):
        if test_baud_rate(port, baud):
            working_bauds.append(baud)
            print(f"✅ {baud} baud: OK")
        else:
            print(f"❌ {baud} baud: Failed")
    
    return max(working_bauds) if working_bauds else None

def calculate_transfer_time(data_size, baudrate, data_bits=8, parity_bits=0, stop_bits=1):
    """Calculate time to transfer data"""
    
    # Total bits per byte
    bits_per_byte = 1 + data_bits + parity_bits + stop_bits  # 1 start bit
    
    # Total bits to transfer
    total_bits = data_size * bits_per_byte
    
    # Transfer time in seconds
    transfer_time = total_bits / baudrate
    
    return {
        'bits_per_byte': bits_per_byte,
        'total_bits': total_bits,
        'transfer_time': transfer_time,
        'bytes_per_second': baudrate / bits_per_byte
    }

# Example calculation
stats = calculate_transfer_time(1024, 9600)  # 1KB at 9600 baud
print(f"1KB transfer time at 9600 baud: {stats['transfer_time']:.2f} seconds")
print(f"Effective throughput: {stats['bytes_per_second']:.0f} bytes/second")

# Data bits options
ser.bytesize = serial.FIVEBITS    # 5 bits (rare)
ser.bytesize = serial.SIXBITS     # 6 bits (rare)
ser.bytesize = serial.SEVENBITS   # 7 bits (ASCII)
ser.bytesize = serial.EIGHTBITS   # 8 bits (most common)

# Or use integers
ser.bytesize = 8  # Same as EIGHTBITS

# Parity options
ser.parity = serial.PARITY_NONE   # No parity (most common)
ser.parity = serial.PARITY_EVEN   # Even parity
ser.parity = serial.PARITY_ODD    # Odd parity
ser.parity = serial.PARITY_MARK   # Always 1
ser.parity = serial.PARITY_SPACE  # Always 0

# Or use characters
ser.parity = 'N'  # None
ser.parity = 'E'  # Even
ser.parity = 'O'  # Odd
ser.parity = 'M'  # Mark
ser.parity = 'S'  # Space

# Parity checking function
def check_parity(data_byte, parity_type):
    """Check if parity bit is correct"""
    bit_count = bin(data_byte).count('1')
    
    if parity_type == 'E':  # Even
        return bit_count % 2 == 0
    elif parity_type == 'O':  # Odd
        return bit_count % 2 == 1
    elif parity_type == 'M':  # Mark (always 1)
        return True
    elif parity_type == 'S':  # Space (always 0)
        return False
    else:  # None
        return True

# Stop bits options
ser.stopbits = serial.STOPBITS_ONE             # 1 stop bit (most common)
ser.stopbits = serial.STOPBITS_ONE_POINT_FIVE  # 1.5 stop bits (rare)
ser.stopbits = serial.STOPBITS_TWO             # 2 stop bits

# Or use numbers
ser.stopbits = 1    # 1 stop bit
ser.stopbits = 1.5  # 1.5 stop bits
ser.stopbits = 2    # 2 stop bits

def validate_format(data_bits, parity, stop_bits):
    """Validate serial format combination"""
    
    valid_data_bits = [5, 6, 7, 8]
    valid_parity = ['N', 'E', 'O', 'M', 'S']
    valid_stop_bits = [1, 1.5, 2]
    
    errors = []
    
    if data_bits not in valid_data_bits:
        errors.append(f"Invalid data bits: {data_bits}")
    
    if parity not in valid_parity:
        errors.append(f"Invalid parity: {parity}")
    
    if stop_bits not in valid_stop_bits:
        errors.append(f"Invalid stop bits: {stop_bits}")
    
    # Special combinations
    if data_bits == 5 and stop_bits == 1.5:
        errors.append("5 data bits with 1.5 stop bits not recommended")
    
    return len(errors) == 0, errors

# Example usage
is_valid, errors = validate_format(8, 'N', 1)
if is_valid:
    print("✅ Valid format: 8N1")
else:
    for error in errors:
        print(f"❌ {error}")

# Enable software flow control
ser = serial.Serial(
    port='/dev/ttyUSB0',
    baudrate=9600,
    xonxoff=True  # Enable XON/XOFF flow control
)

# XON/XOFF control characters
XON = b'\x11'   # Resume transmission (Ctrl+Q)
XOFF = b'\x13'  # Pause transmission (Ctrl+S)

# Manual XON/XOFF handling (if xonxoff=False)
def manual_flow_control_example(ser):
    """Example of manual flow control handling"""
    
    while True:
        # Check for incoming data
        if ser.in_waiting:
            data = ser.read(1)
            
            if data == XOFF:
                print("⏸️  Received XOFF - pausing transmission")
                # Pause sending data
                continue
            elif data == XON:
                print("▶️  Received XON - resuming transmission")
                # Resume sending data
                continue
            else:
                # Process normal data
                process_data(data)

# Enable hardware flow control
ser = serial.Serial(
    port='/dev/ttyUSB0',
    baudrate=9600,
    rtscts=True  # Enable RTS/CTS flow control
)

# Manual RTS/CTS control
class RTSCTSControl:
    def __init__(self, ser):
        self.ser = ser
    
    def wait_for_clear_to_send(self, timeout=5):
        """Wait for CTS signal"""
        start_time = time.time()
        
        while not self.ser.cts:
            if time.time() - start_time > timeout:
                raise TimeoutError("CTS timeout")
            time.sleep(0.01)
        
        return True
    
    def controlled_write(self, data):
        """Write data with CTS checking"""
        # Request to send
        self.ser.rts = True
        
        try:
            # Wait for clear to send
            self.wait_for_clear_to_send()
            
            # Send data
            bytes_written = self.ser.write(data)
            self.ser.flush()
            
            return bytes_written
            
        finally:
            # Clear request to send
            self.ser.rts = False

# Usage
rts_cts = RTSCTSControl(ser)
rts_cts.controlled_write(b"Data with flow control")

# Enable DSR/DTR flow control
ser = serial.Serial(
    port='/dev/ttyUSB0',
    baudrate=9600,
    dsrdtr=True  # Enable DSR/DTR flow control
)

# Manual DTR control
def modem_control_example(ser):
    """Example of DTR/DSR modem control"""
    
    # Assert DTR (Data Terminal Ready)
    ser.dtr = True
    print("✅ DTR asserted")
    
    # Wait for DSR (Data Set Ready)
    timeout = 10
    start_time = time.time()
    
    while not ser.dsr:
        if time.time() - start_time > timeout:
            print("❌ DSR timeout")
            return False
        time.sleep(0.1)
    
    print("✅ DSR detected - modem ready")
    
    # Check carrier detect
    if ser.cd:
        print("✅ Carrier detected")
    
    # Check ring indicator
    if ser.ri:
        print("📞 Ring indicator active")
    
    return True

# Usage
if modem_control_example(ser):
    print("Modem handshake successful")

# Set buffer sizes (platform dependent)
try:
    ser.set_buffer_size(rx_size=8192, tx_size=4096)
    print("✅ Buffer sizes set")
except AttributeError:
    print("ℹ️  Buffer size control not available on this platform")

# Monitor buffer usage
def monitor_buffers(ser, duration=10):
    """Monitor buffer usage over time"""
    start_time = time.time()
    max_input_waiting = 0
    
    while time.time() - start_time < duration:
        waiting = ser.in_waiting
        max_input_waiting = max(max_input_waiting, waiting)
        
        if waiting > 0:
            print(f"📊 Input buffer: {waiting} bytes")
        
        time.sleep(1)
    
    print(f"📈 Maximum buffer usage: {max_input_waiting} bytes")

class BufferManager:
    """Advanced buffer management"""
    
    def __init__(self, ser, warning_threshold=0.8, critical_threshold=0.95):
        self.ser = ser
        self.warning_threshold = warning_threshold
        self.critical_threshold = critical_threshold
        self.buffer_size = 4096  # Assume default size
        
    def check_buffer_health(self):
        """Check buffer status and warn if needed"""
        waiting = self.ser.in_waiting
        usage_ratio = waiting / self.buffer_size
        
        if usage_ratio > self.critical_threshold:
            print(f"🚨 CRITICAL: Input buffer {usage_ratio:.1%} full ({waiting} bytes)")
            return 'critical'
        elif usage_ratio > self.warning_threshold:
            print(f"⚠️  WARNING: Input buffer {usage_ratio:.1%} full ({waiting} bytes)")
            return 'warning'
        else:
            return 'ok'
    
    def emergency_clear(self):
        """Clear buffers in emergency"""
        print("🗑️  Emergency buffer clear")
        self.ser.reset_input_buffer()
        self.ser.reset_output_buffer()
    
    def smart_read(self, target_size):
        """Read with buffer management"""
        status = self.check_buffer_health()
        
        if status == 'critical':
            self.emergency_clear()
            return b''
        
        # Read appropriate amount based on buffer status
        available = self.ser.in_waiting
        to_read = min(target_size, available)
        
        return self.ser.read(to_read)

# Usage
buffer_mgr = BufferManager(ser)
data = buffer_mgr.smart_read(1024)

# Exclusive access (Linux only)
try:
    ser = serial.Serial(
        port='/dev/ttyUSB0',
        baudrate=9600,
        exclusive=True  # Prevent other processes from using port
    )
    print("✅ Exclusive access granted")
except ValueError:
    print("⚠️  Exclusive access not supported on this platform")
except serial.SerialException as e:
    print(f"❌ Exclusive access failed: {e}")

class SerialProfiles:
    """Predefined configuration profiles"""
    
    # Common device profiles
    PROFILES = {
        'arduino': {
            'baudrate': 9600,
            'bytesize': 8,
            'parity': 'N',
            'stopbits': 1,
            'timeout': 2,
            'write_timeout': 2
        },
        'gps': {
            'baudrate': 4800,
            'bytesize': 8,
            'parity': 'N',
            'stopbits': 1,
            'timeout': 1,
            'inter_byte_timeout': 0.1
        },
        'modem': {
            'baudrate': 115200,
            'bytesize': 8,
            'parity': 'N',
            'stopbits': 1,
            'timeout': 5,
            'rtscts': True
        },
        'industrial': {
            'baudrate': 9600,
            'bytesize': 8,
            'parity': 'E',  # Even parity for error detection
            'stopbits': 1,
            'timeout': 1,
            'write_timeout': 1
        }
    }
    
    @classmethod
    def create_serial(cls, port, profile_name):
        """Create serial connection with profile"""
        if profile_name not in cls.PROFILES:
            raise ValueError(f"Unknown profile: {profile_name}")
        
        config = cls.PROFILES[profile_name].copy()
        config['port'] = port
        
        return serial.Serial(**config)
    
    @classmethod
    def list_profiles(cls):
        """List available profiles"""
        for name, config in cls.PROFILES.items():
            print(f"{name}: {config['baudrate']} baud, "
                  f"{config['bytesize']}{config['parity']}{config['stopbits']}")

# Usage
SerialProfiles.list_profiles()
arduino = SerialProfiles.create_serial('/dev/ttyACM0', 'arduino')
gps = SerialProfiles.create_serial('/dev/ttyUSB0', 'gps')

def validate_serial_config(port, **config):
    """Validate serial configuration before connecting"""
    
    errors = []
    warnings = []
    
    # Check port existence
    if not port:
        errors.append("Port not specified")
    else:
        # Check if port exists
        available_ports = [p.device for p in serial.tools.list_ports.comports()]
        if port not in available_ports:
            warnings.append(f"Port {port} not found in available ports")
    
    # Validate baud rate
    baudrate = config.get('baudrate', 9600)
    if baudrate <= 0:
        errors.append("Baud rate must be positive")
    elif baudrate > 2000000:
        warnings.append(f"Very high baud rate: {baudrate}")
    
    # Validate timeouts
    timeout = config.get('timeout')
    if timeout is not None and timeout < 0:
        errors.append("Timeout cannot be negative")
    
    # Check for conflicting flow control
    xonxoff = config.get('xonxoff', False)
    rtscts = config.get('rtscts', False)
    dsrdtr = config.get('dsrdtr', False)
    
    flow_control_count = sum([xonxoff, rtscts, dsrdtr])
    if flow_control_count > 1:
        warnings.append("Multiple flow control methods enabled")
    
    return {
        'valid': len(errors) == 0,
        'errors': errors,
        'warnings': warnings
    }

# Usage
validation = validate_serial_config(
    port='/dev/ttyUSB0',
    baudrate=115200,
    timeout=1,
    rtscts=True
)

if validation['valid']:
    print("✅ Configuration valid")
    for warning in validation['warnings']:
        print(f"⚠️  {warning}")
else:
    print("❌ Configuration invalid:")
    for error in validation['errors']:
        print(f"  - {error}")

def test_serial_configuration(port, **config):
    """Test serial configuration with actual hardware"""
    
    print(f"🧪 Testing configuration on {port}")
    print(f"   Settings: {config}")
    
    try:
        # Test connection
        ser = serial.Serial(port, **config)
        print("✅ Port opened successfully")
        
        # Test basic communication
        ser.write(b'AT\r\n')
        ser.flush()
        
        response = ser.read(100)
        
        if response:
            print(f"✅ Communication test: received {len(response)} bytes")
            print(f"   Response: {response[:50]}...")  # First 50 bytes
        else:
            print("⚠️  No response received (may be normal)")
        
        # Test buffer status
        waiting = ser.in_waiting
        print(f"ℹ️  Buffer status: {waiting} bytes waiting")
        
        ser.close()
        print("✅ Port closed successfully")
        
        return True
        
    except Exception as e:
        print(f"❌ Configuration test failed: {e}")
        return False

# Usage
success = test_serial_configuration(
    '/dev/ttyUSB0',
    baudrate=9600,
    timeout=2,
    parity='N'
)