Serial Communication

from pyftdi.serialext import serial_for_url

def serial_for_url(url: str, **kwargs):
    """
    Create serial connection from FTDI URL.
    
    Parameters:
    - url: FTDI device URL (e.g., 'ftdi:///1')
    - baudrate: Serial baudrate (default: 9600)
    - bytesize: Data bits (5, 6, 7, 8)
    - parity: Parity setting ('N', 'E', 'O', 'M', 'S')
    - stopbits: Stop bits (1, 1.5, 2)
    - timeout: Read timeout in seconds
    - write_timeout: Write timeout in seconds
    - xonxoff: Software flow control
    - rtscts: Hardware RTS/CTS flow control
    - dsrdtr: Hardware DSR/DTR flow control
    
    Returns:
    Serial: PySerial-compatible serial port object
    
    Raises:
    - SerialException: Device access or configuration error
    """

# PySerial URL registration
import serial

# FTDI URLs automatically handled by PySerial
serial_port = serial.serial_for_url('ftdi:///1', baudrate=115200)

class FtdiSerial:
    """PySerial-compatible FTDI serial interface."""
    
    def read(self, size: int = 1) -> bytes:
        """Read data from serial port."""
    
    def write(self, data: bytes) -> int:
        """Write data to serial port."""
    
    def readline(self, size: int = -1) -> bytes:
        """Read line from serial port."""
    
    def readlines(self, hint: int = -1) -> list:
        """Read multiple lines from serial port."""
    
    def flush(self):
        """Flush write buffer."""
    
    def reset_input_buffer(self):
        """Clear input buffer."""
    
    def reset_output_buffer(self):
        """Clear output buffer."""
    
    def close(self):
        """Close serial port."""

# Baudrate settings
@property
def baudrate(self) -> int:
    """Get/set baudrate."""

@baudrate.setter  
def baudrate(self, value: int):
    """Set baudrate (300-12000000 bps)."""

# Data format settings
@property
def bytesize(self) -> int:
    """Get/set data bits (5, 6, 7, 8)."""

@property
def parity(self) -> str:
    """Get/set parity ('N', 'E', 'O', 'M', 'S')."""

@property
def stopbits(self) -> float:
    """Get/set stop bits (1, 1.5, 2)."""

# Flow control settings
@property
def rtscts(self) -> bool:
    """Get/set RTS/CTS hardware flow control."""

@property
def dsrdtr(self) -> bool:
    """Get/set DSR/DTR hardware flow control."""

@property
def xonxoff(self) -> bool:
    """Get/set XON/XOFF software flow control."""

@property
def rts(self) -> bool:
    """Get/set RTS (Request To Send) line state."""

@property
def dtr(self) -> bool:
    """Get/set DTR (Data Terminal Ready) line state."""

@property  
def cts(self) -> bool:
    """Get CTS (Clear To Send) line state."""

@property
def dsr(self) -> bool:
    """Get DSR (Data Set Ready) line state."""

@property
def ri(self) -> bool:
    """Get RI (Ring Indicator) line state."""

@property
def cd(self) -> bool:
    """Get CD (Carrier Detect) line state."""

@property
def timeout(self) -> float:
    """Get/set read timeout in seconds (None = blocking)."""

@property
def write_timeout(self) -> float:
    """Get/set write timeout in seconds (None = blocking)."""

@property
def inter_byte_timeout(self) -> float:
    """Get/set inter-byte timeout for reads."""

from pyftdi.serialext import serial_for_url

# Open serial connection
serial_port = serial_for_url('ftdi:///1', baudrate=115200, timeout=1)

# Send data
serial_port.write(b'Hello, FTDI!\r\n')

# Read response
response = serial_port.read(100)
print(f"Received: {response}")

# Close connection
serial_port.close()

import serial

# Use FTDI device with standard PySerial
with serial.serial_for_url('ftdi:///1', baudrate=9600, timeout=0.5) as ser:
    ser.write(b'AT\r\n')
    response = ser.readline()
    print(f"Response: {response.decode().strip()}")

from pyftdi.serialext import serial_for_url

ser = serial_for_url('ftdi:///1')

# Configure serial parameters
ser.baudrate = 38400
ser.bytesize = 8
ser.parity = 'N'
ser.stopbits = 1
ser.rtscts = False

# Control DTR and RTS lines
ser.dtr = True   # Assert DTR
ser.rts = False  # Deassert RTS

# Check status lines
print(f"CTS: {ser.cts}")
print(f"DSR: {ser.dsr}")

# Send data with line control
ser.rts = True   # Request to send
ser.write(b'Data with RTS control\r\n')
ser.rts = False  # End transmission

ser.close()

from pyftdi.serialext import serial_for_url
import time

# High-speed serial connection
ser = serial_for_url('ftdi:///1', baudrate=3000000, timeout=0.1)

# Send large data block
data_block = b'X' * 1024  # 1KB test data
start_time = time.time()

ser.write(data_block)
ser.flush()  # Ensure data is sent

elapsed = time.time() - start_time
throughput = len(data_block) / elapsed
print(f"Throughput: {throughput:.0f} bytes/sec")

ser.close()

from pyftdi.serialext import serial_for_url
import struct
import time

class SerialProtocol:
    """Example serial protocol implementation."""
    
    def __init__(self, device_url, baudrate=115200):
        self.ser = serial_for_url(device_url, baudrate=baudrate, timeout=1)
        
    def send_command(self, cmd_id, payload=b''):
        """Send protocol command."""
        # Protocol: STX + CMD + LEN + PAYLOAD + CHECKSUM + ETX
        stx = 0x02
        etx = 0x03
        length = len(payload)
        
        frame = struct.pack('BBB', stx, cmd_id, length) + payload
        checksum = sum(frame) & 0xFF
        frame += struct.pack('BB', checksum, etx)
        
        self.ser.write(frame)
        return self.receive_response()
    
    def receive_response(self):
        """Receive protocol response."""
        # Wait for STX
        while True:
            byte = self.ser.read(1)
            if not byte:
                return None
            if byte[0] == 0x02:  # STX found
                break
        
        # Read header
        header = self.ser.read(2)
        if len(header) != 2:
            return None
            
        cmd_id, length = struct.unpack('BB', header)
        
        # Read payload and trailer
        remaining = self.ser.read(length + 2)  # payload + checksum + ETX
        if len(remaining) != length + 2:
            return None
            
        payload = remaining[:-2]
        checksum, etx = struct.unpack('BB', remaining[-2:])
        
        if etx != 0x03:
            return None
            
        return {'cmd_id': cmd_id, 'payload': payload}
    
    def close(self):
        self.ser.close()

# Usage
protocol = SerialProtocol('ftdi:///1')

# Send ping command
response = protocol.send_command(0x01, b'ping')
if response:
    print(f"Response: {response}")

protocol.close()

from pyftdi.serialext import serial_for_url
import threading
import sys

class SerialTerminal:
    """Simple serial terminal emulator."""
    
    def __init__(self, device_url, baudrate=115200):
        self.ser = serial_for_url(device_url, baudrate=baudrate, timeout=0.1)
        self.running = False
        
    def start(self):
        """Start terminal session."""
        self.running = True
        
        # Start receive thread
        receive_thread = threading.Thread(target=self._receive_loop)
        receive_thread.daemon = True
        receive_thread.start()
        
        print("Serial terminal started. Type 'exit' to quit.")
        
        # Main input loop
        try:
            while self.running:
                line = input()
                if line.lower() == 'exit':
                    break
                self.ser.write((line + '\r\n').encode())
        except KeyboardInterrupt:
            pass
        
        self.running = False
        self.ser.close()
    
    def _receive_loop(self):
        """Receive data from serial port."""
        while self.running:
            data = self.ser.read(100)
            if data:
                try:
                    text = data.decode('utf-8', errors='replace')
                    sys.stdout.write(text)
                    sys.stdout.flush()
                except:
                    pass

# Usage
terminal = SerialTerminal('ftdi:///1', 9600)
terminal.start()

from pyftdi.serialext import serial_for_url
import time
import csv

def log_serial_data(device_url, log_file, duration=60):
    """Log serial data to CSV file."""
    ser = serial_for_url(device_url, baudrate=9600, timeout=0.5)
    
    with open(log_file, 'w', newline='') as csvfile:
        writer = csv.writer(csvfile)
        writer.writerow(['Timestamp', 'Data'])
        
        start_time = time.time()
        
        while time.time() - start_time < duration:
            data = ser.readline()
            if data:
                timestamp = time.strftime('%Y-%m-%d %H:%M:%S')
                data_str = data.decode('utf-8', errors='replace').strip()
                writer.writerow([timestamp, data_str])
                print(f"{timestamp}: {data_str}")
    
    ser.close()
    print(f"Logging complete. Data saved to {log_file}")

# Usage
log_serial_data('ftdi:///1', 'serial_log.csv', duration=300)  # 5 minutes

from pyftdi.serialext import serial_for_url
from serial import SerialException, SerialTimeoutException

try:
    ser = serial_for_url('ftdi:///1', baudrate=115200, timeout=1)
    
    ser.write(b'Test data\r\n')
    response = ser.read(100)
    
    print(f"Response: {response}")
    
except SerialException as e:
    print(f"Serial communication error: {e}")
except SerialTimeoutException as e:
    print(f"Serial timeout: {e}")
finally:
    if 'ser' in locals():
        ser.close()

# Import PySerial exceptions
from serial import (
    SerialException,
    SerialTimeoutException,
    PortNotOpenError
)

# Parity constants
PARITY_NONE = 'N'
PARITY_EVEN = 'E'
PARITY_ODD = 'O'
PARITY_MARK = 'M'
PARITY_SPACE = 'S'

# Stop bit constants
STOPBITS_ONE = 1
STOPBITS_ONE_POINT_FIVE = 1.5
STOPBITS_TWO = 2

# Bytesize constants
FIVEBITS = 5
SIXBITS = 6
SEVENBITS = 7  
EIGHTBITS = 8