import threading import time from deadbands import DeadbandController try: import gpiod GPIO_AVAILABLE = True print("gpiod library loaded successfully - GPIO functions enabled") except ImportError: GPIO_AVAILABLE = False print("Warning: gpiod not available, GPIO functions disabled") class GPIOController: """Handles all GPIO operations for the CSEM Triggerbox""" def __init__(self, status_callback=None): self.status_callback = status_callback or (lambda msg: print(f"[GPIO] {msg}")) # GPIO pins self.tx_status_pin = 13 # GPIO 13 (Pin 33) - HIGH when transmitting self.frequency_pin = 6 # GPIO 6 (Pin 31) - toggles at 2x frequency self.gpio26_pin = 26 # GPIO 26 - toggles at selected frequency self.pps_pin = 17 # GPIO 17 (Pin 11) - PPS input # GPIO state tracking self.gpio_state = False self.gpio26_state = False self.pps_state = False # Threading control self.transmitting = False self.frequency_output_thread = None self.gpio26_thread = None self.pps_monitoring_thread = None self.pps_monitoring = False # GPIO resources self.gpio_chip = None self.tx_status_line = None self.frequency_line = None self.gpio26_line = None self.pps_line = None self.gpio_available = GPIO_AVAILABLE # PPS callback self.pps_callback = None # Initialize deadband controller self.deadband_controller = DeadbandController() # CUSTOMIZE DEADBANDS HERE - Change these values to adjust timing: self.deadband_controller.set_duty_deadbands(rising_ms=5, falling_ms=5) self.deadband_controller.set_polarity_deadbands(rising_ms=5, falling_ms=5) self._init_gpio() def _init_gpio(self): """Initialize GPIO pins""" if not self.gpio_available: self.status_callback("GPIO not available - PPS will be simulated") return try: self.gpio_chip = gpiod.Chip('gpiochip4') # Pi 5 uses gpiochip4 # Initialize TX status pin (GPIO 13) self.tx_status_line = self.gpio_chip.get_line(self.tx_status_pin) self.tx_status_line.request(consumer="tx_status", type=gpiod.LINE_REQ_DIR_OUT) self.tx_status_line.set_value(0) # Start LOW # Initialize frequency output pin (GPIO 6) self.frequency_line = self.gpio_chip.get_line(self.frequency_pin) self.frequency_line.request(consumer="frequency_output", type=gpiod.LINE_REQ_DIR_OUT) self.frequency_line.set_value(0) # Start LOW # Initialize GPIO 26 pin self.gpio26_line = self.gpio_chip.get_line(self.gpio26_pin) self.gpio26_line.request(consumer="gpio26_output", type=gpiod.LINE_REQ_DIR_OUT) self.gpio26_line.set_value(0) # Start LOW # Initialize PPS input pin (GPIO 17) self.pps_line = self.gpio_chip.get_line(self.pps_pin) self.pps_line.request(consumer="pps_input", type=gpiod.LINE_REQ_DIR_IN) self.status_callback(f"GPIO initialized successfully - TX Status: {self.tx_status_pin}, Frequency: {self.frequency_pin}, GPIO26: {self.gpio26_pin}, PPS: {self.pps_pin}") except Exception as e: self.status_callback(f"Failed to initialize GPIO: {e} - PPS will be simulated") self.gpio_available = False def start_pps_monitoring(self, pps_callback): """ Start monitoring PPS signal on GPIO 17 Args: pps_callback: Function to call with PPS state changes (True/False) """ if self.pps_monitoring: return # Already monitoring self.pps_callback = pps_callback self.pps_monitoring = True # Start PPS monitoring thread self.pps_monitoring_thread = threading.Thread( target=self._pps_monitoring_loop, daemon=True ) self.pps_monitoring_thread.start() if self.gpio_available and self.pps_line: self.status_callback(f"Started REAL PPS monitoring on GPIO {self.pps_pin}") else: self.status_callback(f"Started SIMULATED PPS monitoring (GPIO not available)") def stop_pps_monitoring(self): """Stop PPS monitoring""" self.pps_monitoring = False if self.pps_monitoring_thread and self.pps_monitoring_thread.is_alive(): self.pps_monitoring_thread.join(timeout=0.5) self.status_callback("Stopped PPS monitoring") def _pps_monitoring_loop(self): """PPS monitoring loop running in separate thread""" last_pps_state = False while self.pps_monitoring: if self.gpio_available and self.pps_line: try: current_pps_state = bool(self.pps_line.get_value()) # Only call callback if state changed if current_pps_state != last_pps_state: self.pps_state = current_pps_state if self.pps_callback: self.pps_callback(current_pps_state) last_pps_state = current_pps_state except Exception as e: self.status_callback(f"PPS monitoring error: {e}") break else: # Simulate PPS for testing without GPIO # Toggle every 1 second to simulate 1 PPS current_time = time.time() simulated_pps = (int(current_time) % 2) == 0 if simulated_pps != last_pps_state: self.pps_state = simulated_pps if self.pps_callback: self.pps_callback(simulated_pps) last_pps_state = simulated_pps self.status_callback(f"PPS (simulated): {'HIGH' if simulated_pps else 'LOW'}") time.sleep(0.01) # 10ms polling rate def start_transmission(self, frequency, indicator_callbacks=None): """ Start GPIO transmission at specified frequency Args: frequency: Transmission frequency in Hz indicator_callbacks: Dict with 'onoff', 'duty', 'polarity' callback functions """ if frequency <= 0: raise ValueError("Frequency must be positive") self.transmitting = True # Reset deadband state for new transmission self.deadband_controller.reset_state() # Get original callbacks original_onoff_callback = indicator_callbacks.get('onoff') if indicator_callbacks else None original_duty_callback = indicator_callbacks.get('duty') if indicator_callbacks else None original_polarity_callback = indicator_callbacks.get('polarity') if indicator_callbacks else None # Create deadband-wrapped callbacks def duty_with_deadband(state): if original_duty_callback: self.deadband_controller.apply_duty_deadband(state, original_duty_callback) def polarity_with_deadband(state): if original_polarity_callback: self.deadband_controller.apply_polarity_deadband(state, original_polarity_callback) # Set TX status pin HIGH if self.gpio_available and self.tx_status_line: try: self.tx_status_line.set_value(1) if original_onoff_callback: original_onoff_callback(True) self.status_callback(f"TX Status pin {self.tx_status_pin} set HIGH") except Exception as e: self.status_callback(f"Failed to set TX status HIGH: {e}") else: # Simulate for GUI when GPIO not available if original_onoff_callback: original_onoff_callback(True) self.status_callback(f"TX Status (simulated): HIGH") # Start frequency output thread (toggles at 2x frequency) self.frequency_output_thread = threading.Thread( target=self._frequency_output_loop, args=(frequency * 2, duty_with_deadband), daemon=True ) self.frequency_output_thread.start() # Start GPIO 26 thread (toggles at selected frequency) self.gpio26_thread = threading.Thread( target=self._gpio26_loop, args=(frequency, polarity_with_deadband), daemon=True ) self.gpio26_thread.start() # Get deadband info for status deadband_info = self.deadband_controller.get_deadband_info() transmission_mode = "REAL GPIO" if self.gpio_available else "SIMULATED" self.status_callback(f"Started {transmission_mode} - Status pin HIGH, Frequency pin toggling at {frequency * 2} Hz, GPIO26 toggling at {frequency} Hz") self.status_callback(f"Deadbands: Duty({deadband_info['duty_rising_ms']:.1f}/{deadband_info['duty_falling_ms']:.1f}ms), Polarity({deadband_info['polarity_rising_ms']:.1f}/{deadband_info['polarity_falling_ms']:.1f}ms)") def stop_transmission(self, indicator_callbacks=None): """Stop GPIO transmission""" # Stop the transmission self.transmitting = False # Reset deadband state self.deadband_controller.reset_state() # Wait for threads to stop if self.frequency_output_thread and self.frequency_output_thread.is_alive(): self.frequency_output_thread.join(timeout=0.5) if self.gpio26_thread and self.gpio26_thread.is_alive(): self.gpio26_thread.join(timeout=0.5) # Set all GPIO pins LOW if self.gpio_available: if self.tx_status_line: try: self.tx_status_line.set_value(0) if indicator_callbacks and 'onoff' in indicator_callbacks: indicator_callbacks['onoff'](False) self.status_callback(f"TX Status pin {self.tx_status_pin} set LOW") except Exception as e: self.status_callback(f"Failed to set TX status LOW: {e}") if self.frequency_line: try: self.frequency_line.set_value(0) self.gpio_state = False if indicator_callbacks and 'duty' in indicator_callbacks: indicator_callbacks['duty'](False) self.status_callback(f"Frequency pin {self.frequency_pin} set LOW") except Exception as e: self.status_callback(f"Failed to set frequency output LOW: {e}") if self.gpio26_line: try: self.gpio26_line.set_value(0) self.gpio26_state = False if indicator_callbacks and 'polarity' in indicator_callbacks: indicator_callbacks['polarity'](False) self.status_callback(f"GPIO26 pin {self.gpio26_pin} set LOW") except Exception as e: self.status_callback(f"Failed to set GPIO26 LOW: {e}") else: # Update indicators even when GPIO not available if indicator_callbacks: for callback in indicator_callbacks.values(): if callback: callback(False) transmission_mode = "REAL GPIO" if self.gpio_available else "SIMULATED" self.status_callback(f"Stopped {transmission_mode} - All GPIO pins set LOW") def _gpio26_loop(self, frequency, indicator_callback): """GPIO 26 output loop running in separate thread""" half_period = 0.5 / frequency while self.transmitting: if self.gpio_available and self.gpio26_line: try: self.gpio26_state = not self.gpio26_state self.gpio26_line.set_value(1 if self.gpio26_state else 0) if indicator_callback: indicator_callback(self.gpio26_state) except Exception as e: self.status_callback(f"GPIO 26 output error: {e}") break else: # Simulate for testing without GPIO self.gpio26_state = not self.gpio26_state if indicator_callback: indicator_callback(self.gpio26_state) time.sleep(half_period) def _frequency_output_loop(self, toggle_frequency, indicator_callback): """Frequency output loop running in separate thread""" half_period = 0.5 / toggle_frequency while self.transmitting: if self.gpio_available and self.frequency_line: try: self.gpio_state = not self.gpio_state self.frequency_line.set_value(1 if self.gpio_state else 0) if indicator_callback: indicator_callback(self.gpio_state) except Exception as e: self.status_callback(f"Frequency output error: {e}") break else: # Simulate for testing without GPIO self.gpio_state = not self.gpio_state if indicator_callback: indicator_callback(self.gpio_state) time.sleep(half_period) def configure_deadbands(self, duty_rising_ms=None, duty_falling_ms=None, polarity_rising_ms=None, polarity_falling_ms=None): """ Configure deadband timing for GPIO signals Args: duty_rising_ms: Duty signal rising edge deadband (milliseconds) duty_falling_ms: Duty signal falling edge deadband (milliseconds) polarity_rising_ms: Polarity signal rising edge deadband (milliseconds) polarity_falling_ms: Polarity signal falling edge deadband (milliseconds) """ if duty_rising_ms is not None or duty_falling_ms is not None: current = self.deadband_controller.get_deadband_info() rising = duty_rising_ms if duty_rising_ms is not None else current['duty_rising_ms'] falling = duty_falling_ms if duty_falling_ms is not None else current['duty_falling_ms'] self.deadband_controller.set_duty_deadbands(rising, falling) if polarity_rising_ms is not None or polarity_falling_ms is not None: current = self.deadband_controller.get_deadband_info() rising = polarity_rising_ms if polarity_rising_ms is not None else current['polarity_rising_ms'] falling = polarity_falling_ms if polarity_falling_ms is not None else current['polarity_falling_ms'] self.deadband_controller.set_polarity_deadbands(rising, falling) deadband_info = self.deadband_controller.get_deadband_info() self.status_callback(f"Deadbands updated: Duty({deadband_info['duty_rising_ms']:.1f}/{deadband_info['duty_falling_ms']:.1f}ms), Polarity({deadband_info['polarity_rising_ms']:.1f}/{deadband_info['polarity_falling_ms']:.1f}ms)") def get_deadband_status(self): """Get current deadband configuration""" return self.deadband_controller.get_deadband_info() def set_custom_deadbands(self, profile='standard'): """Set deadbands using predefined profiles""" if profile == 'csem_optimized': self.deadband_controller.configure_deadbands( duty_rising=0.008, # 8ms duty_falling=0.003, # 3ms polarity_rising=0.012, # 12ms polarity_falling=0.005 # 5ms ) elif profile == 'fast_response': self.deadband_controller.configure_deadbands( duty_rising=0.001, # 1ms duty_falling=0.001, # 1ms polarity_rising=0.002, # 2ms polarity_falling=0.002 # 2ms ) elif profile == 'standard': self.deadband_controller.configure_deadbands( duty_rising=0.005, # 5ms duty_falling=0.005, # 5ms polarity_rising=0.005, # 5ms polarity_falling=0.005 # 5ms ) elif profile == 'no_deadband': self.deadband_controller.configure_deadbands( duty_rising=0.0, # No deadband duty_falling=0.0, # No deadband polarity_rising=0.0, # No deadband polarity_falling=0.0 # No deadband ) deadband_info = self.deadband_controller.get_deadband_info() self.status_callback(f"Profile '{profile}' loaded: Duty({deadband_info['duty_rising_ms']:.1f}/{deadband_info['duty_falling_ms']:.1f}ms), Polarity({deadband_info['polarity_rising_ms']:.1f}/{deadband_info['polarity_falling_ms']:.1f}ms)") def adjust_deadbands_for_frequency(self, frequency): """Automatically adjust deadbands based on transmission frequency""" period = 1.0 / frequency # Use smaller deadbands for higher frequencies if frequency > 10: # High frequency base_deadband = 0.001 # 1ms elif frequency > 1: # Medium frequency base_deadband = 0.005 # 5ms else: # Low frequency base_deadband = 0.010 # 10ms # Ensure deadband is not too large compared to signal period max_deadband = period * 0.1 # Max 10% of period actual_deadband = min(base_deadband, max_deadband) self.deadband_controller.set_duty_deadbands(actual_deadband * 1000, actual_deadband * 1000) self.deadband_controller.set_polarity_deadbands(actual_deadband * 1000, actual_deadband * 1000) self.status_callback(f"Auto-adjusted deadbands for {frequency} Hz: {actual_deadband * 1000:.1f}ms") def cleanup(self): """Clean up GPIO resources""" self.transmitting = False # Stop PPS monitoring self.stop_pps_monitoring() # Reset deadband state self.deadband_controller.reset_state() if self.gpio_available: if self.tx_status_line: try: self.tx_status_line.set_value(0) self.tx_status_line.release() except Exception as e: self.status_callback(f"TX status cleanup error: {e}") if self.frequency_line: try: self.frequency_line.set_value(0) self.frequency_line.release() except Exception as e: self.status_callback(f"Frequency output cleanup error: {e}") if self.gpio26_line: try: self.gpio26_line.set_value(0) self.gpio26_line.release() except Exception as e: self.status_callback(f"GPIO26 cleanup error: {e}") if self.pps_line: try: self.pps_line.release() except Exception as e: self.status_callback(f"PPS cleanup error: {e}") if self.gpio_chip: try: self.gpio_chip.close() except Exception as e: self.status_callback(f"Chip close error: {e}") # Example usage and testing if __name__ == "__main__": # Test the GPIO controller def test_callback(msg): print(f"[TEST] {msg}") def pps_test_callback(state): print(f"[PPS] {'HIGH' if state else 'LOW'}") gpio = GPIOController(status_callback=test_callback) # Test PPS monitoring gpio.start_pps_monitoring(pps_test_callback) # Test deadband configuration print("\nTesting deadband configuration:") print("Current deadbands:", gpio.get_deadband_status()) # Configure custom deadbands gpio.configure_deadbands(duty_rising_ms=8, duty_falling_ms=3, polarity_rising_ms=12, polarity_falling_ms=5) print("After custom config:", gpio.get_deadband_status()) # Test profiles gpio.set_custom_deadbands('fast_response') print("After fast profile:", gpio.get_deadband_status()) # Test frequency adaptive gpio.adjust_deadbands_for_frequency(0.5) # Low frequency print("After frequency adaptive (0.5 Hz):", gpio.get_deadband_status()) # Let PPS monitor run for a few seconds try: time.sleep(5) except KeyboardInterrupt: pass gpio.cleanup()