PySerial on Windows: Complete Setup Guide

def diagnose_driver_issues():
    """Diagnose common Windows driver problems"""
    
    print("🔍 Windows Driver Diagnostic")
    print("=" * 40)
    
    import serial.tools.list_ports
    import platform
    
    # Check Windows version
    win_version = platform.win32_ver()
    print(f"Windows Version: {win_version[0]} {win_version[1]}")
    
    # Check available ports
    ports = list(serial.tools.list_ports.comports())
    print(f"\nFound {len(ports)} serial port(s):")
    
    if not ports:
        print("❌ No serial ports found!")
        print("\n💡 Troubleshooting steps:")
        print("   1. Check Device Manager for 'Other devices'")
        print("   2. Install proper drivers")
        print("   3. Try different USB port")
        print("   4. Check cable integrity")
        return
    
    # Analyze each port
    for port in ports:
        print(f"\n📍 {port.device}: {port.description}")
        
        # Check for common driver issues
        desc_lower = port.description.lower()
        
        if 'prolific' in desc_lower:
            print("   ⚠️  Prolific chip detected")
            print("   💡 Known issues with clone chips")
            print("   🔧 Try official Prolific drivers")
        
        elif 'ch340' in desc_lower or 'ch341' in desc_lower:
            print("   ℹ️  CH340/341 chip detected")
            print("   💡 Usually works with basic drivers")
        
        elif 'ftdi' in desc_lower:
            print("   ✅ FTDI chip detected (most reliable)")
        
        # Test the port
        try:
            test_ser = serial.Serial(port.device, 9600, timeout=0.1)
            print(f"   ✅ Port opens successfully")
            test_ser.close()
        except Exception as e:
            print(f"   ❌ Port test failed: {e}")

# Run diagnostic
diagnose_driver_issues()

import winreg
import serial.tools.list_ports

def get_com_port_registry_info(com_port):
    """Get COM port info from Windows registry"""
    
    try:
        # Open registry key
        key_path = r"HARDWARE\DEVICEMAP\SERIALCOMM"
        key = winreg.OpenKey(winreg.HKEY_LOCAL_MACHINE, key_path)
        
        # Enumerate values
        registry_info = {}
        i = 0
        
        try:
            while True:
                name, value, reg_type = winreg.EnumValue(key, i)
                if value == com_port:
                    registry_info['device_path'] = name
                    registry_info['com_port'] = value
                    break
                i += 1
        except WindowsError:
            pass
        
        winreg.CloseKey(key)
        return registry_info
        
    except WindowsError as e:
        return {'error': str(e)}

# Usage
for port in serial.tools.list_ports.comports():
    reg_info = get_com_port_registry_info(port.device)
    print(f"{port.device}: {reg_info}")

def configure_windows_buffers(ser, rx_buffer=4096, tx_buffer=2048):
    """Configure Windows-specific buffer settings"""
    
    try:
        # Try to set buffer sizes
        ser.set_buffer_size(rx_size=rx_buffer, tx_size=tx_buffer)
        print(f"✅ Buffers set: RX={rx_buffer}, TX={tx_buffer}")
    except AttributeError:
        print("ℹ️  Buffer size control not available")
    
    # Windows-specific timeout handling
    if hasattr(ser, '_timeout'):
        print(f"Read timeout: {ser.timeout}s")
    
    # Check Windows COM port properties
    try:
        import serial.win32
        
        # Get DCB (Device Control Block)
        dcb = serial.win32.DCB()
        success = serial.win32.GetCommState(ser._port_handle, dcb)
        
        if success:
            print(f"DCB Settings:")
            print(f"  Baud Rate: {dcb.BaudRate}")
            print(f"  Byte Size: {dcb.ByteSize}")
            print(f"  Parity: {dcb.Parity}")
            print(f"  Stop Bits: {dcb.StopBits}")
            
    except (ImportError, AttributeError):
        print("⚠️  Windows-specific settings not accessible")

class WindowsSerialOptimizer:
    """Optimize PySerial for Windows performance"""
    
    def __init__(self, ser):
        self.ser = ser
        self.optimizations_applied = []
    
    def optimize_for_high_speed(self):
        """Apply high-speed optimizations"""
        
        # 1. Disable Windows power management
        self.disable_power_management()
        
        # 2. Set high-priority buffer sizes
        try:
            self.ser.set_buffer_size(rx_size=8192, tx_size=4096)
            self.optimizations_applied.append("Large buffers")
        except:
            pass
        
        # 3. Minimize timeouts for high-speed data
        if self.ser.baudrate > 115200:
            self.ser.timeout = 0.1
            self.ser.write_timeout = 0.1
            self.optimizations_applied.append("Reduced timeouts")
        
        # 4. Disable flow control if not needed
        if not self.requires_flow_control():
            self.ser.xonxoff = False
            self.ser.rtscts = False
            self.ser.dsrdtr = False
            self.optimizations_applied.append("Disabled flow control")
        
        print(f"✅ Applied optimizations: {', '.join(self.optimizations_applied)}")
    
    def disable_power_management(self):
        """Disable USB power management (requires admin rights)"""
        try:
            import subprocess
            
            # Disable USB selective suspend
            cmd = [
                'powercfg', '/setacvalueindex', 'SCHEME_CURRENT',
                '2a737441-1930-4402-8d77-b2bebba308a3',
                '48e6b7a6-50f5-4782-a5d4-53bb8f07e226',
                '0'
            ]
            
            result = subprocess.run(cmd, capture_output=True, text=True)
            
            if result.returncode == 0:
                self.optimizations_applied.append("USB power management")
            
        except Exception as e:
            print(f"⚠️  Could not disable power management: {e}")
    
    def requires_flow_control(self):
        """Check if device likely needs flow control"""
        # Heuristics based on baud rate and buffer usage
        return self.ser.baudrate > 460800 or self.ser.in_waiting > 1024

# Usage
optimizer = WindowsSerialOptimizer(ser)
optimizer.optimize_for_high_speed()

import threading
import queue
import time

class WindowsSerialThread:
    """Windows-optimized threaded serial reading"""
    
    def __init__(self, port, baudrate=9600):
        self.ser = serial.Serial(port, baudrate, timeout=0.1)
        self.data_queue = queue.Queue(maxsize=1000)
        self.running = False
        self.thread = None
        
        # Windows-specific: Set thread priority
        import win32api
        import win32process
        import win32con
        
        self.process_handle = win32api.GetCurrentProcess()
    
    def start_reading(self):
        """Start high-priority reading thread"""
        self.running = True
        self.thread = threading.Thread(target=self._read_loop)
        self.thread.daemon = True
        
        # Set high thread priority on Windows
        self.thread.start()
        
        try:
            import win32api
            import win32process
            import win32con
            
            thread_handle = win32api.OpenThread(
                win32con.THREAD_SET_INFORMATION,
                False,
                self.thread.ident
            )
            
            win32process.SetThreadPriority(
                thread_handle,
                win32process.THREAD_PRIORITY_ABOVE_NORMAL
            )
            
            win32api.CloseHandle(thread_handle)
            print("✅ High-priority thread started")
            
        except ImportError:
            print("ℹ️  pywin32 not available - using normal priority")
    
    def _read_loop(self):
        """High-performance read loop"""
        while self.running:
            try:
                if self.ser.in_waiting:
                    data = self.ser.read(self.ser.in_waiting)
                    if data:
                        try:
                            self.data_queue.put(data, block=False)
                        except queue.Full:
                            # Drop oldest data if queue full
                            try:
                                self.data_queue.get(block=False)
                                self.data_queue.put(data, block=False)
                            except queue.Empty:
                                pass
                else:
                    time.sleep(0.001)  # 1ms sleep when no data
                    
            except Exception as e:
                if self.running:
                    print(f"Read error: {e}")
    
    def get_data(self, timeout=0.1):
        """Get data from queue"""
        try:
            return self.data_queue.get(timeout=timeout)
        except queue.Empty:
            return None
    
    def stop(self):
        """Stop reading thread"""
        self.running = False
        if self.thread:
            self.thread.join(timeout=1)
        self.ser.close()

# Usage
reader = WindowsSerialThread('COM3', 115200)
reader.start_reading()

try:
    while True:
        data = reader.get_data()
        if data:
            print(f"Received: {len(data)} bytes")
        time.sleep(0.01)
finally:
    reader.stop()

def windows_serial_diagnostic():
    """Comprehensive Windows serial diagnostic"""
    
    print("🖥️  Windows PySerial Diagnostic Tool")
    print("=" * 45)
    
    import platform
    import serial
    import serial.tools.list_ports
    
    # System info
    print(f"Windows Version: {platform.win32_ver()}")
    print(f"Python Version: {platform.python_version()}")
    print(f"PySerial Version: {serial.__version__}")
    
    # Check ports
    ports = list(serial.tools.list_ports.comports())
    print(f"\nSerial Ports Found: {len(ports)}")
    
    if not ports:
        print("❌ No serial ports detected!")
        print("\n🔍 Troubleshooting checklist:")
        print("   □ Check Device Manager for 'Other devices'")
        print("   □ Install/update drivers")
        print("   □ Try different USB port/cable")
        print("   □ Restart computer")
        return
    
    # Test each port
    for i, port in enumerate(ports, 1):
        print(f"\n📍 Port {i}: {port.device}")
        print(f"   Description: {port.description}")
        print(f"   Manufacturer: {getattr(port, 'manufacturer', 'Unknown')}")
        
        if hasattr(port, 'vid') and port.vid:
            print(f"   VID:PID: {port.vid:04X}:{port.pid:04X}")
        
        # Test opening port
        try:
            test_ser = serial.Serial(port.device, 9600, timeout=0.1)
            print(f"   ✅ Opens successfully")
            
            # Test basic operations
            test_ser.write(b'test')
            test_ser.flush()
            test_ser.read(10)  # Try to read
            
            test_ser.close()
            print(f"   ✅ Read/write test passed")
            
        except PermissionError:
            print(f"   ❌ Permission denied (try running as Administrator)")
        except serial.SerialException as e:
            print(f"   ❌ Serial error: {e}")
        except Exception as e:
            print(f"   ⚠️  Unexpected error: {e}")
    
    # Windows-specific checks
    try:
        import winreg
        print(f"\n🔍 Registry check:")
        
        key = winreg.OpenKey(winreg.HKEY_LOCAL_MACHINE, 
                            r"HARDWARE\DEVICEMAP\SERIALCOMM")
        
        i = 0
        reg_ports = []
        try:
            while True:
                name, value, _ = winreg.EnumValue(key, i)
                reg_ports.append(value)
                i += 1
        except WindowsError:
            pass
        
        winreg.CloseKey(key)
        print(f"   Registry shows {len(reg_ports)} COM ports: {reg_ports}")
        
    except Exception as e:
        print(f"   ⚠️  Registry check failed: {e}")
    
    print(f"\n✅ Diagnostic complete")

# Run the diagnostic
if __name__ == "__main__":
    windows_serial_diagnostic()