HuangMason320
ccd7cdd6b9
- Move test scripts to tests/ directory for better organization - Add improved YOLOv5 postprocessing with reference implementation - Update gitignore to exclude *.mflow files and include main.spec - Add debug capabilities and coordinate scaling improvements - Enhance multi-series support with proper validation - Add AGENTS.md documentation and example utilities 🤖 Generated with [Claude Code](https://claude.ai/code) Co-Authored-By: Claude <noreply@anthropic.com>
946 lines
41 KiB
Python
946 lines
41 KiB
Python
"""
|
|
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 with full MultiDongle postprocessing support."""
|
|
|
|
__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)
|
|
|
|
# Enhanced properties with MultiDongle postprocessing support
|
|
self.create_property('postprocess_type', 'fire_detection')
|
|
self.create_property('class_names', 'No Fire,Fire')
|
|
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)
|
|
self.create_property('enable_confidence_filter', True)
|
|
self.create_property('enable_nms', True)
|
|
self.create_property('coordinate_system', 'relative')
|
|
self.create_property('operations', 'filter,nms,format')
|
|
|
|
# Enhanced property options with MultiDongle integration
|
|
self._property_options = {
|
|
'postprocess_type': ['fire_detection', 'yolo_v3', 'yolo_v5', 'classification', 'raw_output'],
|
|
'class_names': {
|
|
'placeholder': 'comma-separated class names',
|
|
'description': 'Class names for model output (e.g., "No Fire,Fire" or "person,car,bicycle")'
|
|
},
|
|
'output_format': ['JSON', 'XML', 'CSV', 'Binary', 'MessagePack', 'YAML'],
|
|
'confidence_threshold': {'min': 0.0, 'max': 1.0, 'step': 0.01},
|
|
'nms_threshold': {'min': 0.0, 'max': 1.0, 'step': 0.01},
|
|
'max_detections': {'min': 1, 'max': 1000},
|
|
'enable_confidence_filter': {'type': 'bool', 'default': True},
|
|
'enable_nms': {'type': 'bool', 'default': True},
|
|
'coordinate_system': ['relative', 'absolute', 'center', 'custom'],
|
|
'operations': {'placeholder': 'comma-separated: filter,nms,format,validate,transform'}
|
|
}
|
|
|
|
# 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_multidongle_postprocess_options(self):
|
|
"""Create PostProcessorOptions from node configuration."""
|
|
try:
|
|
from ..functions.Multidongle import PostProcessType, PostProcessorOptions
|
|
|
|
postprocess_type_str = self.get_property('postprocess_type')
|
|
|
|
# Map string to enum
|
|
type_mapping = {
|
|
'fire_detection': PostProcessType.FIRE_DETECTION,
|
|
'yolo_v3': PostProcessType.YOLO_V3,
|
|
'yolo_v5': PostProcessType.YOLO_V5,
|
|
'classification': PostProcessType.CLASSIFICATION,
|
|
'raw_output': PostProcessType.RAW_OUTPUT
|
|
}
|
|
|
|
postprocess_type = type_mapping.get(postprocess_type_str, PostProcessType.FIRE_DETECTION)
|
|
|
|
# Parse class names
|
|
class_names_str = self.get_property('class_names')
|
|
class_names = [name.strip() for name in class_names_str.split(',') if name.strip()] if class_names_str else []
|
|
|
|
return PostProcessorOptions(
|
|
postprocess_type=postprocess_type,
|
|
threshold=self.get_property('confidence_threshold'),
|
|
class_names=class_names,
|
|
nms_threshold=self.get_property('nms_threshold'),
|
|
max_detections_per_class=self.get_property('max_detections')
|
|
)
|
|
except ImportError:
|
|
print("Warning: PostProcessorOptions not available")
|
|
return None
|
|
except Exception as e:
|
|
print(f"Error creating PostProcessorOptions: {e}")
|
|
return None
|
|
|
|
def get_postprocessing_config(self):
|
|
"""Get postprocessing configuration for pipeline execution."""
|
|
return {
|
|
'node_id': self.id,
|
|
'node_name': self.name(),
|
|
# MultiDongle postprocessing integration
|
|
'postprocess_type': self.get_property('postprocess_type'),
|
|
'class_names': self._parse_class_list(self.get_property('class_names')),
|
|
'multidongle_options': self.get_multidongle_postprocess_options(),
|
|
# Core postprocessing properties
|
|
'output_format': self.get_property('output_format'),
|
|
'confidence_threshold': self.get_property('confidence_threshold'),
|
|
'enable_confidence_filter': self.get_property('enable_confidence_filter'),
|
|
'nms_threshold': self.get_property('nms_threshold'),
|
|
'enable_nms': self.get_property('enable_nms'),
|
|
'max_detections': self.get_property('max_detections'),
|
|
'coordinate_system': self.get_property('coordinate_system'),
|
|
'operations': self._parse_operations_list(self.get_property('operations'))
|
|
}
|
|
|
|
def _parse_class_list(self, value_str):
|
|
"""Parse comma-separated class names or indices."""
|
|
if not value_str:
|
|
return []
|
|
return [x.strip() for x in value_str.split(',') if x.strip()]
|
|
|
|
def _parse_operations_list(self, operations_str):
|
|
"""Parse comma-separated operations list."""
|
|
if not operations_str:
|
|
return []
|
|
return [op.strip() for op in operations_str.split(',') if op.strip()]
|
|
|
|
def validate_configuration(self):
|
|
"""Validate the current node configuration."""
|
|
try:
|
|
# Check confidence threshold
|
|
confidence_threshold = self.get_property('confidence_threshold')
|
|
if not isinstance(confidence_threshold, (int, float)) or confidence_threshold < 0 or confidence_threshold > 1:
|
|
return False, "Confidence threshold must be between 0 and 1"
|
|
|
|
# Check NMS threshold
|
|
nms_threshold = self.get_property('nms_threshold')
|
|
if not isinstance(nms_threshold, (int, float)) or nms_threshold < 0 or nms_threshold > 1:
|
|
return False, "NMS threshold must be between 0 and 1"
|
|
|
|
# Check max detections
|
|
max_detections = self.get_property('max_detections')
|
|
if not isinstance(max_detections, int) or max_detections < 1:
|
|
return False, "Max detections must be at least 1"
|
|
|
|
# Validate operations string
|
|
operations = self.get_property('operations')
|
|
valid_operations = ['filter', 'nms', 'format', 'validate', 'transform', 'track', 'aggregate']
|
|
|
|
if operations:
|
|
ops_list = [op.strip() for op in operations.split(',')]
|
|
invalid_ops = [op for op in ops_list if op not in valid_operations]
|
|
if invalid_ops:
|
|
return False, f"Invalid operations: {', '.join(invalid_ops)}"
|
|
|
|
return True, ""
|
|
except Exception as e:
|
|
return False, f"Validation error: {str(e)}"
|
|
|
|
def get_display_properties(self):
|
|
"""Return properties that should be displayed in the UI panel."""
|
|
# Core properties that should always be visible for easy mode switching
|
|
return [
|
|
'postprocess_type',
|
|
'class_names',
|
|
'confidence_threshold',
|
|
'nms_threshold',
|
|
'output_format',
|
|
'enable_confidence_filter',
|
|
'enable_nms',
|
|
'max_detections'
|
|
]
|
|
|
|
|
|
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
|
|
} |