""" 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. """ 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', '') # Original property options - exact match self._property_options = { 'dongle_series': ['520', '720', '1080', 'Custom'], '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'} } # 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 for Model nodes return ['model_path', 'dongle_series', 'num_dongles'] # Skip port_id 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 }