""" Base node functionality for the Cluster4NPU pipeline system. This module provides the common base functionality for all pipeline nodes, including property management, validation, and common node operations. Main Components: - BaseNodeWithProperties: Enhanced base node with business property support - Property validation and management utilities - Common node operations and interfaces Usage: from cluster4npu_ui.core.nodes.base_node import BaseNodeWithProperties class MyNode(BaseNodeWithProperties): def __init__(self): super().__init__() self.setup_properties() """ try: from NodeGraphQt import BaseNode NODEGRAPH_AVAILABLE = True except ImportError: # Fallback if NodeGraphQt is not available class BaseNode: def __init__(self): pass def create_property(self, name, value): pass def set_property(self, name, value): pass def get_property(self, name): return None NODEGRAPH_AVAILABLE = False from typing import Dict, Any, Optional, Union, List class BaseNodeWithProperties(BaseNode): """ Enhanced base node with business property support. This class extends the NodeGraphQt BaseNode to provide enhanced property management capabilities specifically for ML pipeline nodes. """ def __init__(self): super().__init__() self._property_options: Dict[str, Any] = {} self._property_validators: Dict[str, callable] = {} self._business_properties: Dict[str, Any] = {} def setup_properties(self): """Setup node-specific properties. Override in subclasses.""" pass def create_business_property(self, name: str, default_value: Any, options: Optional[Dict[str, Any]] = None): """ Create a business property with validation options. Args: name: Property name default_value: Default value for the property options: Validation and UI options dictionary """ self.create_property(name, default_value) self._business_properties[name] = default_value if options: self._property_options[name] = options def set_property_validator(self, name: str, validator: callable): """Set a custom validator for a property.""" self._property_validators[name] = validator def validate_property(self, name: str, value: Any) -> bool: """Validate a property value.""" if name in self._property_validators: return self._property_validators[name](value) # Default validation based on options if name in self._property_options: options = self._property_options[name] # Numeric range validation if 'min' in options and isinstance(value, (int, float)): if value < options['min']: return False if 'max' in options and isinstance(value, (int, float)): if value > options['max']: return False # Choice validation if isinstance(options, list) and value not in options: return False return True def get_property_options(self, name: str) -> Optional[Dict[str, Any]]: """Get property options for UI generation.""" return self._property_options.get(name) def get_business_properties(self) -> Dict[str, Any]: """Get all business properties.""" return self._business_properties.copy() def update_business_property(self, name: str, value: Any) -> bool: """Update a business property with validation.""" if self.validate_property(name, value): self._business_properties[name] = value self.set_property(name, value) return True return False def get_node_config(self) -> Dict[str, Any]: """Get node configuration for serialization.""" return { 'type': self.__class__.__name__, 'name': self.name(), 'properties': self.get_business_properties(), 'position': self.pos() } def load_node_config(self, config: Dict[str, Any]): """Load node configuration from serialized data.""" if 'name' in config: self.set_name(config['name']) if 'properties' in config: for name, value in config['properties'].items(): if name in self._business_properties: self.update_business_property(name, value) if 'position' in config: self.set_pos(*config['position']) def create_node_property_widget(node: BaseNodeWithProperties, prop_name: str, prop_value: Any, options: Optional[Dict[str, Any]] = None): """ Create appropriate widget for a node property. This function analyzes the property type and options to create the most appropriate Qt widget for editing the property value. Args: node: The node instance prop_name: Property name prop_value: Current property value options: Property options dictionary Returns: Appropriate Qt widget for editing the property """ from PyQt5.QtWidgets import (QLineEdit, QSpinBox, QDoubleSpinBox, QComboBox, QCheckBox, QFileDialog, QPushButton) if options is None: options = {} # File path property if options.get('type') == 'file_path': widget = QPushButton(str(prop_value) if prop_value else 'Select File...') def select_file(): file_filter = options.get('filter', 'All Files (*)') file_path, _ = QFileDialog.getOpenFileName(None, f'Select {prop_name}', str(prop_value) if prop_value else '', file_filter) if file_path: widget.setText(file_path) node.update_business_property(prop_name, file_path) widget.clicked.connect(select_file) return widget # Boolean property elif isinstance(prop_value, bool): widget = QCheckBox() widget.setChecked(prop_value) widget.stateChanged.connect( lambda state: node.update_business_property(prop_name, state == 2) ) return widget # Choice property elif isinstance(options, list): widget = QComboBox() widget.addItems(options) if prop_value in options: widget.setCurrentText(str(prop_value)) widget.currentTextChanged.connect( lambda text: node.update_business_property(prop_name, text) ) return widget # Numeric properties elif isinstance(prop_value, int): widget = QSpinBox() widget.setMinimum(options.get('min', -999999)) widget.setMaximum(options.get('max', 999999)) widget.setValue(prop_value) widget.valueChanged.connect( lambda value: node.update_business_property(prop_name, value) ) return widget elif isinstance(prop_value, float): widget = QDoubleSpinBox() widget.setMinimum(options.get('min', -999999.0)) widget.setMaximum(options.get('max', 999999.0)) widget.setDecimals(options.get('decimals', 2)) widget.setSingleStep(options.get('step', 0.1)) widget.setValue(prop_value) widget.valueChanged.connect( lambda value: node.update_business_property(prop_name, value) ) return widget # String property (default) else: widget = QLineEdit() widget.setText(str(prop_value)) widget.setPlaceholderText(options.get('placeholder', '')) widget.textChanged.connect( lambda text: node.update_business_property(prop_name, text) ) return widget