9.9 KiB
9.9 KiB
- Node Template for Metadata Integration Let's create a template that any node can use to properly integrate with the metadata system. This template will follow best practices and provide a solid foundation for new nodes.
Template for nodes that integrate with the metadata system
import os
import datetime
from typing import Dict, Any, List, Optional
# Import the metadata service
from eric_metadata.service import MetadataService
class MetadataEnabledNode:
"""Base template for a node that integrates with the metadata system"""
def __init__(self):
"""Initialize the node and metadata service"""
self.metadata_service = MetadataService(debug=False)
@classmethod
def INPUT_TYPES(cls):
"""Define node inputs with metadata options"""
return {
"required": {
"image": ("IMAGE",),
"input_filepath": ("STRING", {"default": ""}),
},
"optional": {
# Metadata writing options
"write_metadata_to_xmp": ("BOOLEAN", {"default": True}),
"embed_metadata": ("BOOLEAN", {"default": True}),
"write_text_file": ("BOOLEAN", {"default": False}),
"write_to_database": ("BOOLEAN", {"default": False}),
"debug_logging": ("BOOLEAN", {"default": False})
}
}
@classmethod
def RETURN_TYPES(cls):
"""Define node outputs"""
return ("IMAGE", "STRING")
@classmethod
def RETURN_NAMES(cls):
"""Define node output names"""
return ("image", "message")
@classmethod
def FUNCTION(cls):
"""Define the function name called"""
return "process"
@classmethod
def CATEGORY(cls):
"""Define node category"""
return "Eric's Nodes/Metadata"
def process(self, image, input_filepath="",
write_metadata_to_xmp=True, embed_metadata=True,
write_text_file=False, write_to_database=False,
debug_logging=False) -> tuple:
"""
Process image and write metadata
Args:
image: Input image
input_filepath: Path to image file
write_metadata_to_xmp: Whether to write to XMP sidecar
embed_metadata: Whether to embed metadata in image
write_text_file: Whether to write to text file
write_to_database: Whether to write to database
debug_logging: Whether to enable debug logging
Returns:
tuple: (image, message)
"""
try:
# Enable debug logging if requested
if debug_logging:
self.metadata_service.debug = True
# Skip metadata operations if no filepath provided
if not input_filepath:
return (image, "No filepath provided, skipping metadata operations")
# Perform your node's core functionality here
# ------------------------------------------
result_data = self._analyze_image(image)
# ------------------------------------------
# Create metadata structure according to standards
metadata = self._create_metadata(result_data)
# Set targets based on user preferences
targets = []
if write_metadata_to_xmp: targets.append('xmp')
if embed_metadata: targets.append('embedded')
if write_text_file: targets.append('txt')
if write_to_database: targets.append('db')
# Skip metadata writing if no targets selected
if not targets:
return (image, "No metadata targets selected, skipping metadata write")
# Set resource identifier (important for proper XMP handling)
filename = os.path.basename(input_filepath)
resource_uri = f"file:///{filename}"
self.metadata_service.set_resource_identifier(resource_uri)
# Write metadata
write_results = self.metadata_service.write_metadata(input_filepath, metadata, targets=targets)
# Prepare status message
success_targets = [t for t, success in write_results.items() if success]
fail_targets = [t for t, success in write_results.items() if not success]
if all(write_results.values()):
message = f"Successfully wrote metadata to all targets: {', '.join(targets)}"
elif any(write_results.values()):
message = f"Partially wrote metadata. Success: {', '.join(success_targets)}. Failed: {', '.join(fail_targets)}"
else:
message = "Failed to write metadata to any target"
return (image, message)
except Exception as e:
error_msg = f"Error in process: {str(e)}"
print(error_msg)
return (image, error_msg)
finally:
# Always ensure cleanup happens
self.cleanup()
def _analyze_image(self, image) -> Dict[str, Any]:
"""
Perform the core analysis on the image
Args:
image: Input image
Returns:
dict: Analysis results
"""
# OVERRIDE THIS METHOD IN YOUR SUBCLASS
# This is where you put your node's specific functionality
# Example placeholder implementation
return {
'score': 0.5,
'confidence': 0.9,
'timestamp': datetime.datetime.now().isoformat()
}
def _create_metadata(self, result_data: Dict[str, Any]) -> Dict[str, Any]:
"""
Create metadata structure from analysis results
Args:
result_data: Analysis results
Returns:
dict: Structured metadata
"""
# OVERRIDE THIS METHOD IN YOUR SUBCLASS
# This is where you convert your node's results into a standardized metadata structure
# Example placeholder implementation for a technical analysis node
metadata = {
'analysis': {
'technical': {
'my_metric': {
'score': result_data.get('score', 0.0),
'confidence': result_data.get('confidence', 1.0),
'higher_better': True,
'timestamp': result_data.get('timestamp', self.metadata_service.get_timestamp())
}
}
}
}
return metadata
def cleanup(self):
"""Clean up resources"""
if hasattr(self, 'metadata_service'):
self.metadata_service.cleanup()
def __del__(self):
"""Ensure cleanup on deletion"""
self.cleanup()
```python
Node Implementation Examples
Here are examples of how to implement specific node types using this template:
1. Technical Analysis Node Example (e.g., Blur Detection)
class BlurDetectionNode(MetadataEnabledNode):
"""Node for detecting image blur and storing results in metadata"""
@classmethod
def CATEGORY(cls):
return "Eric's Nodes/Technical Analysis"
def _analyze_image(self, image):
# Implement blur detection algorithm here
# This is just a placeholder
blur_score = self._calculate_blur(image)
return {
'blur_score': blur_score,
'confidence': 0.95,
'timestamp': datetime.datetime.now().isoformat()
}
def _calculate_blur(self, image):
# Actual blur detection implementation would go here
# This is just a placeholder
return 85.3
def _create_metadata(self, result_data):
# Structure metadata according to standards
metadata = {
'analysis': {
'technical': {
'blur': {
'score': float(result_data['blur_score']),
'higher_better': True, # Higher is sharper
'confidence': float(result_data['confidence']),
'timestamp': result_data['timestamp']
}
}
}
}
return metadata
2. Classification Node Example (e.g., Style Classification)
class StyleClassificationNode(MetadataEnabledNode):
"""Node for classifying image style and storing results in metadata"""
@classmethod
def CATEGORY(cls):
return "Eric's Nodes/Classification"
def _analyze_image(self, image):
# Implement style classification here
# This is just a placeholder
style, confidence = self._classify_style(image)
return {
'style': style,
'confidence': confidence,
'timestamp': datetime.datetime.now().isoformat()
}
def _classify_style(self, image):
# Actual style classification implementation would go here
# This is just a placeholder
return "minimalist", 0.87
def _create_metadata(self, result_data):
# Structure metadata according to standards
metadata = {
'analysis': {
'classification': {
'style': result_data['style'],
'style_confidence': float(result_data['confidence']),
'timestamp': result_data['timestamp']
}
}
}
return metadata
This template and examples should provide a solid foundation for integrating any node with your metadata system. The template handles all the common functionality (service initialization, cleanup, resource identification, error handling), while the specific node implementations only need to focus on their core functionality and proper metadata structure.