You can not select more than 25 topics Topics must start with a letter or number, can include dashes ('-') and can be up to 35 characters long.

643 lines
24 KiB

This file contains ambiguous Unicode characters!

This file contains ambiguous Unicode characters that may be confused with others in your current locale. If your use case is intentional and legitimate, you can safely ignore this warning. Use the Escape button to highlight these characters.

#!/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: <span style='color: gray;'>Idle</span>")
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_commands
self.command_label.setText(f"Command: {command_no} / {total}")
self._log_info(f"[{command_no}/{total}] {command_name}")
@pyqtSlot(int, int)
def _on_run_completed(self, run_no: int, packet_count: int):
"""Handle run completed signal."""
self._log_info(f"Run {run_no} complete: {packet_count} packets detected")
@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", or "ERROR"
hex_string: Hex bytes (e.g., "EF FE 01 02 03") or error 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"<span style='color: {color};'>[{prefix}] {hex_string}</span>"
)
# 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: <span style='color: {color};'>{status}</span>")
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"<span style='color: red;'>[ERROR] {message}</span>")
# 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())