370 lines
13 KiB
Python
370 lines
13 KiB
Python
"""
|
|
Output node implementation for data sink operations.
|
|
|
|
This module provides the OutputNode class which handles various output destinations
|
|
including files, databases, APIs, and display systems for pipeline results.
|
|
|
|
Main Components:
|
|
- OutputNode: Core output data sink node implementation
|
|
- Output destination configuration and validation
|
|
- Format conversion and export functionality
|
|
|
|
Usage:
|
|
from cluster4npu_ui.core.nodes.output_node import OutputNode
|
|
|
|
node = OutputNode()
|
|
node.set_property('output_type', 'File')
|
|
node.set_property('destination', '/path/to/output.json')
|
|
"""
|
|
|
|
from .base_node import BaseNodeWithProperties
|
|
|
|
|
|
class OutputNode(BaseNodeWithProperties):
|
|
"""
|
|
Output data sink node for pipeline result export.
|
|
|
|
This node handles various output destinations including files, databases,
|
|
API endpoints, and display systems for processed pipeline results.
|
|
"""
|
|
|
|
__identifier__ = 'com.cluster.output_node'
|
|
NODE_NAME = 'Output Node'
|
|
|
|
def __init__(self):
|
|
super().__init__()
|
|
|
|
# Setup node connections (only input)
|
|
self.add_input('input', multi_input=False, color=(255, 140, 0))
|
|
self.set_color(255, 140, 0)
|
|
|
|
# Initialize properties
|
|
self.setup_properties()
|
|
|
|
def setup_properties(self):
|
|
"""Initialize output destination-specific properties."""
|
|
# Output type configuration
|
|
self.create_business_property('output_type', 'File', [
|
|
'File', 'API Endpoint', 'Database', 'Display', 'MQTT', 'WebSocket', 'Console'
|
|
])
|
|
|
|
# File output configuration
|
|
self.create_business_property('destination', '', {
|
|
'type': 'file_path',
|
|
'filter': 'Output files (*.json *.xml *.csv *.txt *.log)',
|
|
'description': 'Output file path or URL'
|
|
})
|
|
|
|
self.create_business_property('format', 'JSON', [
|
|
'JSON', 'XML', 'CSV', 'Binary', 'MessagePack', 'YAML', 'Parquet'
|
|
])
|
|
|
|
self.create_business_property('save_interval', 1.0, {
|
|
'min': 0.1,
|
|
'max': 60.0,
|
|
'step': 0.1,
|
|
'description': 'Save interval in seconds'
|
|
})
|
|
|
|
# File management
|
|
self.create_business_property('enable_rotation', False, {
|
|
'description': 'Enable file rotation based on size or time'
|
|
})
|
|
|
|
self.create_business_property('rotation_type', 'size', [
|
|
'size', 'time', 'count'
|
|
])
|
|
|
|
self.create_business_property('rotation_size_mb', 100, {
|
|
'min': 1,
|
|
'max': 1000,
|
|
'description': 'Rotation size in MB'
|
|
})
|
|
|
|
self.create_business_property('rotation_time_hours', 24, {
|
|
'min': 1,
|
|
'max': 168,
|
|
'description': 'Rotation time in hours'
|
|
})
|
|
|
|
# API endpoint configuration
|
|
self.create_business_property('api_url', '', {
|
|
'placeholder': 'https://api.example.com/data',
|
|
'description': 'API endpoint URL'
|
|
})
|
|
|
|
self.create_business_property('api_method', 'POST', [
|
|
'POST', 'PUT', 'PATCH'
|
|
])
|
|
|
|
self.create_business_property('api_headers', '', {
|
|
'placeholder': 'Authorization: Bearer token\\nContent-Type: application/json',
|
|
'description': 'API headers (one per line)'
|
|
})
|
|
|
|
self.create_business_property('api_timeout', 30, {
|
|
'min': 1,
|
|
'max': 300,
|
|
'description': 'API request timeout in seconds'
|
|
})
|
|
|
|
# Database configuration
|
|
self.create_business_property('db_connection_string', '', {
|
|
'placeholder': 'postgresql://user:pass@host:port/db',
|
|
'description': 'Database connection string'
|
|
})
|
|
|
|
self.create_business_property('db_table', '', {
|
|
'placeholder': 'results',
|
|
'description': 'Database table name'
|
|
})
|
|
|
|
self.create_business_property('db_batch_size', 100, {
|
|
'min': 1,
|
|
'max': 1000,
|
|
'description': 'Batch size for database inserts'
|
|
})
|
|
|
|
# MQTT configuration
|
|
self.create_business_property('mqtt_broker', '', {
|
|
'placeholder': 'mqtt://broker.example.com:1883',
|
|
'description': 'MQTT broker URL'
|
|
})
|
|
|
|
self.create_business_property('mqtt_topic', '', {
|
|
'placeholder': 'cluster4npu/results',
|
|
'description': 'MQTT topic for publishing'
|
|
})
|
|
|
|
self.create_business_property('mqtt_qos', 0, [
|
|
0, 1, 2
|
|
])
|
|
|
|
# Display configuration
|
|
self.create_business_property('display_type', 'console', [
|
|
'console', 'window', 'overlay', 'web'
|
|
])
|
|
|
|
self.create_business_property('display_format', 'pretty', [
|
|
'pretty', 'compact', 'raw'
|
|
])
|
|
|
|
# Buffer and queuing
|
|
self.create_business_property('enable_buffering', True, {
|
|
'description': 'Enable output buffering'
|
|
})
|
|
|
|
self.create_business_property('buffer_size', 1000, {
|
|
'min': 1,
|
|
'max': 10000,
|
|
'description': 'Buffer size in number of results'
|
|
})
|
|
|
|
self.create_business_property('flush_interval', 5.0, {
|
|
'min': 0.1,
|
|
'max': 60.0,
|
|
'step': 0.1,
|
|
'description': 'Buffer flush interval in seconds'
|
|
})
|
|
|
|
# Error handling
|
|
self.create_business_property('retry_on_error', True, {
|
|
'description': 'Retry on output errors'
|
|
})
|
|
|
|
self.create_business_property('max_retries', 3, {
|
|
'min': 0,
|
|
'max': 10,
|
|
'description': 'Maximum number of retries'
|
|
})
|
|
|
|
self.create_business_property('retry_delay', 1.0, {
|
|
'min': 0.1,
|
|
'max': 10.0,
|
|
'step': 0.1,
|
|
'description': 'Delay between retries in seconds'
|
|
})
|
|
|
|
def validate_configuration(self) -> tuple[bool, str]:
|
|
"""
|
|
Validate the current node configuration.
|
|
|
|
Returns:
|
|
Tuple of (is_valid, error_message)
|
|
"""
|
|
output_type = self.get_property('output_type')
|
|
|
|
# Validate based on output type
|
|
if output_type == 'File':
|
|
destination = self.get_property('destination')
|
|
if not destination:
|
|
return False, "Destination path is required for file output"
|
|
|
|
elif output_type == 'API Endpoint':
|
|
api_url = self.get_property('api_url')
|
|
if not api_url:
|
|
return False, "API URL is required for API endpoint output"
|
|
|
|
# Basic URL validation
|
|
if not (api_url.startswith('http://') or api_url.startswith('https://')):
|
|
return False, "Invalid API URL format"
|
|
|
|
elif output_type == 'Database':
|
|
db_connection = self.get_property('db_connection_string')
|
|
if not db_connection:
|
|
return False, "Database connection string is required"
|
|
|
|
db_table = self.get_property('db_table')
|
|
if not db_table:
|
|
return False, "Database table name is required"
|
|
|
|
elif output_type == 'MQTT':
|
|
mqtt_broker = self.get_property('mqtt_broker')
|
|
if not mqtt_broker:
|
|
return False, "MQTT broker URL is required"
|
|
|
|
mqtt_topic = self.get_property('mqtt_topic')
|
|
if not mqtt_topic:
|
|
return False, "MQTT topic is required"
|
|
|
|
# Validate save interval
|
|
save_interval = self.get_property('save_interval')
|
|
if not isinstance(save_interval, (int, float)) or save_interval <= 0:
|
|
return False, "Save interval must be greater than 0"
|
|
|
|
return True, ""
|
|
|
|
def get_output_config(self) -> dict:
|
|
"""
|
|
Get output configuration for pipeline execution.
|
|
|
|
Returns:
|
|
Dictionary containing output configuration
|
|
"""
|
|
return {
|
|
'node_id': self.id,
|
|
'node_name': self.name(),
|
|
'output_type': self.get_property('output_type'),
|
|
'destination': self.get_property('destination'),
|
|
'format': self.get_property('format'),
|
|
'save_interval': self.get_property('save_interval'),
|
|
'enable_rotation': self.get_property('enable_rotation'),
|
|
'rotation_type': self.get_property('rotation_type'),
|
|
'rotation_size_mb': self.get_property('rotation_size_mb'),
|
|
'rotation_time_hours': self.get_property('rotation_time_hours'),
|
|
'api_url': self.get_property('api_url'),
|
|
'api_method': self.get_property('api_method'),
|
|
'api_headers': self._parse_headers(self.get_property('api_headers')),
|
|
'api_timeout': self.get_property('api_timeout'),
|
|
'db_connection_string': self.get_property('db_connection_string'),
|
|
'db_table': self.get_property('db_table'),
|
|
'db_batch_size': self.get_property('db_batch_size'),
|
|
'mqtt_broker': self.get_property('mqtt_broker'),
|
|
'mqtt_topic': self.get_property('mqtt_topic'),
|
|
'mqtt_qos': self.get_property('mqtt_qos'),
|
|
'display_type': self.get_property('display_type'),
|
|
'display_format': self.get_property('display_format'),
|
|
'enable_buffering': self.get_property('enable_buffering'),
|
|
'buffer_size': self.get_property('buffer_size'),
|
|
'flush_interval': self.get_property('flush_interval'),
|
|
'retry_on_error': self.get_property('retry_on_error'),
|
|
'max_retries': self.get_property('max_retries'),
|
|
'retry_delay': self.get_property('retry_delay')
|
|
}
|
|
|
|
def _parse_headers(self, headers_str: str) -> dict:
|
|
"""Parse API headers from string format."""
|
|
headers = {}
|
|
if not headers_str:
|
|
return headers
|
|
|
|
for line in headers_str.split('\\n'):
|
|
line = line.strip()
|
|
if ':' in line:
|
|
key, value = line.split(':', 1)
|
|
headers[key.strip()] = value.strip()
|
|
|
|
return headers
|
|
|
|
def get_supported_formats(self) -> list[str]:
|
|
"""Get list of supported output formats."""
|
|
return ['JSON', 'XML', 'CSV', 'Binary', 'MessagePack', 'YAML', 'Parquet']
|
|
|
|
def get_estimated_throughput(self) -> dict:
|
|
"""
|
|
Estimate output throughput capabilities.
|
|
|
|
Returns:
|
|
Dictionary with throughput information
|
|
"""
|
|
output_type = self.get_property('output_type')
|
|
format_type = self.get_property('format')
|
|
|
|
# Estimated throughput (items per second) for different output types
|
|
throughput_map = {
|
|
'File': {
|
|
'JSON': 1000,
|
|
'XML': 800,
|
|
'CSV': 2000,
|
|
'Binary': 5000,
|
|
'MessagePack': 3000,
|
|
'YAML': 600,
|
|
'Parquet': 1500
|
|
},
|
|
'API Endpoint': {
|
|
'JSON': 100,
|
|
'XML': 80,
|
|
'CSV': 120,
|
|
'Binary': 150
|
|
},
|
|
'Database': {
|
|
'JSON': 500,
|
|
'XML': 400,
|
|
'CSV': 800,
|
|
'Binary': 1200
|
|
},
|
|
'MQTT': {
|
|
'JSON': 2000,
|
|
'XML': 1500,
|
|
'CSV': 3000,
|
|
'Binary': 5000
|
|
},
|
|
'Display': {
|
|
'JSON': 100,
|
|
'XML': 80,
|
|
'CSV': 120,
|
|
'Binary': 150
|
|
},
|
|
'Console': {
|
|
'JSON': 50,
|
|
'XML': 40,
|
|
'CSV': 60,
|
|
'Binary': 80
|
|
}
|
|
}
|
|
|
|
base_throughput = throughput_map.get(output_type, {}).get(format_type, 100)
|
|
|
|
# Adjust for buffering
|
|
if self.get_property('enable_buffering'):
|
|
buffer_multiplier = 1.5
|
|
else:
|
|
buffer_multiplier = 1.0
|
|
|
|
return {
|
|
'estimated_throughput': base_throughput * buffer_multiplier,
|
|
'output_type': output_type,
|
|
'format': format_type,
|
|
'buffering_enabled': self.get_property('enable_buffering'),
|
|
'buffer_size': self.get_property('buffer_size')
|
|
}
|
|
|
|
def requires_network(self) -> bool:
|
|
"""Check if the current output type requires network connectivity."""
|
|
output_type = self.get_property('output_type')
|
|
return output_type in ['API Endpoint', 'Database', 'MQTT', 'WebSocket']
|
|
|
|
def supports_real_time(self) -> bool:
|
|
"""Check if the current output type supports real-time output."""
|
|
output_type = self.get_property('output_type')
|
|
return output_type in ['Display', 'Console', 'MQTT', 'WebSocket', 'API Endpoint'] |