3. 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 ```python 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.