""" Exact node implementations matching the original UI.py properties. This module provides node implementations that exactly match the original properties and behavior from the monolithic UI.py file. """ import os try: from NodeGraphQt import BaseNode NODEGRAPH_AVAILABLE = True except ImportError: NODEGRAPH_AVAILABLE = False # Create a mock base class class BaseNode: def __init__(self): pass class ExactInputNode(BaseNode): """Input data source node - exact match to original.""" __identifier__ = 'com.cluster.input_node.ExactInputNode' NODE_NAME = 'Input Node' def __init__(self): super().__init__() if NODEGRAPH_AVAILABLE: # Setup node connections - exact match self.add_output('output', color=(0, 255, 0)) self.set_color(83, 133, 204) # Original properties - exact match self.create_property('source_type', 'Camera') self.create_property('device_id', 0) self.create_property('source_path', '') self.create_property('resolution', '1920x1080') self.create_property('fps', 30) # Original property options - exact match self._property_options = { 'source_type': ['Camera', 'Microphone', 'File', 'RTSP Stream', 'HTTP Stream'], 'device_id': {'min': 0, 'max': 10}, 'resolution': ['640x480', '1280x720', '1920x1080', '3840x2160', 'Custom'], 'fps': {'min': 1, 'max': 120}, 'source_path': {'type': 'file_path', 'filter': 'Media files (*.mp4 *.avi *.mov *.mkv *.wav *.mp3)'} } # Create custom properties dictionary for UI compatibility self._populate_custom_properties() def _populate_custom_properties(self): """Populate the custom properties dictionary for UI compatibility.""" if not NODEGRAPH_AVAILABLE: return # Get all business properties defined in _property_options business_props = list(self._property_options.keys()) # Create custom dictionary containing current property values custom_dict = {} for prop_name in business_props: try: # Skip 'custom' property to avoid infinite recursion if prop_name != 'custom': custom_dict[prop_name] = self.get_property(prop_name) except: # If property doesn't exist, skip it pass # Create the custom property that contains all business properties self.create_property('custom', custom_dict) def get_business_properties(self): """Get all business properties for serialization.""" if not NODEGRAPH_AVAILABLE: return {} properties = {} for prop_name in self._property_options.keys(): try: properties[prop_name] = self.get_property(prop_name) except: pass return properties def get_display_properties(self): """Return properties that should be displayed in the UI panel.""" # Customize which properties appear in the properties panel # You can reorder, filter, or modify this list return ['source_type', 'resolution', 'fps'] # Only show these 3 properties class ExactModelNode(BaseNode): """Model node for ML inference - exact match to original.""" __identifier__ = 'com.cluster.model_node.ExactModelNode' NODE_NAME = 'Model Node' def __init__(self): super().__init__() if NODEGRAPH_AVAILABLE: # Setup node connections - exact match self.add_input('input', multi_input=False, color=(255, 140, 0)) self.add_output('output', color=(0, 255, 0)) self.set_color(65, 84, 102) # Original properties - exact match self.create_property('model_path', '') self.create_property('scpu_fw_path', '') self.create_property('ncpu_fw_path', '') self.create_property('dongle_series', '520') self.create_property('num_dongles', 1) self.create_property('port_id', '') self.create_property('upload_fw', True) # Multi-series properties self.create_property('multi_series_mode', False) self.create_property('assets_folder', '') self.create_property('enabled_series', ['520', '720']) # Series-specific port ID configurations self.create_property('kl520_port_ids', '') self.create_property('kl720_port_ids', '') self.create_property('kl630_port_ids', '') self.create_property('kl730_port_ids', '') # self.create_property('kl540_port_ids', '') self.create_property('max_queue_size', 100) self.create_property('result_buffer_size', 1000) self.create_property('batch_size', 1) self.create_property('enable_preprocessing', False) self.create_property('enable_postprocessing', False) # Original property options - exact match self._property_options = { 'dongle_series': ['520', '720'], 'num_dongles': {'min': 1, 'max': 16}, 'model_path': {'type': 'file_path', 'filter': 'NEF Model files (*.nef)'}, 'scpu_fw_path': {'type': 'file_path', 'filter': 'SCPU Firmware files (*.bin)'}, 'ncpu_fw_path': {'type': 'file_path', 'filter': 'NCPU Firmware files (*.bin)'}, 'port_id': {'placeholder': 'e.g., 8080 or auto'}, 'upload_fw': {'type': 'bool', 'default': True, 'description': 'Upload firmware to dongle if needed'}, # Multi-series property options 'multi_series_mode': {'type': 'bool', 'default': False, 'description': 'Enable multi-series dongle support'}, 'assets_folder': {'type': 'file_path', 'filter': 'Folder', 'mode': 'directory'}, 'enabled_series': {'type': 'list', 'options': ['520', '720', '630', '730', '540'], 'default': ['520', '720']}, # Series-specific port ID options 'kl520_port_ids': {'placeholder': 'e.g., 28,32 (comma-separated port IDs for KL520)', 'description': 'Port IDs for KL520 dongles'}, 'kl720_port_ids': {'placeholder': 'e.g., 30,34 (comma-separated port IDs for KL720)', 'description': 'Port IDs for KL720 dongles'}, 'kl630_port_ids': {'placeholder': 'e.g., 36,38 (comma-separated port IDs for KL630)', 'description': 'Port IDs for KL630 dongles'}, 'kl730_port_ids': {'placeholder': 'e.g., 40,42 (comma-separated port IDs for KL730)', 'description': 'Port IDs for KL730 dongles'}, # 'kl540_port_ids': {'placeholder': 'e.g., 44,46 (comma-separated port IDs for KL540)', 'description': 'Port IDs for KL540 dongles'}, 'max_queue_size': {'min': 1, 'max': 1000, 'default': 100}, 'result_buffer_size': {'min': 100, 'max': 10000, 'default': 1000}, 'batch_size': {'min': 1, 'max': 32, 'default': 1}, 'enable_preprocessing': {'type': 'bool', 'default': False}, 'enable_postprocessing': {'type': 'bool', 'default': False} } # Create custom properties dictionary for UI compatibility self._populate_custom_properties() # Set up custom property handlers for folder selection if NODEGRAPH_AVAILABLE: self._setup_custom_property_handlers() def _populate_custom_properties(self): """Populate the custom properties dictionary for UI compatibility.""" if not NODEGRAPH_AVAILABLE: return # Get all business properties defined in _property_options business_props = list(self._property_options.keys()) # Create custom dictionary containing current property values custom_dict = {} for prop_name in business_props: try: # Skip 'custom' property to avoid infinite recursion if prop_name != 'custom': custom_dict[prop_name] = self.get_property(prop_name) except: # If property doesn't exist, skip it pass # Create the custom property that contains all business properties self.create_property('custom', custom_dict) def get_business_properties(self): """Get all business properties for serialization.""" if not NODEGRAPH_AVAILABLE: return {} properties = {} for prop_name in self._property_options.keys(): try: properties[prop_name] = self.get_property(prop_name) except: pass return properties def get_display_properties(self): """Return properties that should be displayed in the UI panel.""" if not NODEGRAPH_AVAILABLE: return [] # Base properties that are always shown base_props = ['multi_series_mode'] try: # Check if we're in multi-series mode multi_series_mode = self.get_property('multi_series_mode') if multi_series_mode: # Multi-series mode: show multi-series specific properties multi_props = ['assets_folder', 'enabled_series'] # Add port ID configurations for enabled series try: enabled_series = self.get_property('enabled_series') or [] for series in enabled_series: port_prop = f'kl{series}_port_ids' if port_prop not in multi_props: # Avoid duplicates multi_props.append(port_prop) except: pass # If can't get enabled_series, just show basic properties # Add other multi-series properties multi_props.extend([ 'max_queue_size', 'result_buffer_size', 'batch_size', 'enable_preprocessing', 'enable_postprocessing' ]) return base_props + multi_props else: # Single-series mode: show traditional properties return base_props + [ 'model_path', 'scpu_fw_path', 'ncpu_fw_path', 'dongle_series', 'num_dongles', 'port_id', 'upload_fw' ] except: # Fallback to single-series mode if property access fails return base_props + [ 'model_path', 'scpu_fw_path', 'ncpu_fw_path', 'dongle_series', 'num_dongles', 'port_id', 'upload_fw' ] def get_inference_config(self): """Get configuration for inference pipeline""" if not NODEGRAPH_AVAILABLE: return {} try: multi_series_mode = self.get_property('multi_series_mode') if multi_series_mode: # Multi-series configuration with series-specific port IDs config = { 'multi_series_mode': True, 'assets_folder': self.get_property('assets_folder'), 'enabled_series': self.get_property('enabled_series'), 'max_queue_size': self.get_property('max_queue_size'), 'result_buffer_size': self.get_property('result_buffer_size'), 'batch_size': self.get_property('batch_size'), 'enable_preprocessing': self.get_property('enable_preprocessing'), 'enable_postprocessing': self.get_property('enable_postprocessing') } # Build multi-series config for MultiDongle multi_series_config = self._build_multi_series_config() if multi_series_config: config['multi_series_config'] = multi_series_config return config else: # Single-series configuration return { 'multi_series_mode': False, 'model_path': self.get_property('model_path'), 'scpu_fw_path': self.get_property('scpu_fw_path'), 'ncpu_fw_path': self.get_property('ncpu_fw_path'), 'dongle_series': self.get_property('dongle_series'), 'num_dongles': self.get_property('num_dongles'), 'port_id': self.get_property('port_id'), 'upload_fw': self.get_property('upload_fw') } except: # Fallback to single-series configuration return { 'multi_series_mode': False, 'model_path': self.get_property('model_path', ''), 'scpu_fw_path': self.get_property('scpu_fw_path', ''), 'ncpu_fw_path': self.get_property('ncpu_fw_path', ''), 'dongle_series': self.get_property('dongle_series', '520'), 'num_dongles': self.get_property('num_dongles', 1), 'port_id': self.get_property('port_id', ''), 'upload_fw': self.get_property('upload_fw', True) } def _build_multi_series_config(self): """Build multi-series configuration for MultiDongle""" try: enabled_series = self.get_property('enabled_series') or [] assets_folder = self.get_property('assets_folder') or '' if not enabled_series: return None multi_series_config = {} for series in enabled_series: # Get port IDs for this series port_ids_str = self.get_property(f'kl{series}_port_ids') or '' if not port_ids_str.strip(): continue # Skip series without port IDs # Parse port IDs (comma-separated string to list of integers) try: port_ids = [int(pid.strip()) for pid in port_ids_str.split(',') if pid.strip()] if not port_ids: continue except ValueError: print(f"Warning: Invalid port IDs for KL{series}: {port_ids_str}") continue # Build series configuration series_config = { "port_ids": port_ids } # Add model path if assets folder is configured if assets_folder: import os model_folder = os.path.join(assets_folder, 'Models', f'KL{series}') if os.path.exists(model_folder): # Look for .nef files in the model folder nef_files = [f for f in os.listdir(model_folder) if f.endswith('.nef')] if nef_files: series_config["model_path"] = os.path.join(model_folder, nef_files[0]) # Add firmware paths if available firmware_folder = os.path.join(assets_folder, 'Firmware', f'KL{series}') if os.path.exists(firmware_folder): scpu_path = os.path.join(firmware_folder, 'fw_scpu.bin') ncpu_path = os.path.join(firmware_folder, 'fw_ncpu.bin') if os.path.exists(scpu_path) and os.path.exists(ncpu_path): series_config["firmware_paths"] = { "scpu": scpu_path, "ncpu": ncpu_path } multi_series_config[f'KL{series}'] = series_config return multi_series_config if multi_series_config else None except Exception as e: print(f"Error building multi-series config: {e}") return None def get_hardware_requirements(self): """Get hardware requirements for this node""" if not NODEGRAPH_AVAILABLE: return {} try: multi_series_mode = self.get_property('multi_series_mode') if multi_series_mode: enabled_series = self.get_property('enabled_series') return { 'multi_series_mode': True, 'required_series': enabled_series, 'estimated_dongles': len(enabled_series) * 2 # Assume 2 dongles per series } else: dongle_series = self.get_property('dongle_series') num_dongles = self.get_property('num_dongles') return { 'multi_series_mode': False, 'required_series': [f'KL{dongle_series}'], 'estimated_dongles': num_dongles } except: return {'multi_series_mode': False, 'required_series': ['KL520'], 'estimated_dongles': 1} def _setup_custom_property_handlers(self): """Setup custom property handlers, especially for folder selection.""" try: # For assets_folder, we want to trigger folder selection dialog # This might require custom widget or property handling # For now, we'll use the standard approach but add validation # You can override the property widget here if needed # This is a placeholder for custom folder selection implementation pass except Exception as e: print(f"Warning: Could not setup custom property handlers: {e}") def select_assets_folder(self): """Method to open folder selection dialog for assets folder using improved utility.""" if not NODEGRAPH_AVAILABLE: return "" try: from utils.folder_dialog import select_assets_folder # Get current folder path as initial directory current_folder = "" try: current_folder = self.get_property('assets_folder') or "" except: pass # Use the specialized assets folder dialog with validation result = select_assets_folder(initial_dir=current_folder) if result['path']: # Set the property if NODEGRAPH_AVAILABLE: self.set_property('assets_folder', result['path']) # Print validation results if result['valid']: print(f"✓ Valid Assets folder set to: {result['path']}") if 'details' in result and 'available_series' in result['details']: series = result['details']['available_series'] print(f" Available series: {', '.join(series)}") else: print(f"⚠ Assets folder set to: {result['path']}") print(f" Warning: {result['message']}") print(" Expected structure: Assets/Firmware/ and Assets/Models/ with series subfolders") return result['path'] else: print("No folder selected") return "" except ImportError: print("utils.folder_dialog not available, falling back to simple input") # Fallback to manual input folder_path = input("Enter Assets folder path: ").strip() if folder_path and NODEGRAPH_AVAILABLE: self.set_property('assets_folder', folder_path) return folder_path except Exception as e: print(f"Error selecting assets folder: {e}") return "" def _validate_assets_folder(self, folder_path): """Validate that the assets folder has the expected structure.""" try: import os # Check if Firmware and Models folders exist firmware_path = os.path.join(folder_path, 'Firmware') models_path = os.path.join(folder_path, 'Models') has_firmware = os.path.exists(firmware_path) and os.path.isdir(firmware_path) has_models = os.path.exists(models_path) and os.path.isdir(models_path) if not (has_firmware and has_models): return False # Check for at least one series subfolder expected_series = ['KL520', 'KL720', 'KL630', 'KL730'] firmware_series = [d for d in os.listdir(firmware_path) if os.path.isdir(os.path.join(firmware_path, d)) and d in expected_series] models_series = [d for d in os.listdir(models_path) if os.path.isdir(os.path.join(models_path, d)) and d in expected_series] # At least one series should exist in both firmware and models return len(firmware_series) > 0 and len(models_series) > 0 except Exception as e: print(f"Error validating assets folder: {e}") return False def get_assets_folder_info(self): """Get information about the configured assets folder.""" if not NODEGRAPH_AVAILABLE: return {} try: folder_path = self.get_property('assets_folder') if not folder_path: return {'status': 'not_set', 'message': 'No assets folder selected'} if not os.path.exists(folder_path): return {'status': 'invalid', 'message': 'Selected folder does not exist'} info = {'status': 'valid', 'path': folder_path, 'series': []} # Get available series firmware_path = os.path.join(folder_path, 'Firmware') models_path = os.path.join(folder_path, 'Models') if os.path.exists(firmware_path): firmware_series = [d for d in os.listdir(firmware_path) if os.path.isdir(os.path.join(firmware_path, d))] info['firmware_series'] = firmware_series if os.path.exists(models_path): models_series = [d for d in os.listdir(models_path) if os.path.isdir(os.path.join(models_path, d))] info['models_series'] = models_series # Find common series if 'firmware_series' in info and 'models_series' in info: common_series = list(set(info['firmware_series']) & set(info['models_series'])) info['available_series'] = common_series if not common_series: info['status'] = 'incomplete' info['message'] = 'No series found with both firmware and models' return info except Exception as e: return {'status': 'error', 'message': f'Error reading assets folder: {e}'} def validate_configuration(self) -> tuple[bool, str]: """ Validate the current node configuration. Returns: Tuple of (is_valid, error_message) """ if not NODEGRAPH_AVAILABLE: return True, "" try: multi_series_mode = self.get_property('multi_series_mode') if multi_series_mode: # Multi-series validation enabled_series = self.get_property('enabled_series') if not enabled_series: return False, "No series enabled in multi-series mode" # Check if at least one series has port IDs configured has_valid_series = False for series in enabled_series: port_ids_str = self.get_property(f'kl{series}_port_ids', '') if port_ids_str and port_ids_str.strip(): # Validate port ID format try: port_ids = [int(pid.strip()) for pid in port_ids_str.split(',') if pid.strip()] if port_ids: has_valid_series = True print(f"Valid series config found for KL{series}: ports {port_ids}") except ValueError: print(f"Warning: Invalid port ID format for KL{series}: {port_ids_str}") continue if not has_valid_series: return False, "At least one series must have valid port IDs configured" # Assets folder validation (optional for multi-series) assets_folder = self.get_property('assets_folder') if assets_folder: if not os.path.exists(assets_folder): print(f"Warning: Assets folder does not exist: {assets_folder}") else: # Validate assets folder structure if provided assets_info = self.get_assets_folder_info() if assets_info.get('status') == 'error': print(f"Warning: Assets folder issue: {assets_info.get('message', 'Unknown error')}") print("Multi-series mode validation passed") return True, "" else: # Single-series validation (legacy) model_path = self.get_property('model_path') if not model_path: return False, "Model path is required" if not os.path.exists(model_path): return False, f"Model file does not exist: {model_path}" # Check dongle series dongle_series = self.get_property('dongle_series') if dongle_series not in ['520', '720', '1080', 'Custom']: return False, f"Invalid dongle series: {dongle_series}" # Check number of dongles num_dongles = self.get_property('num_dongles') if not isinstance(num_dongles, int) or num_dongles < 1: return False, "Number of dongles must be at least 1" return True, "" except Exception as e: return False, f"Validation error: {str(e)}" class ExactPreprocessNode(BaseNode): """Preprocessing node - exact match to original.""" __identifier__ = 'com.cluster.preprocess_node.ExactPreprocessNode' NODE_NAME = 'Preprocess Node' def __init__(self): super().__init__() if NODEGRAPH_AVAILABLE: # Setup node connections - exact match self.add_input('input', multi_input=False, color=(255, 140, 0)) self.add_output('output', color=(0, 255, 0)) self.set_color(45, 126, 72) # Original properties - exact match self.create_property('resize_width', 640) self.create_property('resize_height', 480) self.create_property('normalize', True) self.create_property('crop_enabled', False) self.create_property('operations', 'resize,normalize') # Original property options - exact match self._property_options = { 'resize_width': {'min': 64, 'max': 4096}, 'resize_height': {'min': 64, 'max': 4096}, 'operations': {'placeholder': 'comma-separated: resize,normalize,crop'} } # Create custom properties dictionary for UI compatibility self._populate_custom_properties() def _populate_custom_properties(self): """Populate the custom properties dictionary for UI compatibility.""" if not NODEGRAPH_AVAILABLE: return # Get all business properties defined in _property_options business_props = list(self._property_options.keys()) # Create custom dictionary containing current property values custom_dict = {} for prop_name in business_props: try: # Skip 'custom' property to avoid infinite recursion if prop_name != 'custom': custom_dict[prop_name] = self.get_property(prop_name) except: # If property doesn't exist, skip it pass # Create the custom property that contains all business properties self.create_property('custom', custom_dict) def get_business_properties(self): """Get all business properties for serialization.""" if not NODEGRAPH_AVAILABLE: return {} properties = {} for prop_name in self._property_options.keys(): try: properties[prop_name] = self.get_property(prop_name) except: pass return properties class ExactPostprocessNode(BaseNode): """Postprocessing node - exact match to original.""" __identifier__ = 'com.cluster.postprocess_node.ExactPostprocessNode' NODE_NAME = 'Postprocess Node' def __init__(self): super().__init__() if NODEGRAPH_AVAILABLE: # Setup node connections - exact match self.add_input('input', multi_input=False, color=(255, 140, 0)) self.add_output('output', color=(0, 255, 0)) self.set_color(153, 51, 51) # Original properties - exact match self.create_property('output_format', 'JSON') self.create_property('confidence_threshold', 0.5) self.create_property('nms_threshold', 0.4) self.create_property('max_detections', 100) # Original property options - exact match self._property_options = { 'output_format': ['JSON', 'XML', 'CSV', 'Binary'], 'confidence_threshold': {'min': 0.0, 'max': 1.0, 'step': 0.1}, 'nms_threshold': {'min': 0.0, 'max': 1.0, 'step': 0.1}, 'max_detections': {'min': 1, 'max': 1000} } # Create custom properties dictionary for UI compatibility self._populate_custom_properties() def _populate_custom_properties(self): """Populate the custom properties dictionary for UI compatibility.""" if not NODEGRAPH_AVAILABLE: return # Get all business properties defined in _property_options business_props = list(self._property_options.keys()) # Create custom dictionary containing current property values custom_dict = {} for prop_name in business_props: try: # Skip 'custom' property to avoid infinite recursion if prop_name != 'custom': custom_dict[prop_name] = self.get_property(prop_name) except: # If property doesn't exist, skip it pass # Create the custom property that contains all business properties self.create_property('custom', custom_dict) def get_business_properties(self): """Get all business properties for serialization.""" if not NODEGRAPH_AVAILABLE: return {} properties = {} for prop_name in self._property_options.keys(): try: properties[prop_name] = self.get_property(prop_name) except: pass return properties class ExactOutputNode(BaseNode): """Output data sink node - exact match to original.""" __identifier__ = 'com.cluster.output_node.ExactOutputNode' NODE_NAME = 'Output Node' def __init__(self): super().__init__() if NODEGRAPH_AVAILABLE: # Setup node connections - exact match self.add_input('input', multi_input=False, color=(255, 140, 0)) self.set_color(255, 140, 0) # Original properties - exact match self.create_property('output_type', 'File') self.create_property('destination', '') self.create_property('format', 'JSON') self.create_property('save_interval', 1.0) # Original property options - exact match self._property_options = { 'output_type': ['File', 'API Endpoint', 'Database', 'Display', 'MQTT'], 'format': ['JSON', 'XML', 'CSV', 'Binary'], 'destination': {'type': 'file_path', 'filter': 'Output files (*.json *.xml *.csv *.txt)'}, 'save_interval': {'min': 0.1, 'max': 60.0, 'step': 0.1} } # Create custom properties dictionary for UI compatibility self._populate_custom_properties() def _populate_custom_properties(self): """Populate the custom properties dictionary for UI compatibility.""" if not NODEGRAPH_AVAILABLE: return # Get all business properties defined in _property_options business_props = list(self._property_options.keys()) # Create custom dictionary containing current property values custom_dict = {} for prop_name in business_props: try: # Skip 'custom' property to avoid infinite recursion if prop_name != 'custom': custom_dict[prop_name] = self.get_property(prop_name) except: # If property doesn't exist, skip it pass # Create the custom property that contains all business properties self.create_property('custom', custom_dict) def get_business_properties(self): """Get all business properties for serialization.""" if not NODEGRAPH_AVAILABLE: return {} properties = {} for prop_name in self._property_options.keys(): try: properties[prop_name] = self.get_property(prop_name) except: pass return properties # Export the exact nodes EXACT_NODE_TYPES = { 'Input Node': ExactInputNode, 'Model Node': ExactModelNode, 'Preprocess Node': ExactPreprocessNode, 'Postprocess Node': ExactPostprocessNode, 'Output Node': ExactOutputNode }