/
devsec
/
anothertap
Обзор
Документация
Войти
/
devsec
/
anothertap
Код
Запросы
0
Задачи
Вики
Пакеты
0
Релизы
0
CI/CD
Аналитика
Безопасность
master
src/gui/device_manager_widget.py
1 452 строки
58 KB
devsec
clean code gitignore
24 авг 2025, 21:27
24 авг 2025, 21:27
75a9527
Код
Авторство
О чём код?
""" Device Manager Widget for CRUD operations on measurement devices. This module provides a comprehensive GUI for managing measurement devices, including device discovery, connection management, and SCPI command testing. """ from typing import Optional, List, Dict, Any import logging from PySide6.QtWidgets import ( QWidget, QVBoxLayout, QHBoxLayout, QSplitter, QTreeWidget, QTreeWidgetItem, QGroupBox, QFormLayout, QLineEdit, QComboBox, QTextEdit, QPushButton, QMessageBox, QDialog, QDialogButtonBox, QLabel, QTableWidget, QTableWidgetItem, QHeaderView, QMenu, QProgressDialog, QCheckBox ) from PySide6.QtCore import Qt, Signal, QThread from PySide6.QtGui import QAction, QFont from models.device import Device, ConnectionType, DeviceStatus from services.device_manager import DeviceManager class DeviceManagerWidget(QWidget): """ Widget for managing measurement devices with comprehensive CRUD operations. This widget provides a user interface for: - Device list management with connection status - Device configuration and editing - SCPI command testing - Device discovery and auto-identification Signals: device_connected: Emitted when a device is successfully connected device_disconnected: Emitted when a device is disconnected """ # Constants for UI layout SPLITTER_SIZES = [400, 600] SCPI_RESPONSE_HEIGHT = 100 DESCRIPTION_HEIGHT = 80 # Signals device_connected = Signal(str) # device_id device_disconnected = Signal(str) # device_id def __init__(self, device_manager: DeviceManager) -> None: super().__init__() self.device_manager = device_manager self.current_device: Optional[Device] = None self.logger = logging.getLogger(self.__class__.__name__) self._setup_ui() self._setup_scpi_command_history() self.refresh_devices() def _setup_ui(self) -> None: """Set up the main user interface layout.""" layout = QHBoxLayout(self) # Create main splitter for resizable panels splitter = QSplitter(Qt.Orientation.Horizontal) layout.addWidget(splitter) # Setup panels self._setup_device_list_panel(splitter) self._setup_device_details_panel(splitter) # Configure splitter proportions splitter.setSizes(self.SPLITTER_SIZES) def _setup_device_list_panel(self, parent: QSplitter) -> None: """Set up the device list panel with tree widget and controls.""" panel_widget = QWidget() panel_layout = QVBoxLayout(panel_widget) parent.addWidget(panel_widget) # Header with controls self._create_list_header(panel_layout) # Device tree widget self._create_device_tree(panel_layout) def _create_list_header(self, layout: QVBoxLayout) -> None: """Create the header section with title and action buttons.""" header_layout = QHBoxLayout() layout.addLayout(header_layout) header_layout.addWidget(QLabel("Devices")) header_layout.addStretch() # Action buttons self.add_device_btn = QPushButton("Add Device") self.add_device_btn.clicked.connect(self._on_add_device_clicked) header_layout.addWidget(self.add_device_btn) self.discover_btn = QPushButton("Discover") self.discover_btn.clicked.connect(self._on_discover_clicked) header_layout.addWidget(self.discover_btn) self.simulate_btn = QPushButton("Add Simulation") self.simulate_btn.clicked.connect(self._on_simulate_clicked) header_layout.addWidget(self.simulate_btn) def _create_device_tree(self, layout: QVBoxLayout) -> None: """Create and configure the device tree widget.""" self.device_tree = QTreeWidget() self.device_tree.setHeaderLabels(["Device", "Status", "Connection"]) self.device_tree.itemClicked.connect(self._on_device_selected) self.device_tree.setContextMenuPolicy(Qt.ContextMenuPolicy.CustomContextMenu) self.device_tree.customContextMenuRequested.connect(self._show_device_context_menu) layout.addWidget(self.device_tree) # Configure column resizing header = self.device_tree.header() header.setSectionResizeMode(0, QHeaderView.ResizeMode.Stretch) header.setSectionResizeMode(1, QHeaderView.ResizeMode.ResizeToContents) header.setSectionResizeMode(2, QHeaderView.ResizeMode.ResizeToContents) def _setup_device_details_panel(self, parent: QSplitter) -> None: """Set up the device details panel with configuration forms.""" details_widget = QWidget() details_layout = QVBoxLayout(details_widget) parent.addWidget(details_widget) # Create detail sections self._create_device_info_section(details_layout) self._create_connection_section(details_layout) self._create_actions_section(details_layout) self._create_scpi_test_section(details_layout) details_layout.addStretch() def _create_device_info_section(self, layout: QVBoxLayout) -> None: """Create device information form section.""" info_group = QGroupBox("Device Information") info_layout = QFormLayout(info_group) layout.addWidget(info_group) # Create form fields self.name_edit = QLineEdit() self.name_edit.textChanged.connect(self._on_device_info_changed) info_layout.addRow("Name:", self.name_edit) self.manufacturer_edit = QLineEdit() self.manufacturer_edit.textChanged.connect(self._on_device_info_changed) info_layout.addRow("Manufacturer:", self.manufacturer_edit) self.model_edit = QLineEdit() self.model_edit.textChanged.connect(self._on_device_info_changed) info_layout.addRow("Model:", self.model_edit) self.serial_edit = QLineEdit() self.serial_edit.textChanged.connect(self._on_device_info_changed) info_layout.addRow("Serial Number:", self.serial_edit) self.description_edit = QTextEdit() self.description_edit.setMaximumHeight(self.DESCRIPTION_HEIGHT) self.description_edit.textChanged.connect(self._on_device_info_changed) info_layout.addRow("Description:", self.description_edit) def _create_connection_section(self, layout: QVBoxLayout) -> None: """Create connection configuration section.""" conn_group = QGroupBox("Connection") conn_layout = QFormLayout(conn_group) layout.addWidget(conn_group) # Connection type dropdown self.connection_type_combo = QComboBox() for conn_type in ConnectionType: self.connection_type_combo.addItem(conn_type.value) self.connection_type_combo.currentTextChanged.connect(self._on_device_info_changed) conn_layout.addRow("Type:", self.connection_type_combo) # Connection string self.connection_string_edit = QLineEdit() self.connection_string_edit.setPlaceholderText("e.g., TCPIP::192.168.1.100::INSTR") self.connection_string_edit.textChanged.connect(self._on_device_info_changed) conn_layout.addRow("Connection String:", self.connection_string_edit) # Status display self.status_label = QLabel("Disconnected") self.status_label.setStyleSheet("color: red; font-weight: bold;") conn_layout.addRow("Status:", self.status_label) self.error_label = QLabel("") self.error_label.setStyleSheet("color: red;") self.error_label.setWordWrap(True) conn_layout.addRow("Last Error:", self.error_label) def _create_actions_section(self, layout: QVBoxLayout) -> None: """Create action buttons section.""" actions_group = QGroupBox("Actions") actions_layout = QHBoxLayout(actions_group) layout.addWidget(actions_group) self.save_btn = QPushButton("Save") self.save_btn.clicked.connect(self._on_save_clicked) self.save_btn.setEnabled(False) actions_layout.addWidget(self.save_btn) self.connect_btn = QPushButton("Connect") self.connect_btn.clicked.connect(self._on_connect_clicked) actions_layout.addWidget(self.connect_btn) self.test_btn = QPushButton("Test Connection") self.test_btn.clicked.connect(self._on_test_connection_clicked) actions_layout.addWidget(self.test_btn) self.delete_btn = QPushButton("Delete") self.delete_btn.clicked.connect(self._on_delete_clicked) self.delete_btn.setStyleSheet("color: red;") actions_layout.addWidget(self.delete_btn) def _create_scpi_test_section(self, layout: QVBoxLayout) -> None: """Create SCPI command testing section.""" scpi_group = QGroupBox("SCPI Test") scpi_layout = QVBoxLayout(scpi_group) layout.addWidget(scpi_group) # Command input cmd_layout = QHBoxLayout() scpi_layout.addLayout(cmd_layout) self.scpi_command_edit = QLineEdit() self.scpi_command_edit.setPlaceholderText("Enter SCPI command (e.g., *IDN?)") cmd_layout.addWidget(self.scpi_command_edit) self.send_scpi_btn = QPushButton("Send") self.send_scpi_btn.clicked.connect(self._on_send_scpi_clicked) cmd_layout.addWidget(self.send_scpi_btn) # Response display self.scpi_response_edit = QTextEdit() self.scpi_response_edit.setMaximumHeight(self.SCPI_RESPONSE_HEIGHT) self.scpi_response_edit.setReadOnly(True) # Set monospace font for better readability font = QFont("Consolas", 9) font.setStyleHint(QFont.StyleHint.Monospace) self.scpi_response_edit.setFont(font) scpi_layout.addWidget(self.scpi_response_edit) def refresh_devices(self) -> None: """Refresh the device list display with current devices.""" self.device_tree.clear() devices = self.device_manager.get_all_devices() for device in devices: self._add_device_to_tree(device) # Auto-select first device if available if self.device_tree.topLevelItemCount() > 0: first_item = self.device_tree.topLevelItem(0) if first_item: self.device_tree.setCurrentItem(first_item) self._on_device_selected(first_item, 0) def _add_device_to_tree(self, device: Device) -> None: """Add a single device to the tree widget.""" item = QTreeWidgetItem() # Add simulation indicator to device name display_name = device.display_name if device.simulation_mode or device.connection_type.value == "SIMULATION": display_name = f"[SIM] {display_name}" item.setText(0, display_name) item.setText(1, device.status.value) item.setText(2, device.connection_string) item.setData(0, Qt.ItemDataRole.UserRole, device.id) # Set status-based background color self._set_status_color(item, device.status) # Set simulation-based styling if device.simulation_mode or device.connection_type.value == "SIMULATION": # Make simulated devices italic font = item.font(0) font.setItalic(True) item.setFont(0, font) item.setFont(1, font) item.setFont(2, font) self.device_tree.addTopLevelItem(item) def _set_status_color(self, item: QTreeWidgetItem, status: DeviceStatus) -> None: """Set item background color based on device status.""" if status == DeviceStatus.CONNECTED: item.setBackground(1, Qt.GlobalColor.green) elif status == DeviceStatus.ERROR: item.setBackground(1, Qt.GlobalColor.red) def _on_device_selected(self, item: QTreeWidgetItem, column: int) -> None: """Handle device selection from the tree widget.""" if not item: return device_id = item.data(0, Qt.ItemDataRole.UserRole) device = self.device_manager.get_device(device_id) if device: self.current_device = device self._load_device_details(device) else: self.logger.warning(f"Device with ID {device_id} not found") def _load_device_details(self, device: Device) -> None: """Load device details into the form fields.""" try: self.name_edit.setText(device.name) self.manufacturer_edit.setText(device.manufacturer) self.model_edit.setText(device.model) self.serial_edit.setText(device.serial_number) self.description_edit.setText(device.description) # Set connection type self._set_connection_type_combo(device.connection_type) self.connection_string_edit.setText(device.connection_string) # Update status display self._update_device_status_display(device) # Reset save button state self.save_btn.setEnabled(False) except Exception as e: self.logger.error(f"Error loading device details: {e}") self._show_error_message("Failed to load device details") def _set_connection_type_combo(self, connection_type: ConnectionType) -> None: """Set the connection type combo box to the specified type.""" for i in range(self.connection_type_combo.count()): if self.connection_type_combo.itemText(i) == connection_type.value: self.connection_type_combo.setCurrentIndex(i) break def _update_device_status_display(self, device: Device) -> None: """Update the device status display elements.""" status_config = { DeviceStatus.CONNECTED: ("Connected", "color: green; font-weight: bold;", "Disconnect"), DeviceStatus.ERROR: ("Error", "color: red; font-weight: bold;", "Connect"), DeviceStatus.DISCONNECTED: ("Disconnected", "color: red; font-weight: bold;", "Connect") } status_text, style, button_text = status_config.get( device.status, ("Unknown", "color: gray; font-weight: bold;", "Connect") ) self.status_label.setText(status_text) self.status_label.setStyleSheet(style) self.connect_btn.setText(button_text) # Update error message self.error_label.setText(device.last_error or "") # Enable/disable SCPI test controls is_connected = device.status == DeviceStatus.CONNECTED self.send_scpi_btn.setEnabled(is_connected) self.scpi_command_edit.setEnabled(is_connected) def _on_device_info_changed(self) -> None: """Handle changes to device information fields.""" if self.current_device: self.save_btn.setEnabled(True) def _on_save_clicked(self) -> None: """Handle save button click.""" if not self.current_device: return try: self._update_device_from_form() if self.device_manager.update_device(self.current_device): self.save_btn.setEnabled(False) self.refresh_devices() self._show_success_message("Device saved successfully") else: self._show_error_message("Failed to save device") except Exception as e: self.logger.error(f"Error saving device: {e}") self._show_error_message("An error occurred while saving the device") def _update_device_from_form(self) -> None: """Update the current device with form data.""" if not self.current_device: return self.current_device.name = self.name_edit.text() self.current_device.manufacturer = self.manufacturer_edit.text() self.current_device.model = self.model_edit.text() self.current_device.serial_number = self.serial_edit.text() self.current_device.description = self.description_edit.toPlainText() self.current_device.connection_type = ConnectionType( self.connection_type_combo.currentText() ) self.current_device.connection_string = self.connection_string_edit.text() def _show_success_message(self, message: str) -> None: """Show a success message to the user.""" QMessageBox.information(self, "Success", message) def _show_error_message(self, message: str) -> None: """Show an error message to the user.""" QMessageBox.warning(self, "Error", message) def _show_confirmation_dialog(self, message: str) -> bool: """Show a confirmation dialog and return True if user confirms.""" reply = QMessageBox.question( self, "Confirm Action", message, QMessageBox.StandardButton.Yes | QMessageBox.StandardButton.No, QMessageBox.StandardButton.No ) return reply == QMessageBox.StandardButton.Yes def _on_add_device_clicked(self) -> None: """Handle add device button click.""" dialog = DeviceDialog(self) if dialog.exec() == QDialog.DialogCode.Accepted: device = dialog.get_device() if self.device_manager.create_device(device): self.refresh_devices() self._select_device_in_tree(device.id) else: self._show_error_message("Failed to create device") def _select_device_in_tree(self, device_id: str) -> None: """Select a device in the tree by its ID.""" for i in range(self.device_tree.topLevelItemCount()): item = self.device_tree.topLevelItem(i) if item and item.data(0, Qt.ItemDataRole.UserRole) == device_id: self.device_tree.setCurrentItem(item) self._on_device_selected(item, 0) break def _on_delete_clicked(self) -> None: """Handle delete device button click.""" if not self.current_device: return message = f"Are you sure you want to delete device '{self.current_device.display_name}'?" if self._show_confirmation_dialog(message): if self.device_manager.delete_device(self.current_device.id): self.refresh_devices() self._clear_device_form() self.current_device = None else: self._show_error_message("Failed to delete device") def _clear_device_form(self) -> None: """Clear all device form fields.""" self.name_edit.clear() self.manufacturer_edit.clear() self.model_edit.clear() self.serial_edit.clear() self.description_edit.clear() self.connection_string_edit.clear() def _on_connect_clicked(self) -> None: """Handle connect/disconnect button click.""" if not self.current_device: return try: if self.current_device.is_connected(): self._disconnect_device() else: self._connect_device() except Exception as e: self.logger.error(f"Connection operation failed: {e}") self._show_error_message("Connection operation failed") def _connect_device(self) -> None: """Connect to the current device.""" if not self.current_device: return if self.device_manager.connect_device(self.current_device.id): self.device_connected.emit(self.current_device.id) self._update_device_status_display(self.current_device) self.refresh_devices() else: self._show_error_message(f"Failed to connect to {self.current_device.display_name}") def _disconnect_device(self) -> None: """Disconnect from the current device.""" if not self.current_device: return if self.device_manager.disconnect_device(self.current_device.id): self.device_disconnected.emit(self.current_device.id) self._update_device_status_display(self.current_device) self.refresh_devices() else: self._show_error_message(f"Failed to disconnect from {self.current_device.display_name}") def _on_test_connection_clicked(self) -> None: """Handle test connection button click.""" if not self.current_device: return try: if self.device_manager.test_device_connection(self.current_device.id): self._show_success_message("Connection test successful") else: self._show_error_message("Connection test failed") except Exception as e: self.logger.error(f"Connection test failed: {e}") self._show_error_message("Connection test failed") def _on_send_scpi_clicked(self) -> None: """Handle SCPI command send button click.""" if not self._validate_scpi_prerequisites(): return command = self.scpi_command_edit.text().strip() if not self._validate_scpi_command(command): return try: self._execute_scpi_command(command) except Exception as e: self.logger.error(f"SCPI command execution failed: {e}") self._append_scpi_response(f"> {command}\nERROR: Command execution failed: {str(e)}\n") def _validate_scpi_command(self, command: str) -> bool: """Validate SCPI command input.""" if not command: self._show_error_message("Please enter a SCPI command") return False # Basic SCPI command validation if len(command) > 1000: # Reasonable command length limit self._show_error_message("SCPI command is too long (max 1000 characters)") return False # Check for potentially dangerous commands (basic safety) dangerous_commands = ['*RST', 'SYST:ERR:CLE', 'ABORT'] command_upper = command.upper() for dangerous_cmd in dangerous_commands: if dangerous_cmd in command_upper: reply = QMessageBox.question( self, "Confirm Command", f"Command '{command}' may reset or change device state. Continue?", QMessageBox.StandardButton.Yes | QMessageBox.StandardButton.No, QMessageBox.StandardButton.No ) return reply == QMessageBox.StandardButton.Yes return True def _validate_scpi_prerequisites(self) -> bool: """Validate prerequisites for SCPI command execution.""" if not self.current_device: self._show_error_message("No device selected") return False if not self.current_device.is_connected(): self._show_error_message("Device is not connected") return False return True def _execute_scpi_command(self, command: str) -> None: """Execute a SCPI command and display the result.""" if not self.current_device: return # Add timestamp to response from datetime import datetime timestamp = datetime.now().strftime("%H:%M:%S") try: success, response = self.device_manager.scpi_service.send_command( self.current_device, command ) if success: if response is not None: self._append_scpi_response(f"[{timestamp}] > {command}\n{response}\n") else: self._append_scpi_response(f"[{timestamp}] > {command}\n(Command sent successfully)\n") else: self._append_scpi_response(f"[{timestamp}] > {command}\nERROR: Failed to send command\n") except Exception as e: self._append_scpi_response(f"[{timestamp}] > {command}\nERROR: {str(e)}\n") raise # Clear command input for next command self.scpi_command_edit.clear() # Add command to history self._add_command_to_history(command) def _append_scpi_response(self, text: str) -> None: """Append text to the SCPI response display.""" self.scpi_response_edit.append(text) # Auto-scroll to bottom scrollbar = self.scpi_response_edit.verticalScrollBar() scrollbar.setValue(scrollbar.maximum()) def _on_discover_clicked(self) -> None: """Handle device discovery button click.""" try: self._start_device_discovery() except Exception as e: self.logger.error(f"Device discovery failed: {e}") self._show_error_message("Device discovery failed") def _start_device_discovery(self) -> None: """Start the device discovery process.""" # Create and start discovery thread self.discovery_thread = DeviceDiscoveryThread(self.device_manager) self.discovery_thread.devices_discovered.connect(self._on_devices_discovered) self.discovery_thread.start() # Show progress dialog self.progress_dialog = QProgressDialog("Discovering devices...", "Cancel", 0, 0, self) self.progress_dialog.setWindowModality(Qt.WindowModality.WindowModal) self.progress_dialog.show() def _on_devices_discovered(self, resources: List[str]) -> None: """Handle the completion of device discovery.""" if hasattr(self, 'progress_dialog'): self.progress_dialog.close() if not resources: self._show_error_message("No devices found") return try: self._show_discovery_dialog(resources) except Exception as e: self.logger.error(f"Error showing discovery dialog: {e}") self._show_error_message("Error processing discovered devices") def _show_discovery_dialog(self, resources: List[str]) -> None: """Show the device discovery dialog.""" dialog = DiscoveryDialog(resources, self.device_manager, self) if dialog.exec() == QDialog.DialogCode.Accepted: self.refresh_devices() def _show_device_context_menu(self, position) -> None: """Show context menu for device tree.""" item = self.device_tree.itemAt(position) if not item or not self.current_device: return menu = self._create_device_context_menu() menu.exec(self.device_tree.mapToGlobal(position)) def _create_device_context_menu(self) -> QMenu: """Create the context menu for device operations.""" menu = QMenu(self) if not self.current_device: return menu # Connection action connection_text = "Disconnect" if self.current_device.is_connected() else "Connect" connect_action = QAction(connection_text, self) connect_action.triggered.connect(self._on_connect_clicked) menu.addAction(connect_action) # Test connection action test_action = QAction("Test Connection", self) test_action.triggered.connect(self._on_test_connection_clicked) menu.addAction(test_action) menu.addSeparator() # Delete action delete_action = QAction("Delete", self) delete_action.triggered.connect(self._on_delete_clicked) menu.addAction(delete_action) return menu def _add_command_to_history(self, command: str) -> None: """Add command to SCPI command history for easy recall.""" # Simple history implementation - could be enhanced with persistent storage if not hasattr(self, '_scpi_history'): self._scpi_history = [] # Remove if already exists to avoid duplicates if command in self._scpi_history: self._scpi_history.remove(command) # Add to beginning of history self._scpi_history.insert(0, command) # Limit history size if len(self._scpi_history) > 20: self._scpi_history = self._scpi_history[:20] def _setup_scpi_command_history(self) -> None: """Setup SCPI command history functionality (to be called after UI setup).""" # Enable up/down arrow key navigation for command history from PySide6.QtCore import QEvent from PySide6.QtGui import QKeySequence def handle_key_press(event): if hasattr(self, '_scpi_history') and self._scpi_history: if event.key() == Qt.Key.Key_Up: self._navigate_history(-1) event.accept() return True elif event.key() == Qt.Key.Key_Down: self._navigate_history(1) event.accept() return True return False # Store original keyPressEvent original_key_press = self.scpi_command_edit.keyPressEvent def enhanced_key_press(event): if not handle_key_press(event): original_key_press(event) # Override keyPressEvent self.scpi_command_edit.keyPressEvent = enhanced_key_press def _navigate_history(self, direction: int) -> None: """Navigate through SCPI command history.""" if not hasattr(self, '_scpi_history') or not self._scpi_history: return if not hasattr(self, '_history_index'): self._history_index = -1 self._history_index += direction # Clamp to valid range self._history_index = max(0, min(self._history_index, len(self._scpi_history) - 1)) # Set command text if 0 <= self._history_index < len(self._scpi_history): self.scpi_command_edit.setText(self._scpi_history[self._history_index]) def _on_simulate_clicked(self) -> None: """Handle add simulation button click.""" try: self._show_simulation_dialog() except Exception as e: self.logger.error(f"Simulation dialog failed: {e}") self._show_error_message("Failed to open simulation dialog") def _show_simulation_dialog(self) -> None: """Show the device simulation dialog.""" dialog = SimulationDialog(self.device_manager, self) if dialog.exec() == QDialog.DialogCode.Accepted: self.refresh_devices() class DeviceDialog(QDialog): """Dialog for creating and editing device configurations. This dialog provides a form for entering device information including name, manufacturer, model, connection details, and description. """ def __init__(self, parent: Optional[QWidget] = None) -> None: super().__init__(parent) self.setWindowTitle("Add Device") self.setModal(True) self.resize(400, 300) self._setup_ui() self._setup_validation() def _setup_ui(self) -> None: """Set up the dialog user interface.""" layout = QVBoxLayout(self) # Create form self._create_form(layout) # Add buttons self._create_buttons(layout) def _create_form(self, layout: QVBoxLayout) -> None: """Create the device configuration form.""" form_layout = QFormLayout() layout.addLayout(form_layout) self.name_edit = QLineEdit() form_layout.addRow("Name:", self.name_edit) self.manufacturer_edit = QLineEdit() form_layout.addRow("Manufacturer:", self.manufacturer_edit) self.model_edit = QLineEdit() form_layout.addRow("Model:", self.model_edit) self.connection_type_combo = QComboBox() for conn_type in ConnectionType: self.connection_type_combo.addItem(conn_type.value) form_layout.addRow("Connection Type:", self.connection_type_combo) self.connection_string_edit = QLineEdit() self.connection_string_edit.setPlaceholderText("e.g., TCPIP::192.168.1.100::INSTR") form_layout.addRow("Connection String:", self.connection_string_edit) self.description_edit = QTextEdit() self.description_edit.setMaximumHeight(80) form_layout.addRow("Description:", self.description_edit) # Simulation mode checkbox self.simulation_checkbox = QCheckBox("Simulation Mode") self.simulation_checkbox.setToolTip("Enable to create a simulated device for testing") self.simulation_checkbox.stateChanged.connect(self._on_simulation_mode_changed) form_layout.addRow("", self.simulation_checkbox) def _create_buttons(self, layout: QVBoxLayout) -> None: """Create dialog buttons.""" buttons = QDialogButtonBox( QDialogButtonBox.StandardButton.Ok | QDialogButtonBox.StandardButton.Cancel ) buttons.accepted.connect(self._on_accept) buttons.rejected.connect(self.reject) layout.addWidget(buttons) def _setup_validation(self) -> None: """Set up form validation.""" self.name_edit.textChanged.connect(self._validate_form) self.connection_string_edit.textChanged.connect(self._validate_form) self.simulation_checkbox.stateChanged.connect(self._validate_form) def _on_simulation_mode_changed(self, state: int) -> None: """Handle simulation mode checkbox change.""" is_simulation = state == Qt.CheckState.Checked.value # Update connection type for simulation if is_simulation: # Set to simulation and disable connection string for i in range(self.connection_type_combo.count()): if self.connection_type_combo.itemText(i) == "SIMULATION": self.connection_type_combo.setCurrentIndex(i) break self.connection_string_edit.setPlaceholderText("Auto-generated for simulation") self.connection_string_edit.setEnabled(False) else: # Reset to TCPIP and enable connection string for i in range(self.connection_type_combo.count()): if self.connection_type_combo.itemText(i) == "TCPIP": self.connection_type_combo.setCurrentIndex(i) break self.connection_string_edit.setPlaceholderText("e.g., TCPIP::192.168.1.100::INSTR") self.connection_string_edit.setEnabled(True) def _validate_form(self) -> None: """Validate form inputs and enable/disable OK button.""" # Basic validation - could be expanded name_valid = bool(self.name_edit.text().strip()) # Connection validation depends on simulation mode is_simulation = self.simulation_checkbox.isChecked() if is_simulation: connection_valid = True # No connection string needed for simulation else: connection_valid = bool(self.connection_string_edit.text().strip()) # Enable OK button only if basic requirements are met button_box = self.findChild(QDialogButtonBox) if button_box: ok_button = button_box.button(QDialogButtonBox.StandardButton.Ok) if ok_button: ok_button.setEnabled(name_valid and connection_valid) def _on_accept(self) -> None: """Handle dialog acceptance with validation.""" if self._is_form_valid(): self.accept() else: QMessageBox.warning(self, "Validation Error", "Please fill in all required fields") def _is_form_valid(self) -> bool: """Check if the form data is valid.""" name_valid = bool(self.name_edit.text().strip()) # Connection validation depends on simulation mode if self.simulation_checkbox.isChecked(): connection_valid = True # No connection string needed for simulation else: connection_valid = bool(self.connection_string_edit.text().strip()) return name_valid and connection_valid def get_device(self) -> Device: """Get device instance from form data.""" # Handle simulation mode is_simulation = self.simulation_checkbox.isChecked() if is_simulation: # Auto-generate connection string for simulation connection_string = f"SIM::{self.name_edit.text().strip().upper().replace(' ', '_')}" connection_type = ConnectionType.SIMULATION else: connection_string = self.connection_string_edit.text().strip() connection_type = ConnectionType(self.connection_type_combo.currentText()) return Device( name=self.name_edit.text().strip(), manufacturer=self.manufacturer_edit.text().strip(), model=self.model_edit.text().strip(), connection_type=connection_type, connection_string=connection_string, description=self.description_edit.toPlainText().strip(), simulation_mode=is_simulation ) class DeviceDiscoveryThread(QThread): """Background thread for discovering available VISA devices. This thread performs device discovery without blocking the UI, emitting a signal when discovery is complete. """ devices_discovered = Signal(list) # List[str] of resource strings def __init__(self, device_manager: DeviceManager, parent: Optional[QWidget] = None) -> None: super().__init__(parent) self.device_manager = device_manager self.logger = logging.getLogger(self.__class__.__name__) def run(self) -> None: """Run device discovery in background thread.""" try: resources = self.device_manager.discover_devices() self.devices_discovered.emit(resources) except Exception as e: self.logger.error(f"Device discovery failed: {e}") self.devices_discovered.emit([]) class DiscoveryDialog(QDialog): """ Dialog for displaying discovered VISA devices and allowing user selection. This dialog shows a table of discovered device resources, allows identification of devices via SCPI commands, and enables batch adding of selected devices. Attributes: COLUMN_ADD (int): Index of the "Add" checkbox column COLUMN_RESOURCE (int): Index of the resource string column COLUMN_IDENTIFIED (int): Index of the identification column COLUMN_STATUS (int): Index of the status column """ # Table column indices COLUMN_ADD = 0 COLUMN_RESOURCE = 1 COLUMN_IDENTIFIED = 2 COLUMN_STATUS = 3 def __init__(self, resources: List[str], device_manager: DeviceManager, parent: Optional[QWidget] = None) -> None: """ Initialize the discovery dialog. Args: resources: List of discovered VISA resource strings device_manager: Device manager instance for device operations parent: Parent widget for the dialog """ super().__init__(parent) self.resources = resources self.device_manager = device_manager self.logger = logging.getLogger(self.__class__.__name__) self._setup_dialog_properties() self._setup_ui() def _setup_dialog_properties(self) -> None: """Configure basic dialog properties.""" self.setWindowTitle("Discovered Devices") self.setModal(True) self.resize(600, 400) def _setup_ui(self) -> None: """Set up the complete dialog user interface.""" layout = QVBoxLayout(self) # Add instruction label instruction_label = QLabel("Select devices to add:") instruction_label.setStyleSheet("font-weight: bold; margin-bottom: 10px;") layout.addWidget(instruction_label) # Create and setup device table self._create_device_table(layout) # Create action buttons self._create_action_buttons(layout) def _create_device_table(self, layout: QVBoxLayout) -> None: """Create and configure the device discovery table.""" self.device_table = QTableWidget() self.device_table.setColumnCount(4) self.device_table.setHorizontalHeaderLabels([ "Add", "Resource", "Identified As", "Status" ]) # Configure table appearance self._configure_table_appearance() layout.addWidget(self.device_table) # Populate with discovered resources self._populate_device_table() def _configure_table_appearance(self) -> None: """Configure table visual properties and column sizing.""" self.device_table.setAlternatingRowColors(True) self.device_table.setSelectionBehavior(QTableWidget.SelectionBehavior.SelectRows) # Configure column resizing header = self.device_table.horizontalHeader() header.setSectionResizeMode(self.COLUMN_ADD, QHeaderView.ResizeMode.ResizeToContents) header.setSectionResizeMode(self.COLUMN_RESOURCE, QHeaderView.ResizeMode.Stretch) header.setSectionResizeMode(self.COLUMN_IDENTIFIED, QHeaderView.ResizeMode.Stretch) header.setSectionResizeMode(self.COLUMN_STATUS, QHeaderView.ResizeMode.ResizeToContents) def _populate_device_table(self) -> None: """Populate the device table with discovered resources.""" self.device_table.setRowCount(len(self.resources)) for row_index, resource in enumerate(self.resources): self._create_table_row(row_index, resource) def _create_table_row(self, row_index: int, resource: str) -> None: """Create a single row in the device table.""" # Add selection checkbox self._create_checkbox_item(row_index) # Resource string (read-only) resource_item = QTableWidgetItem(resource) resource_item.setFlags(resource_item.flags() & ~Qt.ItemFlag.ItemIsEditable) self.device_table.setItem(row_index, self.COLUMN_RESOURCE, resource_item) # Identification status (initially unknown) identification_item = QTableWidgetItem("Not identified") identification_item.setFlags(identification_item.flags() & ~Qt.ItemFlag.ItemIsEditable) self.device_table.setItem(row_index, self.COLUMN_IDENTIFIED, identification_item) # Operation status status_item = QTableWidgetItem("Ready") status_item.setFlags(status_item.flags() & ~Qt.ItemFlag.ItemIsEditable) self.device_table.setItem(row_index, self.COLUMN_STATUS, status_item) def _create_checkbox_item(self, row_index: int) -> None: """Create checkbox item for device selection.""" checkbox_item = QTableWidgetItem() checkbox_item.setFlags(Qt.ItemFlag.ItemIsUserCheckable | Qt.ItemFlag.ItemIsEnabled) checkbox_item.setCheckState(Qt.CheckState.Unchecked) self.device_table.setItem(row_index, self.COLUMN_ADD, checkbox_item) def _create_action_buttons(self, layout: QVBoxLayout) -> None: """Create the action button section.""" button_layout = QHBoxLayout() layout.addLayout(button_layout) # Identify button self.identify_button = QPushButton("Identify Selected") self.identify_button.clicked.connect(self._on_identify_selected_clicked) self.identify_button.setToolTip("Attempt to identify selected devices via SCPI") button_layout.addWidget(self.identify_button) # Select all button select_all_button = QPushButton("Select All") select_all_button.clicked.connect(self._on_select_all_clicked) button_layout.addWidget(select_all_button) button_layout.addStretch() # Dialog buttons self.dialog_buttons = QDialogButtonBox( QDialogButtonBox.StandardButton.Ok | QDialogButtonBox.StandardButton.Cancel ) self.dialog_buttons.accepted.connect(self._on_add_selected_devices) self.dialog_buttons.rejected.connect(self.reject) button_layout.addWidget(self.dialog_buttons) def _on_select_all_clicked(self) -> None: """Handle select all button click.""" for row in range(self.device_table.rowCount()): checkbox_item = self.device_table.item(row, self.COLUMN_ADD) if checkbox_item: checkbox_item.setCheckState(Qt.CheckState.Checked) def _on_identify_selected_clicked(self) -> None: """Handle identify selected devices button click.""" selected_rows = self._get_selected_rows() if not selected_rows: QMessageBox.information( self, "No Selection", "Please select at least one device to identify." ) return # Disable identify button during operation self.identify_button.setEnabled(False) try: self._identify_devices_in_rows(selected_rows) finally: self.identify_button.setEnabled(True) def _get_selected_rows(self) -> List[int]: """Get list of row indices that are selected for identification.""" selected_rows = [] for row in range(self.device_table.rowCount()): checkbox_item = self.device_table.item(row, self.COLUMN_ADD) if checkbox_item and checkbox_item.checkState() == Qt.CheckState.Checked: selected_rows.append(row) return selected_rows def _identify_devices_in_rows(self, rows: List[int]) -> None: """Identify devices in the specified table rows.""" for row in rows: try: self._identify_device_in_row(row) except Exception as e: self.logger.error(f"Failed to identify device in row {row}: {e}") self._set_row_status(row, "Error", "red") def _identify_device_in_row(self, row: int) -> None: """Identify a single device in the specified table row.""" resource_item = self.device_table.item(row, self.COLUMN_RESOURCE) if not resource_item: return resource = resource_item.text() # Update status to show progress self._set_row_status(row, "Identifying...", "blue") # Attempt device identification device = self.device_manager.auto_identify_device(resource) if device: self._handle_successful_identification(row, device) else: self._handle_failed_identification(row) def _handle_successful_identification(self, row: int, device: Device) -> None: """Handle successful device identification.""" # Create display name for identified device device_name = self._create_device_display_name(device) # Update table with identification results identification_item = self.device_table.item(row, self.COLUMN_IDENTIFIED) if identification_item: identification_item.setText(device_name) self._set_row_status(row, "Identified", "green") def _create_device_display_name(self, device: Device) -> str: """Create a display name for an identified device.""" name_parts = [device.manufacturer, device.model] device_name = " ".join(part for part in name_parts if part.strip()) return device_name if device_name else "Unknown Device" def _handle_failed_identification(self, row: int) -> None: """Handle failed device identification.""" identification_item = self.device_table.item(row, self.COLUMN_IDENTIFIED) if identification_item: identification_item.setText("Failed to identify") self._set_row_status(row, "Error", "red") def _set_row_status(self, row: int, status_text: str, color: str) -> None: """Set the status text and color for a table row.""" status_item = self.device_table.item(row, self.COLUMN_STATUS) if status_item: status_item.setText(status_text) # Use proper color mapping color_map = { 'red': Qt.GlobalColor.red, 'green': Qt.GlobalColor.green, 'blue': Qt.GlobalColor.blue, 'black': Qt.GlobalColor.black } status_item.setForeground(color_map.get(color, Qt.GlobalColor.black)) def _on_add_selected_devices(self) -> None: """Handle adding selected devices to the device manager.""" selected_rows = self._get_selected_rows() if not selected_rows: QMessageBox.information( self, "No Selection", "Please select at least one device to add." ) return added_count = self._add_devices_from_rows(selected_rows) if added_count > 0: QMessageBox.information( self, "Success", f"Successfully added {added_count} device(s)" ) self.accept() else: QMessageBox.warning( self, "No Devices Added", "No devices could be added. Please check the logs for details." ) def _add_devices_from_rows(self, rows: List[int]) -> int: """Add devices from the specified table rows and return count of successful additions.""" added_count = 0 for row in rows: try: if self._add_device_from_row(row): added_count += 1 except Exception as e: self.logger.error(f"Failed to add device from row {row}: {e}") return added_count def _add_device_from_row(self, row: int) -> bool: """Add a device from the specified table row.""" resource_item = self.device_table.item(row, self.COLUMN_RESOURCE) if not resource_item: return False resource = resource_item.text() # Attempt to identify and create device device = self.device_manager.auto_identify_device(resource) if not device: self.logger.warning(f"Could not identify device at {resource}") return False # Add device to manager if self.device_manager.create_device(device): self.logger.info(f"Successfully added device: {device.display_name}") return True else: self.logger.error(f"Failed to create device: {device.display_name}") return False class SimulationDialog(QDialog): """ Dialog for creating simulated devices This dialog allows users to create various types of simulated devices for testing and development without requiring real hardware. """ def __init__(self, device_manager, parent: Optional[QWidget] = None) -> None: super().__init__(parent) self.device_manager = device_manager self.logger = logging.getLogger(self.__class__.__name__) self._setup_dialog_properties() self._setup_ui() def _setup_dialog_properties(self) -> None: """Configure basic dialog properties.""" self.setWindowTitle("Create Simulated Device") self.setModal(True) self.resize(450, 350) def _setup_ui(self) -> None: """Set up the complete dialog user interface.""" layout = QVBoxLayout(self) # Add instruction label instruction_label = QLabel("Create a simulated device for testing without real hardware:") instruction_label.setStyleSheet("font-weight: bold; margin-bottom: 10px;") layout.addWidget(instruction_label) # Create form self._create_device_form(layout) # Create quick examples section self._create_examples_section(layout) # Create action buttons self._create_action_buttons(layout) def _create_device_form(self, layout: QVBoxLayout) -> None: """Create the device configuration form.""" form_group = QGroupBox("Device Configuration") form_layout = QFormLayout(form_group) layout.addWidget(form_group) # Device type selection self.device_type_combo = QComboBox() device_types = self.device_manager.get_available_simulation_types() for device_type in device_types: display_name = device_type.replace('_', ' ').title() self.device_type_combo.addItem(display_name, device_type) form_layout.addRow("Device Type:", self.device_type_combo) # Device name self.name_edit = QLineEdit() self.name_edit.setPlaceholderText("e.g., Lab Power Supply") form_layout.addRow("Device Name:", self.name_edit) # Manufacturer self.manufacturer_edit = QLineEdit() self.manufacturer_edit.setPlaceholderText("e.g., Test Equipment Inc.") form_layout.addRow("Manufacturer:", self.manufacturer_edit) # Model self.model_edit = QLineEdit() self.model_edit.setPlaceholderText("e.g., PS-3030") form_layout.addRow("Model:", self.model_edit) # Description self.description_edit = QTextEdit() self.description_edit.setMaximumHeight(60) self.description_edit.setPlaceholderText("Optional description...") form_layout.addRow("Description:", self.description_edit) def _create_examples_section(self, layout: QVBoxLayout) -> None: """Create quick examples section.""" examples_group = QGroupBox("Quick Examples") examples_layout = QVBoxLayout(examples_group) layout.addWidget(examples_group) examples_label = QLabel("Or create common example devices:") examples_layout.addWidget(examples_label) button_layout = QHBoxLayout() examples_layout.addLayout(button_layout) create_examples_btn = QPushButton("Create All Examples") create_examples_btn.clicked.connect(self._on_create_examples_clicked) create_examples_btn.setToolTip("Create Power Supply, Multimeter, and Function Generator examples") button_layout.addWidget(create_examples_btn) button_layout.addStretch() def _create_action_buttons(self, layout: QVBoxLayout) -> None: """Create the action button section.""" button_layout = QHBoxLayout() layout.addLayout(button_layout) button_layout.addStretch() # Dialog buttons self.dialog_buttons = QDialogButtonBox( QDialogButtonBox.StandardButton.Ok | QDialogButtonBox.StandardButton.Cancel ) self.dialog_buttons.accepted.connect(self._on_create_device) self.dialog_buttons.rejected.connect(self.reject) button_layout.addWidget(self.dialog_buttons) def _on_create_device(self) -> None: """Handle create device button click.""" try: # Get form data device_type = self.device_type_combo.currentData() name = self.name_edit.text().strip() manufacturer = self.manufacturer_edit.text().strip() model = self.model_edit.text().strip() # Validation if not name: QMessageBox.warning(self, "Validation Error", "Device name is required") return # Create simulated device device = self.device_manager.create_simulated_device( device_type=device_type, name=name, manufacturer=manufacturer, model=model ) if device: # Update description if provided description = self.description_edit.toPlainText().strip() if description: device.description = description self.device_manager.update_device(device) QMessageBox.information( self, "Success", f"Successfully created simulated device: {device.display_name}" ) self.accept() else: QMessageBox.warning( self, "Error", "Failed to create simulated device. Check the logs for details." ) except Exception as e: self.logger.error(f"Failed to create simulated device: {e}") QMessageBox.critical( self, "Error", f"An error occurred while creating the device:\n{str(e)}" ) def _on_create_examples_clicked(self) -> None: """Handle create examples button click.""" try: reply = QMessageBox.question( self, "Create Examples", "This will create example simulated devices:\n" "• Power Supply (TPS-3030)\n" "• Multimeter (DMM-1000)\n" "• Function Generator (FG-2500)\n\n" "Continue?", QMessageBox.StandardButton.Yes | QMessageBox.StandardButton.No, QMessageBox.StandardButton.Yes ) if reply == QMessageBox.StandardButton.Yes: examples = self.device_manager.create_simulation_examples() if examples: QMessageBox.information( self, "Success", f"Successfully created {len(examples)} example devices:\n" + "\n".join([f"• {device.display_name}" for device in examples]) ) self.accept() else: QMessageBox.warning( self, "Warning", "No example devices were created. They may already exist." ) except Exception as e: self.logger.error(f"Failed to create example devices: {e}") QMessageBox.critical( self, "Error", f"An error occurred while creating examples:\n{str(e)}" )