utils/export_utils.py

"""
Export Utilities for EmotionMirror Application

This module provides functions for exporting analysis data to various formats,
including CSV, JSON, and visual reports.
"""

import os
import csv
import json
import logging
from typing import Dict, List, Any, Optional
from datetime import datetime
import pandas as pd
import base64
from io import BytesIO, StringIO

# Configure logging
logger = logging.getLogger(__name__)

def export_to_json(data: Dict[str, Any], filepath: Optional[str] = None) -> str:
    """
    Export data to JSON format.
    
    Args:
        data: Data to export
        filepath: Optional path to save the JSON file
        
    Returns:
        JSON string if filepath is None, else filepath where saved
    """
    try:
        json_str = json.dumps(data, indent=4)
        
        if filepath:
            with open(filepath, 'w', encoding='utf-8') as f:
                f.write(json_str)
            logger.info(f"Data exported to JSON file: {filepath}")
            return filepath
        
        return json_str
    
    except Exception as e:
        logger.error(f"Error exporting data to JSON: {e}")
        raise

def export_to_csv(data: Dict[str, Any], filepath: Optional[str] = None) -> str:
    """
    Export analysis data to CSV format.
    
    Args:
        data: Analysis data to export
        filepath: Optional path to save the CSV file
        
    Returns:
        CSV string if filepath is None, else filepath where saved
    """
    try:
        # Flatten the data structure for CSV format
        rows = []
        
        for analysis in data.get('analyses', []):
            base_row = {
                'analysis_id': analysis.get('id'),
                'session_id': analysis.get('session_id'),
                'timestamp': analysis.get('timestamp'),
                'image_path': analysis.get('image_path'),
                'face_count': analysis.get('face_count')
            }
            
            # Add tags if present
            if analysis.get('tags'):
                base_row['tags'] = ','.join(analysis['tags'])
            
            # Add face-specific data
            for i, face in enumerate(analysis.get('faces', [])):
                row = base_row.copy()
                row['face_index'] = i
                row['emotion'] = face.get('emotion')
                row['confidence'] = face.get('confidence')
                
                # Add feature values
                for feature, value in face.get('features', {}).items():
                    row[f'feature_{feature}'] = value
                
                # Add emotion values
                for emotion, value in face.get('emotions', {}).items():
                    row[f'emotion_{emotion}'] = value
                
                rows.append(row)
        
        if not rows:
            logger.warning("No data to export to CSV")
            return "" if filepath is None else filepath
        
        # Create a DataFrame and export to CSV
        df = pd.DataFrame(rows)
        
        if filepath:
            df.to_csv(filepath, index=False)
            logger.info(f"Data exported to CSV file: {filepath}")
            return filepath
        
        # Return CSV string if no filepath provided
        csv_buffer = StringIO()
        df.to_csv(csv_buffer, index=False)
        return csv_buffer.getvalue()
    
    except Exception as e:
        logger.error(f"Error exporting data to CSV: {e}")
        raise

def get_download_link(content: str, filename: str, mimetype: str) -> str:
    """
    Generate a download link for the content.
    
    Args:
        content: String content to download
        filename: Name for the downloaded file
        mimetype: MIME type of the file
        
    Returns:
        HTML download link
    """
    try:
        b64 = base64.b64encode(content.encode()).decode()
        href = f'data:{mimetype};base64,{b64}'
        return f'<a href="{href}" download="{filename}" class="download-button">Download {filename}</a>'
    
    except Exception as e:
        logger.error(f"Error creating download link: {e}")
        return ""

def generate_emotion_summary(data: Dict[str, Any]) -> Dict[str, Any]:
    """
    Generate a summary of emotion data across multiple analyses.
    
    Args:
        data: Dictionary of analysis data
        
    Returns:
        Dictionary with summarized emotion data
    """
    try:
        # Initialize counters
        emotion_counts = {}
        total_faces = 0
        emotion_confidence = {}
        
        # Process each analysis
        for analysis in data.get('analyses', []):
            for face in analysis.get('faces', []):
                # Count primary emotions
                emotion = face.get('emotion')
                if emotion:
                    emotion_counts[emotion] = emotion_counts.get(emotion, 0) + 1
                    
                    # Track confidence values
                    if emotion not in emotion_confidence:
                        emotion_confidence[emotion] = []
                    emotion_confidence[emotion].append(face.get('confidence', 0))
                    
                    total_faces += 1
        
        # Calculate statistics
        if total_faces > 0:
            # Calculate percentages
            emotion_percentages = {
                emotion: (count / total_faces) * 100 
                for emotion, count in emotion_counts.items()
            }
            
            # Calculate average confidence per emotion
            avg_confidence = {
                emotion: sum(values) / len(values) 
                for emotion, values in emotion_confidence.items() if values
            }
            
            return {
                'total_analyses': len(data.get('analyses', [])),
                'total_faces': total_faces,
                'emotion_counts': emotion_counts,
                'emotion_percentages': emotion_percentages,
                'avg_confidence': avg_confidence
            }
        
        return {
            'total_analyses': len(data.get('analyses', [])),
            'total_faces': 0,
            'emotion_counts': {},
            'emotion_percentages': {},
            'avg_confidence': {}
        }
    
    except Exception as e:
        logger.error(f"Error generating emotion summary: {e}")
        return {
            'error': str(e),
            'total_analyses': 0,
            'total_faces': 0
        }