#!/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
# Reference to main window (for tab control)
self.main_window = 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)
# Session notes input
notes_layout = QVBoxLayout()
notes_layout.addWidget(QLabel("Session Notes:"))
self.session_notes_input = QTextEdit()
self.session_notes_input.setMaximumHeight(60)
self.session_notes_input.setPlaceholderText("Optional: Add description or notes for this session...")
notes_layout.addWidget(self.session_notes_input)
config_layout.addLayout(notes_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 and Unload buttons
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)
self.unload_button = QPushButton("Unload Session")
self.unload_button.clicked.connect(self._on_unload_clicked)
self.unload_button.setEnabled(False) # Initially disabled
load_btn_layout.addWidget(self.unload_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()
def set_main_window(self, main_window):
"""
Set reference to main window for tab control.
Args:
main_window: MainWindow instance
"""
self.main_window = main_window
# =========================================================================
# 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
# Check if session name already exists in database
if session_name:
cursor = self.db_conn.execute("""
SELECT COUNT(*) FROM sessions WHERE session_name = ?
""", (session_name,))
count = cursor.fetchone()[0]
if count > 0:
# Session name exists - prompt user
from PyQt6.QtWidgets import QMessageBox
reply = QMessageBox.question(
self,
"Session Name Exists",
f"Session name '{session_name}' already exists in database.\n\n"
f"Override will delete the existing session and all its telemetry data.\n\n"
f"What would you like to do?",
QMessageBox.StandardButton.Cancel | QMessageBox.StandardButton.Ok,
QMessageBox.StandardButton.Cancel
)
# Change OK button text to "Override"
if reply == QMessageBox.StandardButton.Cancel:
self._log_info("Load cancelled - session name already exists")
return
elif reply == QMessageBox.StandardButton.Ok:
# User chose to override - delete existing session data
self._log_info(f"Overriding existing session '{session_name}'...")
try:
# Delete from sessions table
self.db_conn.execute("DELETE FROM sessions WHERE session_name = ?", (session_name,))
# Delete from telemetry_raw table
self.db_conn.execute("DELETE FROM telemetry_raw WHERE session_name = ?", (session_name,))
# Delete from telemetry_decoded table
self.db_conn.execute("DELETE FROM telemetry_decoded WHERE session_name = ?", (session_name,))
self.db_conn.commit()
self._log_info(f"Existing session '{session_name}' deleted")
except Exception as e:
self._log_error(f"Failed to delete existing session: {e}")
return
# 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 and unload button
self.start_button.setEnabled(True)
self.unload_button.setEnabled(True)
self._log_info("Multi-phase session ready to start")
# Disable UART and I2C tabs to prevent port conflicts
self._disable_port_tabs()
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.unload_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)
self.session_notes_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
def _on_unload_clicked(self):
"""Handle unload session button click."""
# Unload the session and re-enable UART/I2C tabs
self._cleanup_session()
self._log_info("Session manually unloaded by user")
@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 configuration controls
# NOTE: Start button will be disabled by _cleanup_session() since session is unloaded
# self.start_button.setEnabled(True) # <-- REMOVED: Don't enable, session is unloaded
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)
self.session_notes_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}")
# Save session notes to database
notes = self.session_notes_input.toPlainText().strip()
if notes:
try:
self.db_conn.execute("""
UPDATE sessions
SET notes = ?
WHERE session_id = ?
""", (notes, session_id))
self.db_conn.commit()
except Exception as e:
print(f"[WARN] Failed to save session notes: {e}")
@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()
# Cleanup: unload session and re-enable port tabs
self._cleanup_session()
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()
# Cleanup: unload session and re-enable port tabs
self._cleanup_session()
@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)
self.session_notes_input.setEnabled(True)
# Reset button states
# NOTE: Start button will be disabled by _cleanup_session() since session is unloaded
# self.start_button.setEnabled(True) # <-- REMOVED: Don't enable, session is unloaded
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()
)
def _disable_port_tabs(self):
"""
Disable UART and I2C tabs to prevent port conflicts.
Called when session is loaded.
"""
if not self.main_window:
return
try:
# UART tab is at index 3, I2C tab is at index 4
self.main_window.tabs.setTabEnabled(3, False) # UART
self.main_window.tabs.setTabEnabled(4, False) # I2C
self._log_info("UART and I2C tabs disabled (port conflict prevention)")
except Exception as e:
print(f"[WARN] Failed to disable port tabs: {e}")
def _enable_port_tabs(self):
"""
Re-enable UART and I2C tabs.
Called when session ends.
"""
if not self.main_window:
return
try:
# UART tab is at index 3, I2C tab is at index 4
self.main_window.tabs.setTabEnabled(3, True) # UART
self.main_window.tabs.setTabEnabled(4, True) # I2C
self._log_info("UART and I2C tabs re-enabled")
except Exception as e:
print(f"[WARN] Failed to enable port tabs: {e}")
def _cleanup_session(self):
"""
Cleanup after session ends.
- Unload session (closes ports)
- Re-enable UART/I2C tabs
- Disable Start/Pause/Stop/Unload buttons (no session loaded)
- Keep UI selections intact
"""
try:
# Unload session (closes all ports)
if self.session:
self.session.unload_session()
self._log_info("Session unloaded, ports closed")
# Re-enable port tabs
self._enable_port_tabs()
# Disable Start/Pause/Stop/Unload buttons (session is unloaded)
self.start_button.setEnabled(False)
self.pause_button.setEnabled(False)
self.stop_button.setEnabled(False)
self.unload_button.setEnabled(False)
# NOTE: We deliberately do NOT clear the UI selections
# (session_name_input, interface_profile_combo, etc.)
# so the user can easily re-run the same configuration
except Exception as e:
self._log_error(f"Cleanup error: {e}")
# =============================================================================
# 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())