231 lines
8.0 KiB
Python
231 lines
8.0 KiB
Python
"""
|
|
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 |