#!/usr/bin/env python3 """ Session Widget - vzug-e-hinge ============================== PyQt6 GUI widget for session orchestration. This widget provides the user interface for automated test sessions: - Load interface and session profiles - Start/pause/resume/stop session execution - Display real-time status updates - Show command progress and packet counts - Display countdown timers during delays - Monitor execution via data log The widget connects to Session class via Qt signals/slots for asynchronous updates without blocking the GUI. Author: Kynsight Version: 2.0.0 Date: 2025-11-09 """ # ============================================================================= # IMPORTS # ============================================================================= from PyQt6.QtWidgets import ( QWidget, QVBoxLayout, QHBoxLayout, QGroupBox, QLabel, QLineEdit, QComboBox, QPushButton, QTextEdit, QFrame ) from PyQt6.QtCore import Qt, pyqtSlot from PyQt6.QtGui import QFont from session import Session from session_worker import SessionWorker from database.init_database import DatabaseManager # ============================================================================= # SESSION WIDGET # ============================================================================= class SessionWidget(QWidget): """ Session widget - GUI for automated test sessions. Layout: ┌─ Session Widget ────────────────────────────────────────┐ │ Session Name: [________________] [Load Session ▼] │ │ Interface Profile: [Auto-loaded ▼] │ │ Session Profile: [Auto-loaded ▼] │ │ │ │ Status: Idle │ │ Executing: --- │ │ Command: 0 / 0 │ │ │ │ [▶️ Start] [⏸️ Pause] [⏹️ Stop] │ │ │ │ ┌─ Data Monitor ──────────────────────────────────┐ │ │ │ [INFO] Session loaded: Test_Session_01 │ │ │ │ [INFO] Command 1/4: Open Door │ │ │ │ [INFO] Run 1 complete: 127 packets detected │ │ │ │ [INFO] Waiting... (3s remaining) │ │ │ └─────────────────────────────────────────────────┘ │ └──────────────────────────────────────────────────────────┘ Status Values: - Idle: No session loaded (gray) - Running: Executing run (green) - Waiting: Delay between runs (yellow) - Paused: User paused (orange) - Finished: Session completed (green) - Error: Communication error (red) """ # ========================================================================= # CONSTRUCTOR # ========================================================================= def __init__(self, db_manager: DatabaseManager): """ Initialize session widget. Args: db_manager: Database manager instance """ super().__init__() # Store database manager self.db_manager = db_manager self.db_conn = db_manager.get_connection() # Create session object self.session = Session(db_manager) # Worker thread for non-blocking execution self.worker = None # Connect signals from session to widget slots self._connect_signals() # Initialize UI self._init_ui() # Load initial data self._load_profiles() # ========================================================================= # UI INITIALIZATION # ========================================================================= def _init_ui(self): """Initialize user interface.""" # Main layout layout = QVBoxLayout() layout.setContentsMargins(10, 10, 10, 10) layout.setSpacing(10) # ===================================================================== # Configuration Section # ===================================================================== config_group = QGroupBox("Session Configuration") config_layout = QVBoxLayout() # Session name input name_layout = QHBoxLayout() name_layout.addWidget(QLabel("Session Name:")) self.session_name_input = QLineEdit() self.session_name_input.setPlaceholderText("Auto-generated if empty") name_layout.addWidget(self.session_name_input) config_layout.addLayout(name_layout) # Interface profile dropdown interface_layout = QHBoxLayout() interface_layout.addWidget(QLabel("Interface Profile:")) self.interface_profile_combo = QComboBox() interface_layout.addWidget(self.interface_profile_combo) config_layout.addLayout(interface_layout) # Init session profile dropdown init_layout = QHBoxLayout() init_layout.addWidget(QLabel("Init Session:")) self.init_session_combo = QComboBox() init_layout.addWidget(self.init_session_combo) config_layout.addLayout(init_layout) # Execute session profile dropdown execute_layout = QHBoxLayout() execute_layout.addWidget(QLabel("Execute Session:")) self.execute_session_combo = QComboBox() execute_layout.addWidget(self.execute_session_combo) config_layout.addLayout(execute_layout) # De-init session profile dropdown deinit_layout = QHBoxLayout() deinit_layout.addWidget(QLabel("De-init Session:")) self.deinit_session_combo = QComboBox() deinit_layout.addWidget(self.deinit_session_combo) config_layout.addLayout(deinit_layout) # Load button load_btn_layout = QHBoxLayout() load_btn_layout.addStretch() self.load_button = QPushButton("Load Session") self.load_button.clicked.connect(self._on_load_clicked) load_btn_layout.addWidget(self.load_button) config_layout.addLayout(load_btn_layout) config_group.setLayout(config_layout) layout.addWidget(config_group) # ===================================================================== # Status Section # ===================================================================== status_group = QGroupBox("Session Status") status_layout = QVBoxLayout() # Status label self.status_label = QLabel("Status: Idle") self.status_label.setTextFormat(Qt.TextFormat.RichText) status_layout.addWidget(self.status_label) # Executing label self.executing_label = QLabel("Executing: ---") status_layout.addWidget(self.executing_label) # Command progress label self.command_label = QLabel("Command: 0 / 0") status_layout.addWidget(self.command_label) status_group.setLayout(status_layout) layout.addWidget(status_group) # ===================================================================== # Control Buttons # ===================================================================== btn_layout = QHBoxLayout() # Start button self.start_button = QPushButton("▶️ Start") self.start_button.clicked.connect(self._on_start_clicked) self.start_button.setEnabled(False) btn_layout.addWidget(self.start_button) # Pause button self.pause_button = QPushButton("⏸️ Pause") self.pause_button.clicked.connect(self._on_pause_clicked) self.pause_button.setEnabled(False) btn_layout.addWidget(self.pause_button) # Resume button self.resume_button = QPushButton("▶️ Resume") self.resume_button.clicked.connect(self._on_resume_clicked) self.resume_button.setEnabled(False) self.resume_button.setVisible(False) btn_layout.addWidget(self.resume_button) # Stop button self.stop_button = QPushButton("⏹️ Stop") self.stop_button.clicked.connect(self._on_stop_clicked) self.stop_button.setEnabled(False) btn_layout.addWidget(self.stop_button) layout.addLayout(btn_layout) # ===================================================================== # Data Monitor # ===================================================================== monitor_group = QGroupBox("Data Monitor") monitor_layout = QVBoxLayout() # Text display for log messages self.log_display = QTextEdit() self.log_display.setReadOnly(True) self.log_display.setMinimumHeight(150) # Minimum height, can expand # Set monospace font for log font = QFont("Courier New") font.setPointSize(9) self.log_display.setFont(font) monitor_layout.addWidget(self.log_display) monitor_group.setLayout(monitor_layout) layout.addWidget(monitor_group, 1) # Stretch factor 1 = expands to fill space # ===================================================================== # Finalize Layout # ===================================================================== self.setLayout(layout) self.setMinimumWidth(600) # ========================================================================= # SIGNAL CONNECTION # ========================================================================= def _connect_signals(self): """Connect session signals to widget slots.""" self.session.session_started.connect(self._on_session_started) self.session.command_started.connect(self._on_command_started) self.session.run_completed.connect(self._on_run_completed) self.session.delay_countdown.connect(self._on_delay_countdown) self.session.session_paused.connect(self._on_session_paused) self.session.session_finished.connect(self._on_session_finished) self.session.error_occurred.connect(self._on_error_occurred) self.session.status_changed.connect(self._on_status_changed) self.session.raw_data_received.connect(self._on_raw_data_received) # ========================================================================= # PROFILE LOADING # ========================================================================= def _load_profiles(self): """Load interface and session profiles from database into dropdowns.""" try: # Load interface profiles cursor = self.db_conn.execute(""" SELECT profile_id, profile_name FROM interface_profiles ORDER BY profile_name """) self.interface_profile_combo.clear() for row in cursor.fetchall(): self.interface_profile_combo.addItem(row[1], row[0]) # text, data # Load session profiles (for all 3 phase dropdowns) cursor = self.db_conn.execute(""" SELECT profile_id, profile_name FROM session_profiles ORDER BY profile_name """) session_profiles = cursor.fetchall() # Populate Init session dropdown self.init_session_combo.clear() self.init_session_combo.addItem("(None - Skip Init)", None) # Optional for row in session_profiles: self.init_session_combo.addItem(row[1], row[0]) # Populate Execute session dropdown self.execute_session_combo.clear() self.execute_session_combo.addItem("(None - Skip Execute)", None) # Optional for row in session_profiles: self.execute_session_combo.addItem(row[1], row[0]) # Populate De-init session dropdown self.deinit_session_combo.clear() self.deinit_session_combo.addItem("(None - Skip De-init)", None) # Optional for row in session_profiles: self.deinit_session_combo.addItem(row[1], row[0]) self._log_info("Profiles loaded from database") except Exception as e: self._log_error(f"Failed to load profiles: {str(e)}") def refresh_profiles(self): """ Public method to refresh profile dropdowns. Called when Session tab becomes active. """ self._load_profiles() # ========================================================================= # BUTTON HANDLERS # ========================================================================= def _on_load_clicked(self): """Handle load session button click.""" # Get selected profile IDs interface_id = self.interface_profile_combo.currentData() init_id = self.init_session_combo.currentData() execute_id = self.execute_session_combo.currentData() deinit_id = self.deinit_session_combo.currentData() # Check interface profile selected if interface_id is None: self._log_error("Please select an interface profile") return # Check at least one session phase selected if init_id is None and execute_id is None and deinit_id is None: self._log_error("Please select at least one session phase (Init/Execute/De-init)") return # Get session name session_name = self.session_name_input.text().strip() if not session_name: session_name = None # Will be auto-generated # Load session with 3 phases success = self.session.load_session( interface_profile_id=interface_id, init_session_id=init_id, execute_session_id=execute_id, deinit_session_id=deinit_id, session_name=session_name ) if success: # Enable start button self.start_button.setEnabled(True) self._log_info("Multi-phase session ready to start") else: self._log_error("Failed to load session") def _on_start_clicked(self): """Handle start button click.""" # Disable controls during startup self.start_button.setEnabled(False) self.load_button.setEnabled(False) self.interface_profile_combo.setEnabled(False) self.init_session_combo.setEnabled(False) self.execute_session_combo.setEnabled(False) self.deinit_session_combo.setEnabled(False) self.session_name_input.setEnabled(False) # Create worker thread for non-blocking execution # Pass database path for thread-local connection self.worker = SessionWorker(self.session, self.db_manager.db_path) # Connect worker signals self.worker.finished.connect(self._on_worker_finished) # Start worker thread self.worker.start() # Enable pause/stop buttons immediately # (worker is running, session will start soon) self.pause_button.setEnabled(True) self.stop_button.setEnabled(True) def _on_pause_clicked(self): """Handle pause button click.""" self.session.pause_session() # Button states will be updated by signal handlers def _on_resume_clicked(self): """Handle resume button click.""" # Disable resume button self.resume_button.setEnabled(False) # Create worker thread for resume # Pass database path for thread-local connection self.worker = SessionWorker(self.session, self.db_manager.db_path, resume_mode=True) # Connect worker signals self.worker.finished.connect(self._on_worker_finished) # Start worker thread self.worker.start() # Enable pause/stop buttons self.pause_button.setEnabled(True) self.stop_button.setEnabled(True) def _on_stop_clicked(self): """Handle stop button click.""" self.session.stop_session() # Button states will be updated by signal handlers @pyqtSlot(bool) def _on_worker_finished(self, success: bool): """ Handle worker thread finished. Called when session execution completes (success or failure). Re-enables UI controls for next session. """ # Cleanup worker if self.worker: self.worker = None # Re-enable controls self.start_button.setEnabled(True) self.load_button.setEnabled(True) self.interface_profile_combo.setEnabled(True) self.init_session_combo.setEnabled(True) self.execute_session_combo.setEnabled(True) self.deinit_session_combo.setEnabled(True) self.session_name_input.setEnabled(True) # Disable pause/stop buttons self.pause_button.setEnabled(False) self.stop_button.setEnabled(False) if not success: self._log_error("Session execution failed") # ========================================================================= # SIGNAL HANDLERS (SLOTS) # ========================================================================= @pyqtSlot(str) def _on_session_started(self, session_id: str): """Handle session started signal.""" self._update_status("Running", "green") self._log_info(f"Session started: {session_id}") @pyqtSlot(int, str) def _on_command_started(self, command_no: int, command_name: str): """Handle command started signal.""" self.executing_label.setText(f"Executing: {command_name}") total = self.session.total_uart_runs self.command_label.setText(f"Run: {command_no} / {total}") # Don't log here - already logged via raw_data_received with formatted tags @pyqtSlot(int, int) def _on_run_completed(self, run_no: int, packet_count: int): """Handle run completed signal.""" # Don't log here - already logged via raw_data_received with formatted tags pass @pyqtSlot(int) def _on_delay_countdown(self, seconds_remaining: int): """Handle delay countdown signal.""" self._update_status("Waiting", "orange") self.executing_label.setText(f"Waiting... ({seconds_remaining}s remaining)") @pyqtSlot() def _on_session_paused(self): """Handle session paused signal.""" self._update_status("Paused", "orange") self.executing_label.setText("Paused by user") # Update button states self.pause_button.setVisible(False) self.pause_button.setEnabled(False) self.resume_button.setVisible(True) self.resume_button.setEnabled(True) self.stop_button.setEnabled(False) self._log_info("Session paused") @pyqtSlot() def _on_session_finished(self): """Handle session finished signal.""" self._update_status("Finished", "green") self.executing_label.setText("Session completed successfully") # Reset button states self._reset_controls() self._log_info("Session completed successfully ✓") @pyqtSlot(str) def _on_error_occurred(self, error_msg: str): """Handle error signal.""" self._update_status("Error", "red") self._log_error(error_msg) # Reset button states self._reset_controls() @pyqtSlot(str) def _on_status_changed(self, status_text: str): """Handle status changed signal.""" # Log status updates (but filter out countdown spam) if "remaining)" not in status_text: self._log_info(status_text) @pyqtSlot(str, str) def _on_raw_data_received(self, direction: str, hex_string: str): """ Handle raw UART data display. Args: direction: "TX", "RX", "ERROR", "INFO" or formatted "[Phase] [Interface] [Action]" hex_string: Hex bytes (e.g., "EF FE 01 02 03") or error message """ # Check if direction is already formatted (starts with '[') if direction.startswith('['): # New format: [Phase] [Interface] [Action] message # Extract action for color coding if '[TX]' in direction: color = "green" elif '[ERROR]' in direction: color = "red" elif '[INFO]' in direction: color = "gray" else: # RX or other color = "blue" # Display full formatted string self.log_display.append( f"{direction} {hex_string}" ) else: # Old format: Simple direction + message if direction == "TX": color = "green" prefix = "→ TX" elif direction == "ERROR": color = "red" prefix = "✗ ERROR" elif direction == "INFO": color = "gray" prefix = "ℹ INFO" else: color = "blue" prefix = "← RX" self.log_display.append( f"[{prefix}] {hex_string}" ) # Auto-scroll to bottom self.log_display.verticalScrollBar().setValue( self.log_display.verticalScrollBar().maximum() ) # ========================================================================= # UI HELPER METHODS # ========================================================================= def _update_status(self, status: str, color: str): """ Update status label with color. Args: status: Status text color: HTML color name """ self.status_label.setText(f"Status: {status}") def _reset_controls(self): """Reset controls to initial state after session ends.""" # Enable configuration controls self.load_button.setEnabled(True) self.interface_profile_combo.setEnabled(True) self.init_session_combo.setEnabled(True) self.execute_session_combo.setEnabled(True) self.deinit_session_combo.setEnabled(True) self.session_name_input.setEnabled(True) # Reset button states self.start_button.setEnabled(True) self.pause_button.setVisible(True) self.pause_button.setEnabled(False) self.resume_button.setVisible(False) self.resume_button.setEnabled(False) self.stop_button.setEnabled(False) # Reset labels self.executing_label.setText("Executing: ---") self.command_label.setText("Command: 0 / 0") def _log_info(self, message: str): """ Log info message to data monitor. Args: message: Message to log """ self.log_display.append(f"[INFO] {message}") # Auto-scroll to bottom self.log_display.verticalScrollBar().setValue( self.log_display.verticalScrollBar().maximum() ) def _log_error(self, message: str): """ Log error message to data monitor. Args: message: Error message to log """ self.log_display.append(f"[ERROR] {message}") # Auto-scroll to bottom self.log_display.verticalScrollBar().setValue( self.log_display.verticalScrollBar().maximum() ) # ============================================================================= # MAIN (for testing) # ============================================================================= if __name__ == "__main__": import sys from PyQt6.QtWidgets import QApplication # Create application app = QApplication(sys.argv) # Create database manager db_manager = DatabaseManager("database/ehinge.db") db_manager.initialize() # Create widget widget = SessionWidget(db_manager) widget.setWindowTitle("Session Widget - vzug-e-hinge") widget.show() # Run application sys.exit(app.exec())