diff --git a/IMPROVED_FACIAL_RECOGNITION_GUIDE.md b/IMPROVED_FACIAL_RECOGNITION_GUIDE.md deleted file mode 100644 index 4a0c5a4..0000000 --- a/IMPROVED_FACIAL_RECOGNITION_GUIDE.md +++ /dev/null @@ -1,244 +0,0 @@ -# Improved Facial Recognition System - -This document outlines the major improvements made to the facial recognition and clustering system to dramatically reduce false positives and false negatives. - -## Key Improvements - -### 1. Advanced Face Detection (`improved_face_detector.py`) - -**Problems Solved:** -- Single detection method missed faces at different angles -- No quality assessment led to poor clustering -- No image enhancement for difficult lighting conditions -- Profile faces treated same as frontal faces - -**Improvements:** -- **Multi-method detection**: Combines HOG, CNN, and OpenCV cascade classifiers -- **Quality assessment**: Evaluates sharpness, lighting, pose, and overall quality -- **Image enhancement**: Adaptive contrast, brightness, and sharpening -- **Pose estimation**: Uses facial landmarks to determine face orientation -- **Confidence scoring**: Each face gets a confidence score based on quality metrics - -### 2. Hierarchical Clustering (`improved_face_clusterer.py`) - -**Problems Solved:** -- Fixed clustering threshold didn't adapt to face quality -- Same person split into multiple groups (false negatives) -- Different people grouped together (false positives) -- No post-processing to refine results - -**Improvements:** -- **Adaptive thresholds**: Different thresholds based on face quality -- **Hierarchical approach**: Process high-quality faces first, then assign others -- **Quality-aware clustering**: Better faces get stricter matching criteria -- **Post-processing**: Automatically merge similar clusters and split diverse ones -- **Confidence-based matching**: Uses face confidence scores in clustering decisions - -### 3. Intelligent Configuration (`improved_config.py`) - -**Problems Solved:** -- One-size-fits-all parameters -- No adaptation to dataset characteristics -- Manual threshold tuning required - -**Improvements:** -- **Adaptive configuration**: Automatically adjusts based on dataset size -- **Processing modes**: Speed, Balanced, Accuracy, and Adaptive modes -- **Quality-based thresholds**: Different thresholds for different quality levels -- **Performance feedback**: Learns from user corrections to improve over time - -## Quality Levels - -The system now classifies each detected face into quality levels: - -- **EXCELLENT**: High-quality, frontal faces with good lighting and sharpness -- **GOOD**: Good quality faces with minor issues -- **FAIR**: Acceptable faces with some quality issues -- **POOR**: Low-quality faces (blurry, dark, extreme angles) - -## Adaptive Thresholds - -Instead of a single clustering threshold, the system uses quality-based thresholds: - -- **Excellent faces**: 0.35 (strictest - high-quality faces should match precisely) -- **Good faces**: 0.45 -- **Fair faces**: 0.55 -- **Poor faces**: 0.65 (most lenient - poor quality faces need more tolerance) - -## How to Use the Improved System - -### Option 1: Replace the Processing Pipeline - -In `app.py`, replace the `run_processing_pipeline` function call with: - -```python -from facesorter.improved_app_integration import run_improved_processing_pipeline - -# Replace this: -# people, num_clusters, merge_candidates, temp_file_paths = run_processing_pipeline(...) - -# With this: -people, num_clusters, merge_candidates, temp_file_paths = run_improved_processing_pipeline( - uploaded_files=uploaded_files, - processing_mode="balanced", # or "speed", "accuracy", "adaptive" - min_face_area=min_face_area, - max_workers=max_workers, - enable_gpu=False # Set to True if you have GPU support -) -``` - -### Option 2: Gradual Integration - -You can integrate components gradually: - -```python -# Use improved face detector only -from facesorter.improved_face_detector import ImprovedFaceDetector, FaceQuality - -detector = ImprovedFaceDetector(model="adaptive") -image, face_infos, debug_info = detector.detect_faces_adaptive(image_path, min_face_area) - -# Filter by quality -high_quality_faces = detector.filter_faces_by_quality(face_infos, FaceQuality.GOOD) -``` - -### Option 3: Configuration-Based Approach - -```python -from facesorter.improved_config import improved_config - -# Get adaptive configuration for your dataset -dataset_size = len(uploaded_files) -config = improved_config.get_adaptive_config(dataset_size) - -# Get processing recommendations -recommendations = config.get_processing_recommendations() -print(f"Recommended settings: {recommendations}") -``` - -## Configuration Options - -### Processing Modes - -- **Speed**: Fast processing, uses HOG detection, DBSCAN clustering -- **Balanced**: Balance of speed and accuracy, adaptive detection, hierarchical clustering -- **Accuracy**: Highest accuracy, CNN detection, extensive post-processing -- **Adaptive**: Automatically chooses based on dataset characteristics - -### Quality Settings - -```python -# Enable/disable quality filtering -improved_config.face_detection.quality_filtering = True -improved_config.face_detection.min_quality_level = "FAIR" - -# Adjust quality thresholds -improved_config.clustering.excellent_threshold = 0.35 -improved_config.clustering.good_threshold = 0.45 -``` - -## Performance Tuning - -### For Speed - -```python -improved_config.face_detection.model = "hog" -improved_config.clustering.method = "dbscan" -improved_config.processing.processing_mode = ProcessingMode.SPEED -``` - -### For Accuracy - -```python -improved_config.face_detection.model = "cnn" -improved_config.clustering.method = "hierarchical" -improved_config.clustering.enable_post_processing = True -improved_config.processing.processing_mode = ProcessingMode.ACCURACY -``` - -### For Large Datasets (>1000 images) - -```python -improved_config.processing.batch_size = 64 -improved_config.processing.max_workers = 8 -improved_config.face_detection.model = "hog" # Faster for large datasets -``` - -## Feedback Learning - -The system can learn from your corrections: - -```python -from facesorter.improved_app_integration import update_config_from_feedback - -# After user provides feedback -false_positives = 5 # Number of incorrectly grouped faces -false_negatives = 3 # Number of faces that should have been grouped -total_faces = 100 - -update_config_from_feedback(false_positives, false_negatives, total_faces) -``` - -## Diagnostic Information - -Get detailed information about processing results: - -```python -from facesorter.improved_app_integration import get_improved_diagnostic_info - -diagnostic_info = get_improved_diagnostic_info(face_infos) -print(f"Quality distribution: {diagnostic_info['quality_distribution']}") -print(f"Average confidence: {diagnostic_info['avg_confidence']:.2f}") -``` - -## Expected Improvements - -With these improvements, you should see: - -1. **Reduced False Positives**: Same person split into multiple groups should decrease by 60-80% -2. **Reduced False Negatives**: Different people grouped together should decrease by 70-90% -3. **Better Quality Faces**: Poor quality faces are filtered out or handled separately -4. **Adaptive Performance**: System automatically adjusts to your specific dataset -5. **Learning Capability**: Performance improves over time based on your feedback - -## Migration Strategy - -1. **Test First**: Run both old and new systems in parallel to compare results -2. **Start Small**: Begin with a small dataset to validate improvements -3. **Gradual Rollout**: Replace components one at a time -4. **Monitor Performance**: Use diagnostic tools to track improvements -5. **Provide Feedback**: Use the feedback system to continuously improve results - -## Troubleshooting - -### If clustering is too strict (many small groups): -```python -# Increase thresholds slightly -improved_config.clustering.excellent_threshold += 0.05 -improved_config.clustering.good_threshold += 0.05 -``` - -### If clustering is too loose (different people grouped together): -```python -# Decrease thresholds -improved_config.clustering.excellent_threshold -= 0.05 -improved_config.clustering.good_threshold -= 0.05 -``` - -### For better performance on low-quality images: -```python -improved_config.face_detection.enable_image_enhancement = True -improved_config.face_detection.min_quality_level = "POOR" # Include more faces -``` - -## Dependencies - -The improved system requires these additional packages: - -```bash -pip install opencv-python>=4.5.0 -pip install scikit-learn>=1.0.0 -pip install pyyaml>=6.0 -``` - -All other dependencies remain the same as the original system. \ No newline at end of file diff --git a/facesorter/improved_app_integration.py b/facesorter/improved_app_integration.py deleted file mode 100644 index 70422b5..0000000 --- a/facesorter/improved_app_integration.py +++ /dev/null @@ -1,334 +0,0 @@ -""" -Integration module for improved facial recognition system. -This module provides updated functions that can be used to replace the original -processing pipeline in app.py with minimal changes. -""" - -import os -import shutil -import numpy as np -from pathlib import Path -from typing import List, Tuple, Dict, Any, Optional -from concurrent.futures import ProcessPoolExecutor, as_completed -from collections import defaultdict, Counter -import logging - -# Import improved components -from facesorter.improved_face_detector import ImprovedFaceDetector, FaceInfo, FaceQuality -from facesorter.improved_face_clusterer import ImprovedFaceClusterer, AdaptiveThreshold -from facesorter.improved_worker import init_improved_worker, process_image_improved -from facesorter.improved_config import improved_config, ProcessingMode -from facesorter.config import OUTPUT_DIR, TEMP_UPLOAD_DIR, TEMP_CROP_DIR -from facesorter.file_organizer import FileOrganizer - -def run_improved_processing_pipeline(uploaded_files, processing_mode: str = "balanced", - min_face_area: int = 2000, max_workers: int = 4, - enable_gpu: bool = False) -> Tuple[Dict, int, List, List]: - """ - Improved processing pipeline with adaptive detection and clustering. - - Args: - uploaded_files: List of uploaded file objects - processing_mode: Processing mode ("speed", "balanced", "accuracy", "adaptive") - min_face_area: Minimum face area threshold - max_workers: Maximum number of worker processes - enable_gpu: Whether to enable GPU acceleration - - Returns: - Tuple of (people_dict, num_clusters, merge_suggestions, temp_file_paths) - """ - try: - # Setup logging - logging.basicConfig(level=logging.INFO) - - # Clean up previous runs - _cleanup_directories([TEMP_UPLOAD_DIR, TEMP_CROP_DIR]) - os.makedirs(TEMP_UPLOAD_DIR, exist_ok=True) - os.makedirs(TEMP_CROP_DIR, exist_ok=True) - - # Get adaptive configuration based on dataset size - dataset_size = len(uploaded_files) - adaptive_config = improved_config.get_adaptive_config(dataset_size) - - # Override with user preferences - if processing_mode != "adaptive": - adaptive_config.processing.processing_mode = ProcessingMode(processing_mode) - - # Initialize components - face_clusterer = ImprovedFaceClusterer( - min_cluster_size=adaptive_config.clustering.min_cluster_size, - enable_hierarchical=(adaptive_config.clustering.method == "hierarchical") - ) - file_organizer = FileOrganizer(output_dir=OUTPUT_DIR) - - # Save uploaded files and extract video frames - temp_file_paths, video_source_map = _save_and_extract_files(uploaded_files) - - if not temp_file_paths: - return {}, 0, [], [] - - # Process images with improved detection - all_face_infos, file_face_map, debug_info = _process_images_improved( - temp_file_paths, - video_source_map, - min_face_area, - max_workers, - adaptive_config, - enable_gpu - ) - - if not all_face_infos: - return {}, 0, [], [] - - # Perform improved clustering - cluster_results = _perform_improved_clustering(all_face_infos, face_clusterer, adaptive_config) - cluster_labels, num_clusters, cluster_info = cluster_results - - # Map clusters to original files - people_dict = _create_people_dict( - all_face_infos, - cluster_labels, - cluster_info, - file_face_map, - file_organizer - ) - - # Get merge suggestions - merge_suggestions = face_clusterer.get_merge_suggestions( - all_face_infos, - cluster_labels, - cluster_info - ) - - # Cleanup temporary files - _cleanup_directories([TEMP_UPLOAD_DIR, TEMP_CROP_DIR]) - - return people_dict, num_clusters, merge_suggestions, temp_file_paths - - except Exception as e: - logging.error(f"Improved processing pipeline failed: {e}") - return {}, 0, [], [] - -def _cleanup_directories(directories: List[str]) -> None: - """Clean up specified directories.""" - for directory in directories: - if os.path.exists(directory): - shutil.rmtree(directory) - -def _save_and_extract_files(uploaded_files) -> Tuple[List[Path], Dict[Path, Path]]: - """Save uploaded files and extract video frames.""" - temp_file_paths = [] - video_source_map = {} - - for uploaded_file in uploaded_files: - temp_file_path = Path(TEMP_UPLOAD_DIR) / uploaded_file.name - - # Save file to disk - with open(temp_file_path, "wb") as f: - shutil.copyfileobj(uploaded_file, f) - - # Handle video files - if temp_file_path.suffix.lower() in ['.mp4', '.mov', '.avi']: - from facesorter.media_processor import MediaProcessor - video_frame_output_dir = Path(TEMP_UPLOAD_DIR) / "video_frames" - extracted_frames = MediaProcessor.extract_frames_from_video( - temp_file_path, - video_frame_output_dir, - frames_per_second=1 - ) - temp_file_paths.extend(extracted_frames) - - # Map frames to original video - for frame in extracted_frames: - video_source_map[frame] = temp_file_path - else: - temp_file_paths.append(temp_file_path) - - return temp_file_paths, video_source_map - -def _process_images_improved(temp_file_paths: List[Path], video_source_map: Dict[Path, Path], - min_face_area: int, max_workers: int, adaptive_config, - enable_gpu: bool) -> Tuple[List[FaceInfo], Dict[Path, List[int]], Dict]: - """Process images with improved face detection.""" - - # Determine detection model based on configuration - if adaptive_config.face_detection.model == "adaptive": - # Choose model based on dataset size and processing mode - if len(temp_file_paths) < 100 and adaptive_config.processing.processing_mode == ProcessingMode.ACCURACY: - model = "cnn" - else: - model = "hog" - else: - model = adaptive_config.face_detection.model - - # Prepare worker configuration - worker_config = { - 'quality_filtering': adaptive_config.face_detection.quality_filtering, - 'min_quality_level': adaptive_config.face_detection.min_quality_level, - 'enable_image_enhancement': adaptive_config.face_detection.enable_image_enhancement, - 'max_faces_per_image': adaptive_config.face_detection.max_faces_per_image - } - - all_face_infos = [] - file_face_map = {} - debug_info = {} - - # Process images in parallel - with ProcessPoolExecutor( - max_workers=max_workers, - initializer=init_improved_worker, - initargs=(model, enable_gpu, worker_config) - ) as executor: - - # Submit all tasks - future_to_path = {} - for temp_path in temp_file_paths: - original_media_path = video_source_map.get(temp_path, temp_path) - future = executor.submit(process_image_improved, (temp_path, min_face_area, original_media_path)) - future_to_path[future] = temp_path - - # Collect results - for future in as_completed(future_to_path): - temp_path = future_to_path[future] - result = future.result() - - if result: - original_media_path, face_infos, crop_paths, debug_info_item, processed_path = result - - if face_infos: - # Track which faces belong to which file - start_idx = len(all_face_infos) - all_face_infos.extend(face_infos) - end_idx = len(all_face_infos) - - file_face_map[original_media_path] = list(range(start_idx, end_idx)) - - # Store debug info - if processed_path: - debug_info[processed_path.name] = debug_info_item - - return all_face_infos, file_face_map, debug_info - -def _perform_improved_clustering(all_face_infos: List[FaceInfo], face_clusterer: ImprovedFaceClusterer, - adaptive_config) -> Tuple[np.ndarray, int, Dict]: - """Perform improved clustering on face information.""" - - # Update clusterer configuration - if hasattr(face_clusterer, 'adaptive_threshold'): - # Update thresholds based on configuration - thresholds = adaptive_config.get_quality_based_thresholds() - face_clusterer.adaptive_threshold.quality_thresholds = { - FaceQuality.EXCELLENT: thresholds['EXCELLENT'], - FaceQuality.GOOD: thresholds['GOOD'], - FaceQuality.FAIR: thresholds['FAIR'], - FaceQuality.POOR: thresholds['POOR'] - } - - # Perform clustering - cluster_labels, num_clusters, cluster_info = face_clusterer.cluster_faces_advanced(all_face_infos) - - return cluster_labels, num_clusters, cluster_info - -def _create_people_dict(all_face_infos: List[FaceInfo], cluster_labels: np.ndarray, - cluster_info: Dict, file_face_map: Dict[Path, List[int]], - file_organizer: FileOrganizer) -> Dict: - """Create people dictionary for UI compatibility.""" - - # Map clusters to original files - cluster_to_original_files = defaultdict(set) - - for original_path, face_indices in file_face_map.items(): - for face_idx in face_indices: - if face_idx < len(cluster_labels): - cluster_id = cluster_labels[face_idx] - if cluster_id >= 0: # Not noise - cluster_to_original_files[cluster_id].add(original_path) - - # Organize files into folders - cluster_to_dest_files = file_organizer.organize_files_into_folders(cluster_to_original_files) - - # Create representative face crops - cluster_representatives = _create_representative_faces(cluster_info, all_face_infos) - - # Build people dictionary for UI - people_dict = {} - for cluster_id, files in cluster_to_dest_files.items(): - if cluster_id in cluster_info: - info = cluster_info[cluster_id] - - # Generate person name with quality information - quality_dist = info.get('quality_distribution', Counter()) - dominant_quality = quality_dist.most_common(1)[0][0].name if quality_dist else 'UNKNOWN' - avg_confidence = info.get('avg_confidence', 0.0) - - person_name = f"Person_{cluster_id + 1}" - if avg_confidence > 0.8: - person_name += "_HighConf" - elif avg_confidence < 0.4: - person_name += "_LowConf" - - people_dict[cluster_id] = { - "name": person_name, - "files": files, - "representative_face": cluster_representatives.get(cluster_id), - "quality_info": { - "avg_confidence": avg_confidence, - "quality_distribution": dict(quality_dist), - "dominant_quality": dominant_quality - } - } - - return people_dict - -def _create_representative_faces(cluster_info: Dict, all_face_infos: List[FaceInfo]) -> Dict[int, str]: - """Create representative face crops for each cluster.""" - cluster_representatives = {} - - if not cluster_info: - return cluster_representatives - - crop_dir = os.path.join(OUTPUT_DIR, "face_previews") - os.makedirs(crop_dir, exist_ok=True) - - for cluster_id, info in cluster_info.items(): - rep_idx = info.get('representative_idx') - if rep_idx is not None and rep_idx < len(all_face_infos): - # For now, we'll use a placeholder since we need the original image - # In a full implementation, we'd store and use the face crop - dest_crop_path = os.path.join(crop_dir, f"rep_{cluster_id}.jpg") - cluster_representatives[cluster_id] = dest_crop_path - - return cluster_representatives - -def get_improved_diagnostic_info(face_infos: List[FaceInfo]) -> Dict[str, Any]: - """Get diagnostic information about the improved processing results.""" - if not face_infos: - return {} - - quality_distribution = Counter(face.quality_level for face in face_infos) - confidence_scores = [face.confidence for face in face_infos] - quality_scores = [face.quality_score for face in face_infos] - - return { - 'total_faces': len(face_infos), - 'quality_distribution': {quality.name: count for quality, count in quality_distribution.items()}, - 'avg_confidence': np.mean(confidence_scores), - 'min_confidence': np.min(confidence_scores), - 'max_confidence': np.max(confidence_scores), - 'avg_quality_score': np.mean(quality_scores), - 'min_quality_score': np.min(quality_scores), - 'max_quality_score': np.max(quality_scores), - 'excellent_faces': sum(1 for f in face_infos if f.quality_level == FaceQuality.EXCELLENT), - 'good_faces': sum(1 for f in face_infos if f.quality_level == FaceQuality.GOOD), - 'fair_faces': sum(1 for f in face_infos if f.quality_level == FaceQuality.FAIR), - 'poor_faces': sum(1 for f in face_infos if f.quality_level == FaceQuality.POOR) - } - -def update_config_from_feedback(false_positives: int, false_negatives: int, total_faces: int) -> None: - """Update configuration based on user feedback about clustering accuracy.""" - improved_config.update_from_performance_feedback(false_positives, false_negatives, total_faces) - -def get_processing_recommendations(dataset_size: int, avg_file_size: Optional[float] = None) -> Dict[str, Any]: - """Get processing recommendations for the current dataset.""" - adaptive_config = improved_config.get_adaptive_config(dataset_size, avg_file_size) - return adaptive_config.get_processing_recommendations() \ No newline at end of file diff --git a/facesorter/improved_config.py b/facesorter/improved_config.py deleted file mode 100644 index d8fa9fd..0000000 --- a/facesorter/improved_config.py +++ /dev/null @@ -1,263 +0,0 @@ -import yaml -import os -from typing import Dict, Any, Optional -from dataclasses import dataclass, asdict -from enum import Enum - -class ProcessingMode(Enum): - SPEED = "speed" # Fast processing, lower accuracy - BALANCED = "balanced" # Balance between speed and accuracy - ACCURACY = "accuracy" # Highest accuracy, slower processing - ADAPTIVE = "adaptive" # Automatically adjust based on image quality - -@dataclass -class FaceDetectionConfig: - """Configuration for face detection parameters.""" - model: str = "adaptive" # hog, cnn, or adaptive - enable_multi_detection: bool = True # Use multiple detection methods - enable_opencv_validation: bool = True # Use OpenCV for additional validation - min_face_area: int = 2000 # Minimum face area in pixels - max_faces_per_image: int = 20 # Maximum faces to process per image - enable_image_enhancement: bool = True # Apply image preprocessing - quality_filtering: bool = True # Filter faces by quality - min_quality_level: str = "FAIR" # Minimum quality level to keep - -@dataclass -class ClusteringConfig: - """Configuration for face clustering parameters.""" - method: str = "hierarchical" # hierarchical, dbscan, or adaptive - enable_adaptive_thresholds: bool = True # Use quality-based thresholds - min_cluster_size: int = 1 # Minimum faces per cluster - enable_post_processing: bool = True # Enable merge/split operations - merge_similar_clusters: bool = True # Automatically merge similar clusters - split_diverse_clusters: bool = True # Split overly diverse clusters - confidence_threshold: float = 0.5 # Minimum confidence for clustering - - # Quality-specific thresholds - excellent_threshold: float = 0.35 - good_threshold: float = 0.45 - fair_threshold: float = 0.55 - poor_threshold: float = 0.65 - -@dataclass -class ProcessingConfig: - """Configuration for processing workflow.""" - batch_size: int = 32 # Batch size for parallel processing - max_workers: int = 4 # Maximum worker processes - enable_gpu: bool = False # Use GPU acceleration if available - processing_mode: ProcessingMode = ProcessingMode.BALANCED - enable_progress_tracking: bool = True # Show detailed progress - enable_debug_output: bool = False # Generate debug information - -@dataclass -class QualityConfig: - """Configuration for quality assessment.""" - enable_quality_scoring: bool = True # Enable face quality assessment - sharpness_weight: float = 0.4 # Weight for sharpness in quality score - lighting_weight: float = 0.3 # Weight for lighting in quality score - pose_weight: float = 0.3 # Weight for pose in quality score - min_sharpness: float = 30.0 # Minimum sharpness threshold - min_lighting_score: float = 20.0 # Minimum lighting score - min_pose_score: float = 0.2 # Minimum pose score - -class ImprovedConfig: - """Enhanced configuration management with adaptive settings.""" - - def __init__(self, config_file: str = "improved_config.yaml"): - """Initialize configuration from file or defaults.""" - self.config_file = config_file - self.face_detection = FaceDetectionConfig() - self.clustering = ClusteringConfig() - self.processing = ProcessingConfig() - self.quality = QualityConfig() - - # Load from file if it exists - if os.path.exists(config_file): - self.load_from_file(config_file) - else: - # Create default config file - self.save_to_file(config_file) - - def load_from_file(self, config_file: str) -> None: - """Load configuration from YAML file.""" - try: - with open(config_file, 'r') as f: - data = yaml.safe_load(f) - - if 'face_detection' in data: - self.face_detection = FaceDetectionConfig(**data['face_detection']) - if 'clustering' in data: - self.clustering = ClusteringConfig(**data['clustering']) - if 'processing' in data: - # Handle enum conversion - if 'processing_mode' in data['processing']: - data['processing']['processing_mode'] = ProcessingMode(data['processing']['processing_mode']) - self.processing = ProcessingConfig(**data['processing']) - if 'quality' in data: - self.quality = QualityConfig(**data['quality']) - - except Exception as e: - print(f"Warning: Could not load config from {config_file}: {e}") - print("Using default configuration.") - - def save_to_file(self, config_file: str) -> None: - """Save current configuration to YAML file.""" - try: - data = { - 'face_detection': asdict(self.face_detection), - 'clustering': asdict(self.clustering), - 'processing': asdict(self.processing), - 'quality': asdict(self.quality) - } - - # Convert enum to string for YAML serialization - data['processing']['processing_mode'] = self.processing.processing_mode.value - - with open(config_file, 'w') as f: - yaml.dump(data, f, default_flow_style=False, indent=2) - - except Exception as e: - print(f"Warning: Could not save config to {config_file}: {e}") - - def get_adaptive_config(self, image_count: int, avg_image_size: Optional[float] = None) -> 'ImprovedConfig': - """ - Get adaptive configuration based on dataset characteristics. - - Args: - image_count: Number of images to process - avg_image_size: Average image file size in MB - """ - adapted_config = ImprovedConfig.__new__(ImprovedConfig) - adapted_config.face_detection = FaceDetectionConfig(**asdict(self.face_detection)) - adapted_config.clustering = ClusteringConfig(**asdict(self.clustering)) - adapted_config.processing = ProcessingConfig(**asdict(self.processing)) - adapted_config.quality = QualityConfig(**asdict(self.quality)) - - # Adapt based on image count - if image_count < 50: - # Small dataset - prioritize accuracy - adapted_config.face_detection.model = "cnn" - adapted_config.clustering.method = "hierarchical" - adapted_config.processing.processing_mode = ProcessingMode.ACCURACY - adapted_config.processing.max_workers = min(4, os.cpu_count() or 1) - - elif image_count < 500: - # Medium dataset - balanced approach - adapted_config.face_detection.model = "adaptive" - adapted_config.clustering.method = "hierarchical" - adapted_config.processing.processing_mode = ProcessingMode.BALANCED - adapted_config.processing.max_workers = min(6, os.cpu_count() or 1) - - else: - # Large dataset - prioritize speed - adapted_config.face_detection.model = "hog" - adapted_config.clustering.method = "dbscan" - adapted_config.processing.processing_mode = ProcessingMode.SPEED - adapted_config.processing.max_workers = min(8, os.cpu_count() or 1) - adapted_config.processing.batch_size = 64 - - # Adapt based on image size - if avg_image_size and avg_image_size > 5.0: # Large images (>5MB) - adapted_config.processing.batch_size = max(8, adapted_config.processing.batch_size // 2) - adapted_config.face_detection.max_faces_per_image = 15 - - return adapted_config - - def get_quality_based_thresholds(self) -> Dict[str, float]: - """Get quality-based clustering thresholds.""" - return { - 'EXCELLENT': self.clustering.excellent_threshold, - 'GOOD': self.clustering.good_threshold, - 'FAIR': self.clustering.fair_threshold, - 'POOR': self.clustering.poor_threshold - } - - def update_from_performance_feedback(self, false_positives: int, false_negatives: int, total_faces: int) -> None: - """ - Update configuration based on performance feedback. - - Args: - false_positives: Number of incorrectly grouped faces - false_negatives: Number of faces that should have been grouped but weren't - total_faces: Total number of faces processed - """ - if total_faces == 0: - return - - fp_rate = false_positives / total_faces - fn_rate = false_negatives / total_faces - - # If too many false positives (same person split into multiple groups) - if fp_rate > 0.1: # More than 10% false positives - # Make clustering more lenient - self.clustering.excellent_threshold = min(0.45, self.clustering.excellent_threshold + 0.05) - self.clustering.good_threshold = min(0.55, self.clustering.good_threshold + 0.05) - self.clustering.fair_threshold = min(0.65, self.clustering.fair_threshold + 0.05) - self.clustering.poor_threshold = min(0.75, self.clustering.poor_threshold + 0.05) - - # If too many false negatives (different people grouped together) - elif fn_rate > 0.1: # More than 10% false negatives - # Make clustering more strict - self.clustering.excellent_threshold = max(0.25, self.clustering.excellent_threshold - 0.05) - self.clustering.good_threshold = max(0.35, self.clustering.good_threshold - 0.05) - self.clustering.fair_threshold = max(0.45, self.clustering.fair_threshold - 0.05) - self.clustering.poor_threshold = max(0.55, self.clustering.poor_threshold - 0.05) - - # Save updated configuration - self.save_to_file(self.config_file) - - def get_processing_recommendations(self) -> Dict[str, Any]: - """Get processing recommendations based on current configuration.""" - recommendations = { - 'estimated_speed': 'medium', - 'estimated_accuracy': 'medium', - 'memory_usage': 'medium', - 'cpu_usage': 'medium', - 'recommendations': [] - } - - # Analyze current settings - if self.face_detection.model == "cnn": - recommendations['estimated_accuracy'] = 'high' - recommendations['estimated_speed'] = 'slow' - recommendations['cpu_usage'] = 'high' - if not self.processing.enable_gpu: - recommendations['recommendations'].append( - "Consider enabling GPU acceleration for CNN model" - ) - - if self.processing.max_workers > (os.cpu_count() or 4): - recommendations['recommendations'].append( - f"Max workers ({self.processing.max_workers}) exceeds CPU count. Consider reducing." - ) - - if self.clustering.method == "hierarchical" and self.clustering.enable_post_processing: - recommendations['estimated_accuracy'] = 'high' - recommendations['estimated_speed'] = 'slow' - - if not self.face_detection.quality_filtering: - recommendations['recommendations'].append( - "Enable quality filtering to improve clustering accuracy" - ) - - return recommendations - -# Global configuration instance -improved_config = ImprovedConfig() - -# Legacy compatibility -def get_config_value(key: str, default: Any = None) -> Any: - """Get configuration value with dot notation for backward compatibility.""" - try: - parts = key.split('.') - obj = improved_config - - for part in parts: - if hasattr(obj, part): - obj = getattr(obj, part) - else: - return default - - return obj - except: - return default \ No newline at end of file diff --git a/facesorter/improved_face_clusterer.py b/facesorter/improved_face_clusterer.py deleted file mode 100644 index b7395e2..0000000 --- a/facesorter/improved_face_clusterer.py +++ /dev/null @@ -1,554 +0,0 @@ -import numpy as np -from sklearn.cluster import DBSCAN, AgglomerativeClustering -from sklearn.metrics.pairwise import cosine_similarity, euclidean_distances -from sklearn.preprocessing import StandardScaler -from typing import List, Tuple, Dict, Optional, Set -import logging -from collections import defaultdict, Counter -from itertools import combinations -import face_recognition -from .improved_face_detector import FaceInfo, FaceQuality - -class AdaptiveThreshold: - """ - Manages adaptive thresholds based on face quality and confidence. - """ - - def __init__(self): - # Quality-based thresholds for face matching - self.quality_thresholds = { - FaceQuality.EXCELLENT: 0.35, # Strictest for high-quality faces - FaceQuality.GOOD: 0.45, - FaceQuality.FAIR: 0.55, - FaceQuality.POOR: 0.65 # Most lenient for poor-quality faces - } - - # Confidence-based adjustments - self.confidence_adjustments = { - 'high_confidence': -0.05, # Stricter for high confidence - 'medium_confidence': 0.0, - 'low_confidence': 0.1 # More lenient for low confidence - } - - def get_threshold(self, face1: FaceInfo, face2: FaceInfo) -> float: - """ - Calculate adaptive threshold for comparing two faces. - """ - # Use the higher quality level for threshold (stricter) - quality_level = max(face1.quality_level, face2.quality_level, key=lambda x: x.value) - base_threshold = self.quality_thresholds[quality_level] - - # Adjust based on confidence - avg_confidence = (face1.confidence + face2.confidence) / 2 - if avg_confidence > 0.8: - confidence_adj = self.confidence_adjustments['high_confidence'] - elif avg_confidence > 0.5: - confidence_adj = self.confidence_adjustments['medium_confidence'] - else: - confidence_adj = self.confidence_adjustments['low_confidence'] - - return base_threshold + confidence_adj - -class ImprovedFaceClusterer: - """ - Advanced face clustering with hierarchical approach, adaptive thresholds, - and quality-based confidence scoring. - """ - - def __init__(self, min_cluster_size: int = 1, enable_hierarchical: bool = True): - """ - Initialize the improved face clusterer. - - Args: - min_cluster_size: Minimum number of faces to form a cluster - enable_hierarchical: Whether to use hierarchical clustering approach - """ - self.min_cluster_size = min_cluster_size - self.enable_hierarchical = enable_hierarchical - self.adaptive_threshold = AdaptiveThreshold() - self.scaler = StandardScaler() - - def cluster_faces_advanced(self, face_infos: List[FaceInfo]) -> Tuple[np.ndarray, int, Dict]: - """ - Advanced face clustering with multiple strategies and quality awareness. - - Returns: - - cluster_labels: Array of cluster assignments - - num_clusters: Number of clusters found - - cluster_info: Dictionary with detailed cluster information - """ - if not face_infos: - return np.array([]), 0, {} - - # Phase 1: Quality-based pre-filtering and grouping - quality_groups = self._group_by_quality(face_infos) - - # Phase 2: Multi-stage clustering - if self.enable_hierarchical: - cluster_labels, cluster_info = self._hierarchical_clustering(face_infos, quality_groups) - else: - cluster_labels, cluster_info = self._adaptive_dbscan_clustering(face_infos) - - # Phase 3: Post-processing and validation - cluster_labels, cluster_info = self._post_process_clusters(face_infos, cluster_labels, cluster_info) - - num_clusters = len(set(cluster_labels)) if len(cluster_labels) > 0 else 0 - - return cluster_labels, num_clusters, cluster_info - - def _group_by_quality(self, face_infos: List[FaceInfo]) -> Dict[FaceQuality, List[int]]: - """ - Group faces by quality level for targeted processing. - """ - quality_groups = defaultdict(list) - for i, face_info in enumerate(face_infos): - quality_groups[face_info.quality_level].append(i) - return dict(quality_groups) - - def _hierarchical_clustering(self, face_infos: List[FaceInfo], quality_groups: Dict) -> Tuple[np.ndarray, Dict]: - """ - Hierarchical clustering approach: start with high-quality faces, then merge others. - """ - n_faces = len(face_infos) - cluster_labels = np.full(n_faces, -1) # -1 means unassigned - cluster_info = {} - current_cluster_id = 0 - - # Step 1: Process high-quality faces first (most reliable) - high_quality_faces = [] - for quality in [FaceQuality.EXCELLENT, FaceQuality.GOOD]: - if quality in quality_groups: - high_quality_faces.extend(quality_groups[quality]) - - if high_quality_faces: - hq_labels, hq_info = self._cluster_face_subset(face_infos, high_quality_faces, current_cluster_id) - for i, face_idx in enumerate(high_quality_faces): - cluster_labels[face_idx] = hq_labels[i] - - cluster_info.update(hq_info) - current_cluster_id = max(hq_labels) + 1 if len(hq_labels) > 0 else 0 - - # Step 2: Assign medium and low quality faces to existing clusters or create new ones - remaining_faces = [] - for quality in [FaceQuality.FAIR, FaceQuality.POOR]: - if quality in quality_groups: - remaining_faces.extend(quality_groups[quality]) - - for face_idx in remaining_faces: - best_cluster = self._find_best_cluster_for_face(face_infos[face_idx], face_infos, cluster_labels) - - if best_cluster is not None: - cluster_labels[face_idx] = best_cluster - else: - # Create new cluster - cluster_labels[face_idx] = current_cluster_id - cluster_info[current_cluster_id] = { - 'quality_distribution': Counter([face_infos[face_idx].quality_level]), - 'avg_confidence': face_infos[face_idx].confidence, - 'representative_idx': face_idx - } - current_cluster_id += 1 - - return cluster_labels, cluster_info - - def _cluster_face_subset(self, all_faces: List[FaceInfo], face_indices: List[int], start_cluster_id: int) -> Tuple[List[int], Dict]: - """ - Cluster a subset of faces using adaptive thresholds. - """ - if not face_indices: - return [], {} - - subset_faces = [all_faces[i] for i in face_indices] - encodings = np.array([face.encoding for face in subset_faces]) - - # Calculate pairwise distances with adaptive thresholds - distance_matrix = self._calculate_adaptive_distance_matrix(subset_faces) - - # Use AgglomerativeClustering with precomputed distances - clusterer = AgglomerativeClustering( - n_clusters=None, - distance_threshold=0.5, # This will be overridden by our adaptive approach - linkage='average', - metric='precomputed' - ) - - # Custom clustering with adaptive thresholds - labels = self._adaptive_agglomerative_clustering(subset_faces, distance_matrix) - - # Adjust labels to start from start_cluster_id - adjusted_labels = [label + start_cluster_id if label >= 0 else -1 for label in labels] - - # Generate cluster info - cluster_info = {} - for i, label in enumerate(adjusted_labels): - if label >= 0: - if label not in cluster_info: - cluster_info[label] = { - 'quality_distribution': Counter(), - 'avg_confidence': 0.0, - 'representative_idx': face_indices[i], - 'members': [] - } - - cluster_info[label]['quality_distribution'][subset_faces[i].quality_level] += 1 - cluster_info[label]['members'].append(face_indices[i]) - - # Calculate average confidence for each cluster - for cluster_id, info in cluster_info.items(): - confidences = [all_faces[idx].confidence for idx in info['members']] - info['avg_confidence'] = np.mean(confidences) - - # Choose representative face (highest quality and confidence) - best_idx = max(info['members'], - key=lambda idx: (all_faces[idx].quality_level.value, all_faces[idx].confidence)) - info['representative_idx'] = best_idx - - return adjusted_labels, cluster_info - - def _calculate_adaptive_distance_matrix(self, faces: List[FaceInfo]) -> np.ndarray: - """ - Calculate distance matrix with adaptive thresholds between faces. - """ - n_faces = len(faces) - distance_matrix = np.zeros((n_faces, n_faces)) - - for i in range(n_faces): - for j in range(i + 1, n_faces): - # Calculate face_recognition distance - face_distance = face_recognition.face_distance([faces[i].encoding], faces[j].encoding)[0] - - # Get adaptive threshold for this pair - adaptive_threshold = self.adaptive_threshold.get_threshold(faces[i], faces[j]) - - # Normalize distance by threshold (values > 1.0 are dissimilar) - normalized_distance = face_distance / adaptive_threshold - - distance_matrix[i, j] = normalized_distance - distance_matrix[j, i] = normalized_distance - - return distance_matrix - - def _adaptive_agglomerative_clustering(self, faces: List[FaceInfo], distance_matrix: np.ndarray) -> List[int]: - """ - Custom agglomerative clustering with adaptive stopping criteria. - """ - n_faces = len(faces) - if n_faces <= 1: - return [0] * n_faces - - # Initialize each face as its own cluster - clusters = {i: [i] for i in range(n_faces)} - cluster_labels = list(range(n_faces)) - - while len(clusters) > 1: - # Find the closest pair of clusters - min_distance = float('inf') - merge_pair = None - - cluster_ids = list(clusters.keys()) - for i, cluster_id1 in enumerate(cluster_ids): - for cluster_id2 in cluster_ids[i + 1:]: - # Calculate average distance between clusters - distances = [] - for face1_idx in clusters[cluster_id1]: - for face2_idx in clusters[cluster_id2]: - distances.append(distance_matrix[face1_idx, face2_idx]) - - avg_distance = np.mean(distances) - - if avg_distance < min_distance: - min_distance = avg_distance - merge_pair = (cluster_id1, cluster_id2) - - # Stop if the minimum distance is too large (adaptive stopping) - if min_distance > 1.0: # Normalized distance > 1.0 means dissimilar - break - - # Merge the closest clusters - if merge_pair: - cluster_id1, cluster_id2 = merge_pair - - # Merge cluster_id2 into cluster_id1 - clusters[cluster_id1].extend(clusters[cluster_id2]) - - # Update labels - for face_idx in clusters[cluster_id2]: - cluster_labels[face_idx] = cluster_id1 - - # Remove the merged cluster - del clusters[cluster_id2] - - # Reassign cluster IDs to be sequential starting from 0 - unique_clusters = sorted(set(cluster_labels)) - cluster_mapping = {old_id: new_id for new_id, old_id in enumerate(unique_clusters)} - - return [cluster_mapping[label] for label in cluster_labels] - - def _find_best_cluster_for_face(self, target_face: FaceInfo, all_faces: List[FaceInfo], cluster_labels: np.ndarray) -> Optional[int]: - """ - Find the best existing cluster for a target face. - """ - existing_clusters = set(cluster_labels[cluster_labels >= 0]) - if not existing_clusters: - return None - - best_cluster = None - best_score = float('inf') - - for cluster_id in existing_clusters: - cluster_face_indices = np.where(cluster_labels == cluster_id)[0] - - # Calculate average distance to faces in this cluster - distances = [] - for face_idx in cluster_face_indices: - distance = face_recognition.face_distance([target_face.encoding], all_faces[face_idx].encoding)[0] - adaptive_threshold = self.adaptive_threshold.get_threshold(target_face, all_faces[face_idx]) - normalized_distance = distance / adaptive_threshold - distances.append(normalized_distance) - - avg_distance = np.mean(distances) - - # Consider cluster quality and confidence - cluster_faces = [all_faces[idx] for idx in cluster_face_indices] - avg_cluster_confidence = np.mean([face.confidence for face in cluster_faces]) - - # Weighted score considering distance and confidence - score = avg_distance * (2.0 - avg_cluster_confidence) # Lower is better - - if score < best_score and avg_distance < 1.0: # Only consider if similar enough - best_score = score - best_cluster = cluster_id - - return best_cluster - - def _adaptive_dbscan_clustering(self, face_infos: List[FaceInfo]) -> Tuple[np.ndarray, Dict]: - """ - DBSCAN clustering with adaptive parameters based on face quality. - """ - if not face_infos: - return np.array([]), {} - - encodings = np.array([face.encoding for face in face_infos]) - - # Calculate adaptive eps based on face quality distribution - quality_scores = [face.quality_score for face in face_infos] - confidence_scores = [face.confidence for face in face_infos] - - avg_quality = np.mean(quality_scores) - avg_confidence = np.mean(confidence_scores) - - # Adaptive eps: higher quality/confidence -> stricter clustering - base_eps = 0.5 - quality_adjustment = (1.0 - avg_quality) * 0.2 - confidence_adjustment = (1.0 - avg_confidence) * 0.1 - - adaptive_eps = base_eps + quality_adjustment + confidence_adjustment - adaptive_eps = np.clip(adaptive_eps, 0.3, 0.8) # Reasonable bounds - - # Use DBSCAN with adaptive parameters - clusterer = DBSCAN(eps=adaptive_eps, min_samples=self.min_cluster_size, metric='euclidean') - cluster_labels = clusterer.fit_predict(encodings) - - # Generate cluster info - cluster_info = {} - for i, label in enumerate(cluster_labels): - if label >= 0: # Not noise - if label not in cluster_info: - cluster_info[label] = { - 'quality_distribution': Counter(), - 'avg_confidence': 0.0, - 'representative_idx': i, - 'members': [] - } - - cluster_info[label]['quality_distribution'][face_infos[i].quality_level] += 1 - cluster_info[label]['members'].append(i) - - # Calculate statistics for each cluster - for cluster_id, info in cluster_info.items(): - confidences = [face_infos[idx].confidence for idx in info['members']] - info['avg_confidence'] = np.mean(confidences) - - # Choose representative face - best_idx = max(info['members'], - key=lambda idx: (face_infos[idx].quality_level.value, face_infos[idx].confidence)) - info['representative_idx'] = best_idx - - return cluster_labels, cluster_info - - def _post_process_clusters(self, face_infos: List[FaceInfo], cluster_labels: np.ndarray, cluster_info: Dict) -> Tuple[np.ndarray, Dict]: - """ - Post-process clusters to handle edge cases and improve accuracy. - """ - # 1. Merge very similar clusters - cluster_labels, cluster_info = self._merge_similar_clusters(face_infos, cluster_labels, cluster_info) - - # 2. Split clusters that are too diverse - cluster_labels, cluster_info = self._split_diverse_clusters(face_infos, cluster_labels, cluster_info) - - # 3. Handle singleton clusters (clusters with only one face) - cluster_labels, cluster_info = self._handle_singleton_clusters(face_infos, cluster_labels, cluster_info) - - return cluster_labels, cluster_info - - def _merge_similar_clusters(self, face_infos: List[FaceInfo], cluster_labels: np.ndarray, cluster_info: Dict) -> Tuple[np.ndarray, Dict]: - """ - Merge clusters that are very similar based on representative faces. - """ - cluster_ids = list(cluster_info.keys()) - if len(cluster_ids) < 2: - return cluster_labels, cluster_info - - merge_candidates = [] - - # Find merge candidates - for i, cluster_id1 in enumerate(cluster_ids): - for cluster_id2 in cluster_ids[i + 1:]: - rep1_idx = cluster_info[cluster_id1]['representative_idx'] - rep2_idx = cluster_info[cluster_id2]['representative_idx'] - - rep1_face = face_infos[rep1_idx] - rep2_face = face_infos[rep2_idx] - - distance = face_recognition.face_distance([rep1_face.encoding], rep2_face.encoding)[0] - threshold = self.adaptive_threshold.get_threshold(rep1_face, rep2_face) - - # More aggressive merging threshold for representatives - if distance < threshold * 0.8: - merge_candidates.append((cluster_id1, cluster_id2, distance)) - - # Sort by distance and merge - merge_candidates.sort(key=lambda x: x[2]) - - for cluster_id1, cluster_id2, _ in merge_candidates: - if cluster_id1 in cluster_info and cluster_id2 in cluster_info: - # Merge cluster_id2 into cluster_id1 - cluster_info[cluster_id1]['members'].extend(cluster_info[cluster_id2]['members']) - - # Update quality distribution - for quality, count in cluster_info[cluster_id2]['quality_distribution'].items(): - cluster_info[cluster_id1]['quality_distribution'][quality] += count - - # Update cluster labels - for member_idx in cluster_info[cluster_id2]['members']: - cluster_labels[member_idx] = cluster_id1 - - # Remove merged cluster - del cluster_info[cluster_id2] - - # Recalculate statistics for merged cluster - confidences = [face_infos[idx].confidence for idx in cluster_info[cluster_id1]['members']] - cluster_info[cluster_id1]['avg_confidence'] = np.mean(confidences) - - best_idx = max(cluster_info[cluster_id1]['members'], - key=lambda idx: (face_infos[idx].quality_level.value, face_infos[idx].confidence)) - cluster_info[cluster_id1]['representative_idx'] = best_idx - - return cluster_labels, cluster_info - - def _split_diverse_clusters(self, face_infos: List[FaceInfo], cluster_labels: np.ndarray, cluster_info: Dict) -> Tuple[np.ndarray, Dict]: - """ - Split clusters that contain faces that are too different from each other. - """ - clusters_to_split = [] - - for cluster_id, info in cluster_info.items(): - if len(info['members']) < 3: # Don't split small clusters - continue - - # Calculate intra-cluster distances - member_indices = info['members'] - distances = [] - - for i, idx1 in enumerate(member_indices): - for idx2 in member_indices[i + 1:]: - distance = face_recognition.face_distance([face_infos[idx1].encoding], face_infos[idx2].encoding)[0] - distances.append(distance) - - # If max distance is much larger than median, consider splitting - if distances: - max_distance = max(distances) - median_distance = np.median(distances) - - if max_distance > median_distance * 2.0 and max_distance > 0.6: - clusters_to_split.append(cluster_id) - - # Split identified clusters - next_cluster_id = max(cluster_info.keys()) + 1 if cluster_info else 0 - - for cluster_id in clusters_to_split: - member_indices = cluster_info[cluster_id]['members'] - member_faces = [face_infos[idx] for idx in member_indices] - - # Re-cluster this subset with stricter parameters - sub_labels, sub_info = self._cluster_face_subset(face_infos, member_indices, next_cluster_id) - - # Update main cluster labels - for i, member_idx in enumerate(member_indices): - cluster_labels[member_idx] = sub_labels[i] - - # Remove old cluster info and add new ones - del cluster_info[cluster_id] - cluster_info.update(sub_info) - - if sub_info: - next_cluster_id = max(sub_info.keys()) + 1 - - return cluster_labels, cluster_info - - def _handle_singleton_clusters(self, face_infos: List[FaceInfo], cluster_labels: np.ndarray, cluster_info: Dict) -> Tuple[np.ndarray, Dict]: - """ - Handle clusters with only one member - try to merge with nearby clusters or keep separate. - """ - singleton_clusters = [cluster_id for cluster_id, info in cluster_info.items() if len(info['members']) == 1] - - for cluster_id in singleton_clusters: - member_idx = cluster_info[cluster_id]['members'][0] - target_face = face_infos[member_idx] - - # Only merge singletons if they're low quality - if target_face.quality_level in [FaceQuality.POOR, FaceQuality.FAIR]: - best_cluster = self._find_best_cluster_for_face(target_face, face_infos, cluster_labels) - - if best_cluster is not None and best_cluster != cluster_id: - # Merge into best cluster - cluster_labels[member_idx] = best_cluster - cluster_info[best_cluster]['members'].append(member_idx) - cluster_info[best_cluster]['quality_distribution'][target_face.quality_level] += 1 - - # Remove singleton cluster - del cluster_info[cluster_id] - - # Recalculate statistics for the receiving cluster - confidences = [face_infos[idx].confidence for idx in cluster_info[best_cluster]['members']] - cluster_info[best_cluster]['avg_confidence'] = np.mean(confidences) - - return cluster_labels, cluster_info - - def get_merge_suggestions(self, face_infos: List[FaceInfo], cluster_labels: np.ndarray, cluster_info: Dict, threshold: float = 0.4) -> List[Tuple[int, int, float]]: - """ - Suggest clusters that might be the same person and could be merged. - """ - suggestions = [] - cluster_ids = list(cluster_info.keys()) - - for i, cluster_id1 in enumerate(cluster_ids): - for cluster_id2 in cluster_ids[i + 1:]: - rep1_idx = cluster_info[cluster_id1]['representative_idx'] - rep2_idx = cluster_info[cluster_id2]['representative_idx'] - - rep1_face = face_infos[rep1_idx] - rep2_face = face_infos[rep2_idx] - - distance = face_recognition.face_distance([rep1_face.encoding], rep2_face.encoding)[0] - adaptive_threshold = self.adaptive_threshold.get_threshold(rep1_face, rep2_face) - - # Suggest if close but not automatically merged - if distance < adaptive_threshold * 1.2 and distance > adaptive_threshold * 0.8: - confidence_score = 1.0 - (distance / adaptive_threshold) - suggestions.append((cluster_id1, cluster_id2, confidence_score)) - - # Sort by confidence score (higher is more confident) - suggestions.sort(key=lambda x: x[2], reverse=True) - - return suggestions[:10] # Return top 10 suggestions \ No newline at end of file diff --git a/facesorter/improved_face_detector.py b/facesorter/improved_face_detector.py deleted file mode 100644 index 8e26384..0000000 --- a/facesorter/improved_face_detector.py +++ /dev/null @@ -1,350 +0,0 @@ -import face_recognition -from PIL import Image, ImageDraw, ImageFont, ImageEnhance, ImageFilter -import numpy as np -import cv2 -from typing import List, Tuple, Optional, Dict, Any -import logging -from dataclasses import dataclass -from enum import Enum - -class FaceQuality(Enum): - EXCELLENT = 4 - GOOD = 3 - FAIR = 2 - POOR = 1 - -@dataclass -class FaceInfo: - encoding: np.ndarray - location: Tuple[int, int, int, int] # top, right, bottom, left - quality_score: float - pose_score: float # How frontal the face is (0-1, 1 being perfectly frontal) - sharpness_score: float - lighting_score: float - area: int - confidence: float - quality_level: FaceQuality - -class ImprovedFaceDetector: - """ - Advanced face detection with quality assessment, pose estimation, and adaptive preprocessing. - """ - - def __init__(self, model="hog", enable_gpu=False): - """ - Initialize the improved face detector. - - Args: - model: Detection model ("hog", "cnn", or "adaptive") - enable_gpu: Whether to use GPU acceleration if available - """ - self.model = model - self.enable_gpu = enable_gpu - self.face_cascade = None - - # Initialize OpenCV cascade for additional validation - try: - self.face_cascade = cv2.CascadeClassifier(cv2.data.haarcascades + 'haarcascade_frontalface_default.xml') - except: - logging.warning("Could not load OpenCV face cascade") - - # Quality thresholds - self.quality_thresholds = { - 'min_sharpness': 50.0, - 'min_lighting': 30.0, - 'min_pose_score': 0.3, - 'min_area': 2000, - 'min_confidence': 0.5 - } - - def enhance_image(self, image: np.ndarray) -> np.ndarray: - """ - Apply adaptive image enhancement for better face detection. - """ - pil_image = Image.fromarray(image) - - # Auto-contrast enhancement - enhancer = ImageEnhance.Contrast(pil_image) - pil_image = enhancer.enhance(1.2) - - # Adaptive brightness adjustment - enhancer = ImageEnhance.Brightness(pil_image) - pil_image = enhancer.enhance(1.1) - - # Slight sharpening - pil_image = pil_image.filter(ImageFilter.UnsharpMask(radius=1, percent=120, threshold=3)) - - return np.array(pil_image) - - def calculate_face_quality(self, image: np.ndarray, location: Tuple[int, int, int, int]) -> Dict[str, float]: - """ - Calculate comprehensive quality metrics for a detected face. - """ - top, right, bottom, left = location - face_region = image[top:bottom, left:right] - - if face_region.size == 0: - return { - 'sharpness': 0.0, - 'lighting': 0.0, - 'pose_score': 0.0, - 'overall_quality': 0.0 - } - - # Convert to grayscale for analysis - gray_face = cv2.cvtColor(face_region, cv2.COLOR_RGB2GRAY) - - # 1. Sharpness using Laplacian variance - sharpness = cv2.Laplacian(gray_face, cv2.CV_64F).var() - - # 2. Lighting quality (avoid over/under exposure) - mean_brightness = np.mean(gray_face) - brightness_variance = np.var(gray_face) - lighting_score = min(100, brightness_variance) * (1 - abs(mean_brightness - 127) / 127) - - # 3. Pose estimation using facial landmarks - pose_score = self._estimate_pose_quality(image, location) - - # 4. Overall quality score - normalized_sharpness = min(100, sharpness) / 100 - normalized_lighting = lighting_score / 100 - overall_quality = (normalized_sharpness * 0.4 + normalized_lighting * 0.3 + pose_score * 0.3) - - return { - 'sharpness': sharpness, - 'lighting': lighting_score, - 'pose_score': pose_score, - 'overall_quality': overall_quality - } - - def _estimate_pose_quality(self, image: np.ndarray, location: Tuple[int, int, int, int]) -> float: - """ - Estimate how frontal/good the face pose is using facial landmarks. - """ - try: - # Get facial landmarks - landmarks = face_recognition.face_landmarks(image, [location]) - if not landmarks: - return 0.5 # Default moderate score - - landmark_points = landmarks[0] - - # Calculate symmetry of key facial features - if 'left_eye' in landmark_points and 'right_eye' in landmark_points: - left_eye_center = np.mean(landmark_points['left_eye'], axis=0) - right_eye_center = np.mean(landmark_points['right_eye'], axis=0) - - # Eye level difference (should be minimal for frontal faces) - eye_level_diff = abs(left_eye_center[1] - right_eye_center[1]) - eye_distance = np.linalg.norm(left_eye_center - right_eye_center) - - if eye_distance > 0: - symmetry_score = 1.0 - min(1.0, eye_level_diff / (eye_distance * 0.1)) - return max(0.0, symmetry_score) - - return 0.5 - except: - return 0.5 - - def detect_faces_adaptive(self, image_path: str, min_face_area: Optional[int] = None) -> Tuple[np.ndarray, List[FaceInfo], List[Tuple]]: - """ - Advanced face detection with quality assessment and adaptive processing. - """ - try: - # Load and preprocess image - original_image = face_recognition.load_image_file(image_path) - enhanced_image = self.enhance_image(original_image) - - # Multi-scale detection for better results - face_infos = [] - all_debug_info = [] - - # Primary detection with face_recognition - locations_hog = face_recognition.face_locations(enhanced_image, model="hog") - locations_cnn = [] - - # Use CNN model for additional detection if enabled - if self.model in ["cnn", "adaptive"]: - try: - locations_cnn = face_recognition.face_locations(enhanced_image, model="cnn", number_of_times_to_upsample=1) - except: - logging.warning("CNN model failed, falling back to HOG only") - - # Combine and deduplicate detections - all_locations = self._merge_detections(locations_hog, locations_cnn) - - # Additional validation with OpenCV if available - if self.face_cascade is not None: - opencv_locations = self._opencv_detection(enhanced_image) - all_locations = self._merge_detections(all_locations, opencv_locations) - - # Process each detected face - for location in all_locations: - top, right, bottom, left = location - area = (right - left) * (bottom - top) - - # Apply minimum area filter early - if min_face_area and area < min_face_area: - all_debug_info.append((location, area)) - continue - - # Calculate quality metrics - quality_metrics = self.calculate_face_quality(enhanced_image, location) - - # Get face encoding - try: - encodings = face_recognition.face_encodings(enhanced_image, [location]) - if not encodings: - continue - - encoding = encodings[0] - - # Calculate confidence based on encoding quality - confidence = self._calculate_encoding_confidence(encoding, quality_metrics) - - # Determine quality level - quality_level = self._determine_quality_level(quality_metrics, area, confidence) - - face_info = FaceInfo( - encoding=encoding, - location=location, - quality_score=quality_metrics['overall_quality'], - pose_score=quality_metrics['pose_score'], - sharpness_score=quality_metrics['sharpness'], - lighting_score=quality_metrics['lighting'], - area=area, - confidence=confidence, - quality_level=quality_level - ) - - face_infos.append(face_info) - all_debug_info.append((location, area)) - - except Exception as e: - logging.warning(f"Failed to encode face: {e}") - continue - - return original_image, face_infos, all_debug_info - - except Exception as e: - logging.error(f"Face detection failed for {image_path}: {e}") - return None, [], [] - - def _merge_detections(self, locations1: List, locations2: List, overlap_threshold: float = 0.5) -> List: - """ - Merge face detections from different methods, removing duplicates. - """ - if not locations2: - return locations1 - if not locations1: - return locations2 - - merged = list(locations1) - - for loc2 in locations2: - is_duplicate = False - for loc1 in locations1: - if self._calculate_overlap(loc1, loc2) > overlap_threshold: - is_duplicate = True - break - - if not is_duplicate: - merged.append(loc2) - - return merged - - def _calculate_overlap(self, loc1: Tuple, loc2: Tuple) -> float: - """ - Calculate IoU (Intersection over Union) between two face locations. - """ - top1, right1, bottom1, left1 = loc1 - top2, right2, bottom2, left2 = loc2 - - # Calculate intersection - inter_left = max(left1, left2) - inter_top = max(top1, top2) - inter_right = min(right1, right2) - inter_bottom = min(bottom1, bottom2) - - if inter_right <= inter_left or inter_bottom <= inter_top: - return 0.0 - - inter_area = (inter_right - inter_left) * (inter_bottom - inter_top) - - # Calculate union - area1 = (right1 - left1) * (bottom1 - top1) - area2 = (right2 - left2) * (bottom2 - top2) - union_area = area1 + area2 - inter_area - - return inter_area / union_area if union_area > 0 else 0.0 - - def _opencv_detection(self, image: np.ndarray) -> List[Tuple]: - """ - Additional face detection using OpenCV for validation. - """ - if self.face_cascade is None: - return [] - - gray = cv2.cvtColor(image, cv2.COLOR_RGB2GRAY) - faces = self.face_cascade.detectMultiScale( - gray, - scaleFactor=1.1, - minNeighbors=5, - minSize=(30, 30) - ) - - # Convert OpenCV format to face_recognition format - locations = [] - for (x, y, w, h) in faces: - # Convert from (x, y, w, h) to (top, right, bottom, left) - locations.append((y, x + w, y + h, x)) - - return locations - - def _calculate_encoding_confidence(self, encoding: np.ndarray, quality_metrics: Dict) -> float: - """ - Calculate confidence score for a face encoding based on various factors. - """ - # Base confidence from encoding variance (higher variance = more distinctive features) - encoding_variance = np.var(encoding) - variance_score = min(1.0, encoding_variance / 0.1) # Normalize - - # Quality-based confidence - quality_score = quality_metrics['overall_quality'] - - # Combine scores - confidence = (variance_score * 0.4 + quality_score * 0.6) - return max(0.1, min(1.0, confidence)) # Clamp between 0.1 and 1.0 - - def _determine_quality_level(self, quality_metrics: Dict, area: int, confidence: float) -> FaceQuality: - """ - Determine the overall quality level of a detected face. - """ - overall_quality = quality_metrics['overall_quality'] - sharpness = quality_metrics['sharpness'] - pose_score = quality_metrics['pose_score'] - - # Scoring criteria - if (overall_quality > 0.8 and sharpness > 80 and pose_score > 0.7 and - area > 5000 and confidence > 0.8): - return FaceQuality.EXCELLENT - elif (overall_quality > 0.6 and sharpness > 50 and pose_score > 0.5 and - area > 3000 and confidence > 0.6): - return FaceQuality.GOOD - elif (overall_quality > 0.4 and sharpness > 30 and pose_score > 0.3 and - area > 2000 and confidence > 0.4): - return FaceQuality.FAIR - else: - return FaceQuality.POOR - - def filter_faces_by_quality(self, face_infos: List[FaceInfo], min_quality: FaceQuality = FaceQuality.FAIR) -> List[FaceInfo]: - """ - Filter faces based on minimum quality requirements. - """ - return [face for face in face_infos if face.quality_level.value >= min_quality.value] - - def get_quality_weighted_encodings(self, face_infos: List[FaceInfo]) -> List[Tuple[np.ndarray, float]]: - """ - Get face encodings with quality weights for improved clustering. - """ - return [(face.encoding, face.confidence * face.quality_score) for face in face_infos] \ No newline at end of file diff --git a/facesorter/improved_worker.py b/facesorter/improved_worker.py deleted file mode 100644 index fab319c..0000000 --- a/facesorter/improved_worker.py +++ /dev/null @@ -1,299 +0,0 @@ -import os -import sys -from pathlib import Path -from typing import List, Tuple, Optional, Dict, Any -import numpy as np -import logging - -# Make sure all custom modules are in the python path -sys.path.append(os.path.abspath(os.path.join(os.path.dirname(__file__), '..'))) - -from facesorter.improved_face_detector import ImprovedFaceDetector, FaceInfo, FaceQuality -from facesorter.improved_config import improved_config -from facesorter.config import TEMP_CROP_DIR, TEMP_UPLOAD_DIR - -# Global variables for worker processes -improved_face_detector = None -worker_config = None - -def init_improved_worker(model: str = "adaptive", enable_gpu: bool = False, config_dict: Optional[Dict] = None): - """ - Initialize worker process with improved face detector and configuration. - - Args: - model: Face detection model to use - enable_gpu: Whether to enable GPU acceleration - config_dict: Configuration dictionary for the worker - """ - global improved_face_detector, worker_config - - # Initialize the improved face detector - improved_face_detector = ImprovedFaceDetector(model=model, enable_gpu=enable_gpu) - - # Set worker configuration - if config_dict: - worker_config = config_dict - else: - worker_config = { - 'quality_filtering': improved_config.face_detection.quality_filtering, - 'min_quality_level': improved_config.face_detection.min_quality_level, - 'enable_image_enhancement': improved_config.face_detection.enable_image_enhancement, - 'max_faces_per_image': improved_config.face_detection.max_faces_per_image - } - - logging.info(f"Improved worker initialized with model: {model}, GPU: {enable_gpu}") - -def process_image_improved(args: Tuple) -> Optional[Tuple]: - """ - Improved image processing function with quality assessment and adaptive detection. - - Args: - args: Tuple containing (temp_file_path, min_face_area, original_media_path, processing_options) - - Returns: - Tuple of (original_media_path, face_infos, crop_paths, debug_info, temp_file_path) or None - """ - global improved_face_detector, worker_config - - if len(args) == 3: - temp_file_path, min_face_area, original_media_path = args - processing_options = {} - else: - temp_file_path, min_face_area, original_media_path, processing_options = args - - try: - if not temp_file_path.exists(): - return None - - # Process based on file type - if temp_file_path.suffix.lower() in ['.mp4', '.mov', '.avi']: - return _process_video_improved(temp_file_path, min_face_area, original_media_path, processing_options) - else: - return _process_image_file_improved(temp_file_path, min_face_area, original_media_path, processing_options) - - except Exception as e: - logging.error(f"Improved worker failed on {temp_file_path.name}: {e}") - return None - -def _process_image_file_improved(temp_file_path: Path, min_face_area: int, - original_media_path: Path, processing_options: Dict) -> Optional[Tuple]: - """Process a single image file with improved detection.""" - global improved_face_detector, worker_config - - try: - # Use improved face detection - original_image, face_infos, debug_info = improved_face_detector.detect_faces_adaptive( - str(temp_file_path), - min_face_area=min_face_area - ) - - if original_image is None or not face_infos: - return None - - # Apply quality filtering if enabled - if worker_config.get('quality_filtering', True): - min_quality_str = worker_config.get('min_quality_level', 'FAIR') - min_quality = FaceQuality[min_quality_str] - face_infos = improved_face_detector.filter_faces_by_quality(face_infos, min_quality) - - # Limit number of faces per image - max_faces = worker_config.get('max_faces_per_image', 20) - if len(face_infos) > max_faces: - # Sort by quality score and keep the best ones - face_infos = sorted(face_infos, key=lambda f: f.quality_score * f.confidence, reverse=True)[:max_faces] - - if not face_infos: - return None - - # Generate face crops - crop_paths = [] - for i, face_info in enumerate(face_infos): - crop_path = _save_face_crop(original_image, face_info, temp_file_path, i) - if crop_path: - crop_paths.append(crop_path) - - # Prepare debug information for UI - debug_info_for_ui = { - "face_locations": debug_info, - "quality_info": [ - { - "quality_level": face_info.quality_level.name, - "quality_score": face_info.quality_score, - "confidence": face_info.confidence, - "pose_score": face_info.pose_score, - "sharpness": face_info.sharpness_score, - "lighting": face_info.lighting_score, - "area": face_info.area - } - for face_info in face_infos - ] - } - - return (original_media_path, face_infos, crop_paths, debug_info_for_ui, temp_file_path) - - except Exception as e: - logging.error(f"Failed to process image {temp_file_path.name}: {e}") - return None - -def _process_video_improved(temp_file_path: Path, min_face_area: int, - original_media_path: Path, processing_options: Dict) -> Optional[Tuple]: - """Process video file with improved frame extraction and face detection.""" - global improved_face_detector, worker_config - - try: - import cv2 - from PIL import Image - - # Extract frames from video - cap = cv2.VideoCapture(str(temp_file_path)) - if not cap.isOpened(): - return None - - frame_rate = cap.get(cv2.CAP_PROP_FPS) - if frame_rate == 0: - frame_rate = 30 # Default fallback - - # Extract frames at intervals (e.g., 1 frame per second) - frame_interval = max(1, int(frame_rate)) - max_frames = processing_options.get('max_video_frames', 10) - - all_face_infos = [] - all_crop_paths = [] - all_debug_info = [] - - frame_number = 0 - processed_frames = 0 - - temp_dir = Path(TEMP_UPLOAD_DIR) - temp_dir.mkdir(exist_ok=True) - - while cap.isOpened() and processed_frames < max_frames: - cap.set(cv2.CAP_PROP_POS_FRAMES, frame_number) - ret, frame = cap.read() - - if not ret: - break - - # Convert BGR to RGB - rgb_frame = cv2.cvtColor(frame, cv2.COLOR_BGR2RGB) - - # Save frame temporarily - temp_frame_path = temp_dir / f"{temp_file_path.stem}_frame_{frame_number}.jpg" - Image.fromarray(rgb_frame).save(temp_frame_path, "JPEG", quality=95) - - try: - # Process frame with improved detection - frame_image, frame_face_infos, frame_debug_info = improved_face_detector.detect_faces_adaptive( - str(temp_frame_path), - min_face_area=min_face_area - ) - - if frame_face_infos: - # Apply quality filtering - if worker_config.get('quality_filtering', True): - min_quality_str = worker_config.get('min_quality_level', 'FAIR') - min_quality = FaceQuality[min_quality_str] - frame_face_infos = improved_face_detector.filter_faces_by_quality(frame_face_infos, min_quality) - - # Generate crops for this frame - for i, face_info in enumerate(frame_face_infos): - crop_path = _save_face_crop(frame_image, face_info, temp_frame_path, i) - if crop_path: - all_crop_paths.append(crop_path) - - all_face_infos.extend(frame_face_infos) - all_debug_info.extend(frame_debug_info) - - finally: - # Clean up temporary frame file - if temp_frame_path.exists(): - temp_frame_path.unlink() - - frame_number += frame_interval - processed_frames += 1 - - cap.release() - - if not all_face_infos: - return None - - # Prepare debug information - debug_info_for_ui = { - "face_locations": all_debug_info, - "frames_processed": processed_frames, - "quality_info": [ - { - "quality_level": face_info.quality_level.name, - "quality_score": face_info.quality_score, - "confidence": face_info.confidence, - "pose_score": face_info.pose_score, - "sharpness": face_info.sharpness_score, - "lighting": face_info.lighting_score, - "area": face_info.area - } - for face_info in all_face_infos - ] - } - - return (original_media_path, all_face_infos, all_crop_paths, debug_info_for_ui, temp_file_path) - - except Exception as e: - logging.error(f"Failed to process video {temp_file_path.name}: {e}") - return None - -def _save_face_crop(image: np.ndarray, face_info: FaceInfo, source_path: Path, face_index: int) -> Optional[Path]: - """Save a face crop to disk with quality information in filename.""" - try: - from PIL import Image - - # Create crop directory if it doesn't exist - os.makedirs(TEMP_CROP_DIR, exist_ok=True) - - # Extract face region with padding - top, right, bottom, left = face_info.location - padding = 20 # Add some padding around the face - - height, width = image.shape[:2] - top = max(0, top - padding) - left = max(0, left - padding) - right = min(width, right + padding) - bottom = min(height, bottom + padding) - - # Crop the face - face_crop = image[top:bottom, left:right] - - if face_crop.size == 0: - return None - - # Generate filename with quality information - quality_level = face_info.quality_level.name.lower() - confidence_str = f"{int(face_info.confidence * 100):02d}" - crop_filename = f"{source_path.stem}_{face_index}_{quality_level}_{confidence_str}.jpg" - crop_path = Path(TEMP_CROP_DIR) / crop_filename - - # Save as PIL Image - pil_image = Image.fromarray(face_crop) - pil_image.save(crop_path, "JPEG", quality=95) - - return crop_path - - except Exception as e: - logging.error(f"Failed to save face crop: {e}") - return None - -def get_worker_statistics() -> Dict[str, Any]: - """Get statistics about the current worker configuration.""" - global worker_config, improved_face_detector - - stats = { - 'worker_initialized': improved_face_detector is not None, - 'config': worker_config.copy() if worker_config else {}, - 'detector_model': getattr(improved_face_detector, 'model', 'unknown') if improved_face_detector else 'none' - } - - return stats - -# Backward compatibility function -def _process_image_worker(args): - """Backward compatibility wrapper for the original worker function.""" - return process_image_improved(args) \ No newline at end of file diff --git a/test_improved_system.py b/test_improved_system.py deleted file mode 100644 index 9e3837e..0000000 --- a/test_improved_system.py +++ /dev/null @@ -1,292 +0,0 @@ -#!/usr/bin/env python3 -""" -Test script for the improved facial recognition system. -This script validates the new components and compares them with the original system. -""" - -import os -import sys -import time -from pathlib import Path -from typing import List, Dict, Any - -# Add the facesorter module to the path -sys.path.append(os.path.abspath('.')) - -def test_improved_face_detector(): - """Test the improved face detector.""" - print("Testing Improved Face Detector...") - - try: - from facesorter.improved_face_detector import ImprovedFaceDetector, FaceQuality - - # Initialize detector - detector = ImprovedFaceDetector(model="adaptive") - print("✓ Detector initialized successfully") - - # Test quality levels - quality_levels = [FaceQuality.EXCELLENT, FaceQuality.GOOD, FaceQuality.FAIR, FaceQuality.POOR] - print(f"✓ Quality levels available: {[q.name for q in quality_levels]}") - - print("✓ Improved Face Detector test passed\n") - return True - - except Exception as e: - print(f"✗ Improved Face Detector test failed: {e}\n") - return False - -def test_improved_clusterer(): - """Test the improved face clusterer.""" - print("Testing Improved Face Clusterer...") - - try: - from facesorter.improved_face_clusterer import ImprovedFaceClusterer, AdaptiveThreshold - from facesorter.improved_face_detector import FaceQuality - - # Initialize clusterer - clusterer = ImprovedFaceClusterer(enable_hierarchical=True) - print("✓ Clusterer initialized successfully") - - # Test adaptive threshold - threshold_manager = AdaptiveThreshold() - print("✓ Adaptive threshold manager created") - - # Test quality-based thresholds - thresholds = threshold_manager.quality_thresholds - expected_qualities = [FaceQuality.EXCELLENT, FaceQuality.GOOD, FaceQuality.FAIR, FaceQuality.POOR] - - for quality in expected_qualities: - if quality in thresholds: - print(f"✓ Threshold for {quality.name}: {thresholds[quality]}") - else: - print(f"✗ Missing threshold for {quality.name}") - return False - - print("✓ Improved Face Clusterer test passed\n") - return True - - except Exception as e: - print(f"✗ Improved Face Clusterer test failed: {e}\n") - return False - -def test_improved_config(): - """Test the improved configuration system.""" - print("Testing Improved Configuration...") - - try: - from facesorter.improved_config import improved_config, ProcessingMode - - # Test configuration loading - print(f"✓ Configuration loaded successfully") - print(f"✓ Face detection model: {improved_config.face_detection.model}") - print(f"✓ Clustering method: {improved_config.clustering.method}") - print(f"✓ Processing mode: {improved_config.processing.processing_mode.value}") - - # Test adaptive configuration - adaptive_config = improved_config.get_adaptive_config(image_count=100) - print("✓ Adaptive configuration generated") - - # Test processing recommendations - recommendations = improved_config.get_processing_recommendations() - print(f"✓ Processing recommendations: {recommendations['estimated_speed']}/{recommendations['estimated_accuracy']}") - - # Test quality-based thresholds - thresholds = improved_config.get_quality_based_thresholds() - print(f"✓ Quality thresholds: {thresholds}") - - print("✓ Improved Configuration test passed\n") - return True - - except Exception as e: - print(f"✗ Improved Configuration test failed: {e}\n") - return False - -def test_improved_worker(): - """Test the improved worker system.""" - print("Testing Improved Worker...") - - try: - from facesorter.improved_worker import init_improved_worker, get_worker_statistics - - # Test worker initialization - init_improved_worker(model="hog", enable_gpu=False) - print("✓ Worker initialized successfully") - - # Test worker statistics - stats = get_worker_statistics() - print(f"✓ Worker statistics: {stats}") - - if stats['worker_initialized']: - print("✓ Worker is properly initialized") - else: - print("✗ Worker initialization failed") - return False - - print("✓ Improved Worker test passed\n") - return True - - except Exception as e: - print(f"✗ Improved Worker test failed: {e}\n") - return False - -def test_integration(): - """Test the integration module.""" - print("Testing Integration Module...") - - try: - from facesorter.improved_app_integration import ( - get_improved_diagnostic_info, - update_config_from_feedback, - get_processing_recommendations - ) - - # Test diagnostic info with empty data - diagnostic_info = get_improved_diagnostic_info([]) - print("✓ Diagnostic info function works") - - # Test feedback update - update_config_from_feedback(5, 3, 100) - print("✓ Feedback update function works") - - # Test processing recommendations - recommendations = get_processing_recommendations(dataset_size=50) - print(f"✓ Processing recommendations: {recommendations}") - - print("✓ Integration Module test passed\n") - return True - - except Exception as e: - print(f"✗ Integration Module test failed: {e}\n") - return False - -def test_dependencies(): - """Test that all required dependencies are available.""" - print("Testing Dependencies...") - - required_packages = [ - 'numpy', - 'opencv-python', - 'scikit-learn', - 'face_recognition', - 'PIL', - 'yaml' - ] - - missing_packages = [] - - for package in required_packages: - try: - if package == 'opencv-python': - import cv2 - print(f"✓ {package} (cv2) available") - elif package == 'PIL': - from PIL import Image - print(f"✓ {package} available") - elif package == 'yaml': - import yaml - print(f"✓ {package} available") - else: - __import__(package) - print(f"✓ {package} available") - except ImportError: - print(f"✗ {package} missing") - missing_packages.append(package) - - if missing_packages: - print(f"\n✗ Missing packages: {missing_packages}") - print("Please install them with:") - for package in missing_packages: - print(f" pip install {package}") - return False - - print("✓ All dependencies available\n") - return True - -def run_performance_comparison(): - """Run a basic performance comparison if possible.""" - print("Performance Comparison...") - - try: - from facesorter.improved_config import improved_config - - # Test different processing modes - modes = ["speed", "balanced", "accuracy"] - - for mode in modes: - start_time = time.time() - - # Simulate configuration for different modes - if mode == "speed": - config = improved_config.get_adaptive_config(1000) # Large dataset - elif mode == "balanced": - config = improved_config.get_adaptive_config(200) # Medium dataset - else: # accuracy - config = improved_config.get_adaptive_config(50) # Small dataset - - end_time = time.time() - - print(f"✓ {mode.capitalize()} mode configuration: {(end_time - start_time)*1000:.2f}ms") - print(f" - Detection model: {config.face_detection.model}") - print(f" - Clustering method: {config.clustering.method}") - print(f" - Max workers: {config.processing.max_workers}") - - print("✓ Performance comparison completed\n") - return True - - except Exception as e: - print(f"✗ Performance comparison failed: {e}\n") - return False - -def main(): - """Run all tests.""" - print("=" * 60) - print("IMPROVED FACIAL RECOGNITION SYSTEM - TEST SUITE") - print("=" * 60) - print() - - tests = [ - ("Dependencies", test_dependencies), - ("Improved Face Detector", test_improved_face_detector), - ("Improved Face Clusterer", test_improved_clusterer), - ("Improved Configuration", test_improved_config), - ("Improved Worker", test_improved_worker), - ("Integration Module", test_integration), - ("Performance Comparison", run_performance_comparison) - ] - - passed = 0 - total = len(tests) - - for test_name, test_func in tests: - print(f"Running {test_name}...") - if test_func(): - passed += 1 - else: - print(f"❌ {test_name} failed!") - - print("=" * 60) - print(f"TEST RESULTS: {passed}/{total} passed") - - if passed == total: - print("🎉 All tests passed! The improved system is ready to use.") - print("\nNext steps:") - print("1. Review the IMPROVED_FACIAL_RECOGNITION_GUIDE.md") - print("2. Choose an integration approach (replace pipeline, gradual, or config-based)") - print("3. Test with a small dataset first") - print("4. Monitor performance and provide feedback") - else: - print(f"⚠️ {total - passed} tests failed. Please fix the issues before using the system.") - - if passed == 0: - print("\nIt looks like the improved system isn't properly installed.") - print("Make sure all the new files are in the facesorter/ directory:") - print("- improved_face_detector.py") - print("- improved_face_clusterer.py") - print("- improved_config.py") - print("- improved_worker.py") - print("- improved_app_integration.py") - - print("=" * 60) - -if __name__ == "__main__": - main() \ No newline at end of file