|
|
#!/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: <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()
|
|
|
|
|
|
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"<span style='color: {color};'>{direction} {hex_string}</span>"
|
|
|
)
|
|
|
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"<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)
|
|
|
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"<span style='color: red;'>[ERROR] {message}</span>")
|
|
|
# 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())
|
