serial_reader.py

"""
Serial port communication module for the Qt Signal Analyzer.

This module handles communication with hardware devices via serial ports.
"""

import serial
import serial.tools.list_ports
from typing import List, Optional, Callable
import json
from datetime import datetime
from models import SignalReading
from PyQt5.QtCore import QThread, pyqtSignal


class SerialReader(QThread):
    """Thread for reading data from serial port."""
    
    # Signals
    reading_received = pyqtSignal(object)  # SignalReading
    error_occurred = pyqtSignal(str)  # Error message
    connection_status = pyqtSignal(bool)  # Connected/Disconnected
    
    def __init__(self, port: str, baudrate: int = 9600):
        """
        Initialize the serial reader.
        
        Args:
            port: Serial port name (e.g., 'COM3' or '/dev/ttyUSB0')
            baudrate: Baud rate for serial communication
        """
        super().__init__()
        self.port = port
        self.baudrate = baudrate
        self.serial_connection: Optional[serial.Serial] = None
        self.running = False
    
    def run(self):
        """Main thread execution loop."""
        try:
            # Open serial connection
            self.serial_connection = serial.Serial(
                port=self.port,
                baudrate=self.baudrate,
                timeout=1.0
            )
            self.running = True
            self.connection_status.emit(True)
            
            # Read loop
            while self.running:
                try:
                    if self.serial_connection.in_waiting > 0:
                        line = self.serial_connection.readline().decode('utf-8', errors='ignore').strip()
                        if line:
                            reading = self.parse_json_frame(line)
                            if reading:
                                self.reading_received.emit(reading)
                except UnicodeDecodeError as e:
                    self.error_occurred.emit(f"Decode error: {str(e)}")
                except Exception as e:
                    self.error_occurred.emit(f"Read error: {str(e)}")
        
        except Exception as e:
            self.error_occurred.emit(f"Connection error: {str(e)}")
            self.connection_status.emit(False)
        
        finally:
            if self.serial_connection and self.serial_connection.is_open:
                self.serial_connection.close()
            self.connection_status.emit(False)
    
    def parse_json_frame(self, json_str: str) -> Optional[SignalReading]:
        """
        Parse a JSON frame into a SignalReading object.
        
        Expected JSON format:
        {
            "voltage": 5.0,
            "current": 1.2,
            "frequency": 60.0
        }
        
        Args:
            json_str: JSON string to parse
            
        Returns:
            SignalReading object or None if parsing fails
        """
        try:
            data = json.loads(json_str)
            
            # Validate required fields
            required_fields = ['voltage', 'current', 'frequency']
            if not all(field in data for field in required_fields):
                self.error_occurred.emit(f"Missing required fields in JSON: {json_str}")
                return None
            
            reading = SignalReading(
                timestamp=datetime.now(),
                voltage=float(data['voltage']),
                current=float(data['current']),
                frequency=float(data['frequency'])
            )
            
            return reading
        
        except json.JSONDecodeError as e:
            self.error_occurred.emit(f"JSON parse error: {str(e)}")
            return None
        except (ValueError, KeyError) as e:
            self.error_occurred.emit(f"Data format error: {str(e)}")
            return None
    
    def stop(self):
        """Stop the serial reading thread."""
        self.running = False


class MockSerialReader(QThread):
    """Mock serial reader for testing without hardware."""
    
    # Signals
    reading_received = pyqtSignal(object)  # SignalReading
    error_occurred = pyqtSignal(str)  # Error message
    connection_status = pyqtSignal(bool)  # Connected/Disconnected
    
    def __init__(self):
        """Initialize the mock serial reader."""
        super().__init__()
        self.running = False
        self.counter = 0
    
    def run(self):
        """Generate mock data."""
        import time
        import random
        import math
        
        self.running = True
        self.connection_status.emit(True)
        
        try:
            while self.running:
                time.sleep(0.1)  # 10 Hz update rate
                
                # Generate realistic-looking data with some variation
                self.counter += 1
                t = self.counter * 0.1
                
                voltage = 5.0 + 0.5 * math.sin(t) + random.uniform(-0.1, 0.1)
                current = 1.0 + 0.3 * math.cos(t * 1.5) + random.uniform(-0.05, 0.05)
                frequency = 60.0 + 2.0 * math.sin(t * 0.5) + random.uniform(-0.5, 0.5)
                
                reading = SignalReading(
                    timestamp=datetime.now(),
                    voltage=voltage,
                    current=current,
                    frequency=frequency
                )
                
                self.reading_received.emit(reading)
        
        finally:
            self.connection_status.emit(False)
    
    def stop(self):
        """Stop the mock reader."""
        self.running = False


def list_available_ports() -> List[str]:
    """
    List all available serial ports.
    
    Returns:
        List of port names
    """
    ports = serial.tools.list_ports.comports()
    return [port.device for port in ports]