Photo Content Recognition Curation Expert

Photo Content Recognition & Curation Expert

Expert in photo content analysis and intelligent curation. Combines classical computer vision with modern deep learning for comprehensive photo analysis.

When to Use This Skill

✅ Use for:

• Face recognition and clustering (identifying important people)
• Animal/pet detection and clustering
• Near-duplicate detection using perceptual hashing (DINOHash, pHash, dHash)
• Burst photo selection (finding best frame from 10-50 shots)
• Screenshot vs photo classification
• Meme/download filtering
• NSFW content detection
• Quick indexing for large photo libraries (10K+)
• Aesthetic quality scoring (NIMA)

❌ NOT for:

• GPS-based location clustering → event-detection-temporal-intelligence-expert
• Color palette extraction → color-theory-palette-harmony-expert
• Semantic image-text matching → clip-aware-embeddings
• Video analysis or frame extraction

Quick Decision Tree

What do you need to recognize/filter?
│
├─ Duplicate photos? ─────────────────────────────── Perceptual Hashing
│   ├─ Exact duplicates? ──────────────────────────── dHash (fastest)
│   ├─ Brightness/contrast changes? ───────────────── pHash (DCT-based)
│   ├─ Heavy crops/compression? ───────────────────── DINOHash (2025 SOTA)
│   └─ Production system? ─────────────────────────── Hybrid (pHash → DINOHash)
│
├─ People in photos? ─────────────────────────────── Face Clustering
│   ├─ Known thresholds? ──────────────────────────── Apple-style Agglomerative
│   └─ Unknown data distribution? ─────────────────── HDBSCAN
│
├─ Pets/Animals? ─────────────────────────────────── Pet Recognition
│   ├─ Detection? ─────────────────────────────────── YOLOv8
│   └─ Individual clustering? ─────────────────────── CLIP + HDBSCAN
│
├─ Best from burst? ──────────────────────────────── Burst Selection
│   └─ Score: sharpness + face quality + aesthetics
│
└─ Filter junk? ──────────────────────────────────── Content Detection
    ├─ Screenshots? ───────────────────────────────── Multi-signal classifier
    └─ NSFW? ──────────────────────────────────────── Safety classifier

---

Core Concepts

1. Perceptual Hashing for Near-Duplicate Detection

Problem: Camera bursts, re-saved images, and minor edits create near-duplicates.

Solution: Perceptual hashes generate similar values for visually similar images.

Method Comparison:

Hybrid Pipeline (2025 Best Practice): 1. Stage 1: Fast pHash filtering (eliminates obvious non-duplicates) 2. Stage 2: DINOHash refinement (accurate detection) 3. Stage 3: Optional Siamese ViT verification

Hamming Distance Thresholds:

• Conservative: ≤5 bits different = duplicates
• Aggressive: ≤10 bits different = duplicates

→ Deep dive: references/perceptual-hashing.md

---

2. Face Recognition & Clustering

Goal: Group photos by person without user labeling.

Apple Photos Strategy (2021-2025): 1. Extract face + upper body embeddings (FaceNet, 512-dim) 2. Two-pass agglomerative clustering 3. Conservative first pass (threshold=0.4, high precision) 4. HAC second pass (threshold=0.6, increase recall) 5. Incremental updates for new photos

HDBSCAN Alternative:

• No threshold tuning required
• Robust to noise
• Better for unknown data distributions

Parameters:

→ Deep dive: references/face-clustering.md

---

3. Burst Photo Selection

Problem: Burst mode creates 10-50 nearly identical photos.

Multi-Criteria Scoring:

Burst Detection: Photos within 0.5 seconds of each other.

→ Deep dive: references/content-detection.md

---

4. Screenshot Detection

Multi-Signal Approach:

Decision: Confidence >0.6 = screenshot

→ Deep dive: references/content-detection.md

---

5. Quick Indexing Pipeline

Goal: Index 10K+ photos efficiently with caching.

Features Extracted:

• Perceptual hashes (de-duplication)
• Face embeddings (people clustering)
• CLIP embeddings (semantic search)
• Color palettes
• Aesthetic scores

Performance (10K photos, M1 MacBook Pro):

→ Deep dive: references/photo-indexing.md

---

Common Anti-Patterns

Anti-Pattern: Euclidean Distance for Face Embeddings

What it looks like:

distance = np.linalg.norm(embedding1 - embedding2)  # WRONG

Why it's wrong: Face embeddings are normalized; cosine similarity is the correct metric.

What to do instead:

from scipy.spatial.distance import cosine
distance = cosine(embedding1, embedding2)  # Correct

Anti-Pattern: Fixed Clustering Thresholds

What it looks like: Using same distance threshold for all face clusters.

Why it's wrong: Different people have varying intra-class variance (twins vs. diverse ages).

What to do instead: Use HDBSCAN for automatic threshold discovery, or two-pass clustering with conservative + relaxed passes.

Anti-Pattern: Raw Pixel Comparison for Duplicates

What it looks like:

is_duplicate = np.allclose(img1, img2)  # WRONG

Why it's wrong: Re-saved JPEGs, crops, brightness changes create pixel differences.

What to do instead: Perceptual hashing (pHash or DINOHash) with Hamming distance.

Anti-Pattern: Sequential Face Detection

What it looks like: Processing faces one photo at a time without batching.

Why it's wrong: GPU underutilization, 10x slower than batched.

What to do instead: Batch process images (batch_size=32) with GPU acceleration.

Anti-Pattern: No Confidence Filtering

What it looks like:

for face in all_detected_faces:
    cluster(face)  # No filtering

Why it's wrong: Low-confidence detections create noise clusters (hands, objects).

What to do instead: Filter by confidence (threshold 0.9 for faces).

Anti-Pattern: Forcing Every Photo into Clusters

What it looks like: Assigning noise points to nearest cluster.

Why it's wrong: Solo appearances shouldn't pollute person clusters.

What to do instead: HDBSCAN/DBSCAN naturally identifies noise (label=-1). Keep noise separate.

---

Quick Start

from photo_curation import PhotoCurationPipeline

pipeline = PhotoCurationPipeline()

# Index photo library
index = pipeline.index_library('/path/to/photos')

# De-duplicate
duplicates = index.find_duplicates()
print(f"Found {len(duplicates)} duplicate groups")

# Cluster faces
face_clusters = index.cluster_faces()
print(f"Found {len(face_clusters)} people")

# Select best from bursts
best_photos = pipeline.select_best_from_bursts(index)

# Filter screenshots
real_photos = pipeline.filter_screenshots(index)

# Curate for collage
collage_photos = pipeline.curate_for_collage(index, target_count=100)

---

Python Dependencies

torch transformers facenet-pytorch ultralytics hdbscan opencv-python scipy numpy scikit-learn pillow pytesseract

---

Integration Points

• event-detection-temporal-intelligence-expert: Provides temporal event clustering for event-aware curation
• color-theory-palette-harmony-expert: Extracts color palettes for visual diversity
• collage-layout-expert: Receives curated photos for assembly
• clip-aware-embeddings: Provides CLIP embeddings for semantic search and DeepDBSCAN

---

References

1. DINOHash (2025): "Adversarially Fine-Tuned DINOv2 Features for Perceptual Hashing" 2. Apple Photos (2021): "Recognizing People in Photos Through Private On-Device ML" 3. HDBSCAN: "Hierarchical Density-Based Spatial Clustering" (2013-2025) 4. Perceptual Hashing: dHash (Neal Krawetz), DCT-based pHash

---

Version: 2.0.0 Last Updated: November 2025

Changelog: photo-content-recognition-curation-expert

[2.0.0] - 2025-11-26

Major Refactoring

• Reduced SKILL.md from 1513 lines to 322 lines (79% reduction)
• Extracted detailed implementations to reference files
• Added proper skill-coach compliant structure

Added

• Frontmatter: Updated to allowed-tools format with integration points
• NOT clause: Clear boundaries with sister skills (event-detection, color-theory, clip-aware-embeddings)
• Decision tree: Quick algorithm selection guide for all recognition tasks
• 6 Anti-patterns: Common mistakes specific to photo curation
• Euclidean distance for face embeddings (use cosine)
• Fixed clustering thresholds
• Raw pixel comparison for duplicates
• Sequential face detection (use batching)
• No confidence filtering
• Forcing every photo into clusters
• Quick reference tables: Method comparisons, parameters, performance benchmarks
• Integration points: Links to event-detection, color-theory, collage-layout, clip-aware-embeddings

Reference Files Created

• references/perceptual-hashing.md - Hash algorithms and duplicate detection
• DINOHash (2025 state-of-the-art)
• dHash, pHash implementations
• Hybrid duplicate detection pipeline
• BK-Tree for efficient search (100K+ photos)
• Method comparison table

• references/face-clustering.md - Face recognition and clustering
• FaceEmbeddingExtractor (MTCNN + InceptionResnetV1)
• Apple-style two-pass agglomerative clustering
• HDBSCAN alternative
• Incremental updates for new photos
• Method comparison (Agglomerative vs HDBSCAN)

• references/content-detection.md - Content analysis
• PetRecognizer (YOLO + CLIP)
• BurstPhotoSelector with multi-criteria scoring
• ScreenshotDetector with multi-signal approach
• Scoring weight tables

• references/photo-indexing.md - Indexing pipeline
• QuickPhotoIndexer with caching
• PhotoIndex container class
• Complete curation pipeline
• Performance benchmarks (10K photos)

Performance Targets (Documented)

Dependencies

torch transformers facenet-pytorch ultralytics hdbscan opencv-python scipy numpy scikit-learn pillow pytesseract

[1.0.0] - 2025-11 (Initial)

Initial Implementation

• DINOHash perceptual hashing
• Apple-style face clustering
• HDBSCAN clustering alternative
• Pet/animal recognition
• Burst photo selection
• Screenshot detection
• Quick indexing pipeline
• Complete curation workflow

Content Detection Reference

Pet Recognition & Clustering

from ultralytics import YOLO
from transformers import CLIPProcessor, CLIPModel

class PetRecognizer:
    """Pet detection and clustering."""

    def __init__(self):
        self.yolo = YOLO('yolov8n.pt')
        self.clip_model = CLIPModel.from_pretrained("openai/clip-vit-base-patch32")
        self.clip_processor = CLIPProcessor.from_pretrained("openai/clip-vit-base-patch32")

    def detect_animals(self, image):
        """Detect animals in image."""
        results = self.yolo(image)

        animals = []
        animal_classes = ['cat', 'dog', 'horse', 'bird', 'cow', 'sheep',
                         'elephant', 'bear', 'zebra', 'giraffe']

        for result in results:
            for box in result.boxes:
                class_name = result.names[int(box.cls)]
                if class_name in animal_classes:
                    animals.append({
                        'type': class_name,
                        'bbox': box.xyxy[0].cpu().numpy(),
                        'confidence': float(box.conf)
                    })
        return animals

    def extract_pet_embedding(self, image, bbox):
        """Extract embedding for individual animal using CLIP."""
        x1, y1, x2, y2 = map(int, bbox)
        crop = image.crop((x1, y1, x2, y2))

        inputs = self.clip_processor(images=crop, return_tensors="pt")
        with torch.no_grad():
            embedding = self.clip_model.get_image_features(**inputs)
        return embedding.cpu().numpy().flatten()

    def cluster_pets(self, pet_embeddings, min_cluster_size=5):
        """Cluster pet embeddings (same individual = cluster)."""
        clusterer = hdbscan.HDBSCAN(
            min_cluster_size=min_cluster_size,
            min_samples=2,
            metric='cosine'
        )
        return clusterer.fit_predict(np.array(pet_embeddings))

---

Burst Photo Selection

Problem: Burst mode creates 10-50 nearly identical photos. Need to select best frame.

Solution: Multi-criteria scoring: sharpness, face quality, aesthetics, composition.

class BurstPhotoSelector:
    """Select best photo from camera burst sequence."""

    def __init__(self):
        self.face_detector = FaceEmbeddingExtractor()
        self.aesthetic_scorer = NIMAPredictor()

    def detect_bursts(self, photos_with_timestamps, max_gap_seconds=0.5):
        """Detect burst sequences from timestamps."""
        sorted_photos = sorted(photos_with_timestamps, key=lambda x: x[1])

        bursts = []
        current_burst = [sorted_photos[0]]

        for photo in sorted_photos[1:]:
            time_gap = (photo[1] - current_burst[-1][1]).total_seconds()
            if time_gap <= max_gap_seconds:
                current_burst.append(photo)
            else:
                if len(current_burst) >= 3:
                    bursts.append(current_burst)
                current_burst = [photo]

        if len(current_burst) >= 3:
            bursts.append(current_burst)

        return bursts

    def select_best_from_burst(self, burst_photos):
        """
        Select best photo from burst.
        Criteria: Sharpness, Face quality, Aesthetics, Position, Exposure
        """
        scores = []

        for idx, (photo_id, timestamp, image) in enumerate(burst_photos):
            score = 0.0

            # 1. SHARPNESS (30%)
            gray = cv2.cvtColor(np.array(image), cv2.COLOR_RGB2GRAY)
            sharpness = cv2.Laplacian(gray, cv2.CV_64F).var()
            score += min(1.0, sharpness / 1000) * 0.30

            # 2. FACE QUALITY (35%)
            faces = self.face_detector.extract_faces(image)
            if faces:
                face_scores = [self.assess_face_quality(f) for f in faces]
                score += np.mean(face_scores) * 0.35
            else:
                score += 0.5 * 0.35

            # 3. AESTHETIC SCORE (20%)
            score += self.aesthetic_scorer.predict(image) * 0.20

            # 4. POSITION BONUS - middle frames (10%)
            position = idx / len(burst_photos)
            center_bonus = 1.0 - abs(position - 0.5) * 2
            score += center_bonus * 0.10

            # 5. EXPOSURE (5%)
            score += self.assess_exposure(image) * 0.05

            scores.append((photo_id, score))

        return max(scores, key=lambda x: x[1])[0]

    def assess_face_quality(self, face_dict):
        """Assess quality: eyes open, not blurry, smiling."""
        face_crop = face_dict['crop']
        gray_face = cv2.cvtColor(np.array(face_crop), cv2.COLOR_RGB2GRAY)
        face_sharpness = cv2.Laplacian(gray_face, cv2.CV_64F).var()
        sharpness_score = min(1.0, face_sharpness / 500)

        # Production: Use landmarks or emotion classifier
        eyes_open_score = 0.8
        smiling_score = 0.7

        return (sharpness_score * 0.4 + eyes_open_score * 0.3 + smiling_score * 0.3)

    def assess_exposure(self, image):
        """Check if image is properly exposed."""
        gray = cv2.cvtColor(np.array(image), cv2.COLOR_RGB2GRAY)
        hist = cv2.calcHist([gray], [0], None, [256], [0, 256])
        hist = hist.flatten() / hist.sum()

        clipping = np.sum(hist[:20]) + np.sum(hist[235:])
        return max(0, 1.0 - clipping / 0.05)

---

Screenshot vs Photo Classification

Signals: 1. EXIF metadata (camera info missing) 2. UI elements (status bars, buttons) 3. Text density 4. Perfect rectangles 5. Device-specific aspect ratios 6. Sharpness (screenshots are perfectly sharp)

class ScreenshotDetector:
    """Classify image as screenshot vs photo."""

    def __init__(self):
        self.text_detector = self.init_text_detector()
        self.ui_detector = self.init_ui_detector()

    def is_screenshot(self, image, metadata=None):
        """
        Determine if image is screenshot.
        Returns: (bool, confidence)
        """
        signals = []

        # SIGNAL 1: EXIF metadata
        if metadata:
            has_camera_info = any(k in metadata for k in
                                ['Make', 'Model', 'LensModel', 'FocalLength'])
            if not has_camera_info:
                signals.append(('no_camera_exif', 0.6))
        else:
            signals.append(('no_metadata', 0.5))

        # SIGNAL 2: UI elements
        ui_elements = self.detect_ui_elements(image)
        if ui_elements:
            signals.append(('ui_elements', 0.85))

        # SIGNAL 3: Text density
        text_coverage = self.compute_text_coverage(image)
        if text_coverage > 0.25:
            signals.append(('high_text', 0.7))

        # SIGNAL 4: Perfect rectangles
        perfect_rects = self.detect_perfect_rectangles(image)
        if perfect_rects > 5:
            signals.append(('perfect_rects', 0.75))

        # SIGNAL 5: Device aspect ratio
        h, w = np.array(image).shape[:2]
        aspect = w / h
        device_aspects = [(16/9, 'standard'), (1125/2436, 'iphone_x'),
                         (1080/1920, 'android_fhd'), (1440/2960, 'samsung_s8')]

        for target_aspect, device_name in device_aspects:
            if abs(aspect - target_aspect) < 0.01:
                signals.append((f'device_aspect_{device_name}', 0.6))
                break

        # SIGNAL 6: Perfect sharpness
        sharpness = self.compute_sharpness(image)
        if sharpness > 2000:
            signals.append(('perfect_sharpness', 0.5))

        if not signals:
            return False, 0.0

        max_confidence = max(conf for _, conf in signals)
        return max_confidence > 0.6, max_confidence

    def detect_ui_elements(self, image):
        """Detect status bars, buttons, icons."""
        h, w = np.array(image).shape[:2]
        top_strip = np.array(image)[:int(h * 0.05), :]
        top_variance = np.var(top_strip)

        if top_variance < 100:
            return [{'type': 'status_bar', 'confidence': 0.8}]
        return []

    def compute_text_coverage(self, image):
        """Compute % of image covered by text."""
        import pytesseract
        data = pytesseract.image_to_data(image, output_type=pytesseract.Output.DICT)

        total_area = image.width * image.height
        text_area = sum(
            data['width'][i] * data['height'][i]
            for i, conf in enumerate(data['conf']) if conf > 0
        )
        return text_area / total_area

    def detect_perfect_rectangles(self, image):
        """Detect pixel-perfect rectangles (UI buttons)."""
        gray = cv2.cvtColor(np.array(image), cv2.COLOR_RGB2GRAY)
        edges = cv2.Canny(gray, 50, 150)
        contours, _ = cv2.findContours(edges, cv2.RETR_TREE, cv2.CHAIN_APPROX_SIMPLE)

        perfect_rects = 0
        for contour in contours:
            epsilon = 0.01 * cv2.arcLength(contour, True)
            approx = cv2.approxPolyDP(contour, epsilon, True)
            if len(approx) == 4:
                perfect_rects += 1
        return perfect_rects

    def compute_sharpness(self, image):
        gray = cv2.cvtColor(np.array(image), cv2.COLOR_RGB2GRAY)
        return cv2.Laplacian(gray, cv2.CV_64F).var()

---

Burst Selection Weights

Screenshot Detection Signals

Face Recognition & Clustering Reference

Overview

Group photos by person without user labeling using face detection, embedding extraction, and clustering.

Face Detection & Embedding Extraction

from facenet_pytorch import MTCNN, InceptionResnetV1
import torch

class FaceEmbeddingExtractor:
    """Extract face embeddings using FaceNet (512-dim vectors)."""

    def __init__(self, device='cuda' if torch.cuda.is_available() else 'cpu'):
        self.device = device

        # MTCNN for face detection
        self.mtcnn = MTCNN(
            image_size=160,
            margin=0,
            min_face_size=20,
            device=self.device
        )

        # InceptionResnetV1 for embeddings
        self.resnet = InceptionResnetV1(pretrained='vggface2').eval().to(self.device)

    def extract_faces(self, image):
        """
        Detect and extract face embeddings.
        Returns: List of (face_crop, embedding, bounding_box) tuples
        """
        boxes, probs = self.mtcnn.detect(image)

        if boxes is None:
            return []

        faces = []
        for box, prob in zip(boxes, probs):
            if prob < 0.9:
                continue

            face_crop = self.mtcnn.extract(image, [box], save_path=None)[0]
            face_tensor = face_crop.unsqueeze(0).to(self.device)

            with torch.no_grad():
                embedding = self.resnet(face_tensor).cpu().numpy().flatten()

            faces.append({
                'crop': face_crop,
                'embedding': embedding,
                'bbox': box,
                'confidence': prob
            })

        return faces

---

Apple-Style Two-Pass Agglomerative Clustering

Strategy (Apple Photos 2021-2025): 1. Extract face + upper body embeddings 2. Two-pass agglomerative clustering 3. Conservative first pass (high precision) 4. HAC second pass (increase recall) 5. Incremental updates for new photos

from sklearn.cluster import AgglomerativeClustering
from scipy.spatial.distance import cosine
import numpy as np

class ApplePhotosFaceClustering:
    """
    Two-pass agglomerative clustering inspired by Apple Photos.
    Based on: "Recognizing People in Photos Through Private On-Device ML" (Apple ML, 2021)
    """

    def __init__(self):
        self.distance_threshold_pass1 = 0.4  # Conservative (high precision)
        self.distance_threshold_pass2 = 0.6  # Relaxed (increase recall)

    def cluster_faces(self, face_embeddings, photo_ids):
        """Cluster face embeddings into person clusters."""
        if len(face_embeddings) < 2:
            return {0: list(range(len(face_embeddings)))}

        embeddings = np.array(face_embeddings)

        # PASS 1: Conservative clustering
        clustering_pass1 = AgglomerativeClustering(
            n_clusters=None,
            distance_threshold=self.distance_threshold_pass1,
            linkage='average',
            metric='cosine'
        )
        labels_pass1 = clustering_pass1.fit_predict(embeddings)

        # Compute cluster centroids from pass 1
        unique_labels = np.unique(labels_pass1)
        cluster_centroids = []
        cluster_members = {}

        for label in unique_labels:
            mask = labels_pass1 == label
            cluster_emb = embeddings[mask]
            centroid = np.median(cluster_emb, axis=0)
            cluster_centroids.append(centroid)
            cluster_members[label] = np.where(mask)[0].tolist()

        # PASS 2: Merge similar clusters
        if len(cluster_centroids) > 1:
            clustering_pass2 = AgglomerativeClustering(
                n_clusters=None,
                distance_threshold=self.distance_threshold_pass2,
                linkage='average',
                metric='cosine'
            )
            centroid_labels = clustering_pass2.fit_predict(np.array(cluster_centroids))

            final_clusters = {}
            for old_label, new_label in enumerate(centroid_labels):
                if new_label not in final_clusters:
                    final_clusters[new_label] = []
                final_clusters[new_label].extend(cluster_members[old_label])
        else:
            final_clusters = cluster_members

        return final_clusters

    def incremental_update(self, existing_clusters, new_faces, new_embeddings):
        """Incrementally add new faces to existing clusters."""
        updated_clusters = existing_clusters.copy()
        unassigned_faces = []

        for face_idx, embedding in enumerate(new_embeddings):
            min_distance = float('inf')
            closest_cluster = None

            for cluster_id, face_indices in existing_clusters.items():
                cluster_embeddings = [face_embeddings[i] for i in face_indices]
                cluster_median = np.median(cluster_embeddings, axis=0)
                distance = cosine(embedding, cluster_median)

                if distance < min_distance:
                    min_distance = distance
                    closest_cluster = cluster_id

            if min_distance < self.distance_threshold_pass2:
                updated_clusters[closest_cluster].append(face_idx)
            else:
                unassigned_faces.append(face_idx)

        # Create new clusters for unassigned
        if unassigned_faces:
            next_cluster_id = max(updated_clusters.keys()) + 1
            for face_idx in unassigned_faces:
                updated_clusters[next_cluster_id] = [face_idx]
                next_cluster_id += 1

        return updated_clusters

---

HDBSCAN Alternative (More Robust to Noise)

Advantage: Doesn't require distance threshold, automatically finds optimal clustering.

import hdbscan

class HDBSCANFaceClustering:
    """
    HDBSCAN for face clustering.
    More robust than agglomerative, doesn't need threshold tuning.
    """

    def __init__(self, min_cluster_size=3, min_samples=1):
        self.min_cluster_size = min_cluster_size
        self.min_samples = min_samples

    def cluster_faces(self, face_embeddings):
        """Cluster faces using HDBSCAN."""
        if len(face_embeddings) < self.min_cluster_size:
            return np.zeros(len(face_embeddings), dtype=int)

        embeddings = np.array(face_embeddings)

        clusterer = hdbscan.HDBSCAN(
            min_cluster_size=self.min_cluster_size,
            min_samples=self.min_samples,
            metric='cosine',
            cluster_selection_method='eom'
        )

        return clusterer.fit_predict(embeddings)

---

Method Comparison

Parameters

Face Detection:

• min_face_size: 20px (detect small faces)
• confidence_threshold: 0.9 (high confidence only)

Clustering:

• distance_threshold_pass1: 0.4 (conservative)
• distance_threshold_pass2: 0.6 (relaxed for recall)
• min_cluster_size: 3 (minimum photos of same person)

References

1. "Recognizing People in Photos Through Private On-Device Machine Learning" (Apple ML Research, 2021) 2. FaceNet: A Unified Embedding for Face Recognition and Clustering 3. HDBSCAN: Hierarchical Density-Based Spatial Clustering (2013-2025)

Perceptual Hashing Implementation Reference

Overview

Perceptual hashing generates similar hash values for visually similar images, enabling near-duplicate detection.

DINOHash (2025 State-of-the-Art)

Breakthrough: Adversarially fine-tuned self-supervised DINOv2 features.

Advantages:

• Higher bit-accuracy under heavy crops
• Robust to compression artifacts
• Resilient to adversarial attacks
• Outperforms classical DCT-DWT schemes and NeuralHash

import torch
from transformers import AutoModel, AutoImageProcessor

class DINOHasher:
    """
    DINOHash: State-of-the-art perceptual hashing using DINOv2.
    Based on: "DINOHash: Adversarially Fine-Tuned DINOv2 Features" (2025)
    """

    def __init__(self):
        self.model = AutoModel.from_pretrained('facebook/dinov2-base')
        self.processor = AutoImageProcessor.from_pretrained('facebook/dinov2-base')
        self.model.eval()
        self.hash_bits = 128

    def compute_hash(self, image):
        """Compute perceptual hash from image."""
        inputs = self.processor(images=image, return_tensors="pt")

        with torch.no_grad():
            outputs = self.model(**inputs)
            features = outputs.last_hidden_state[:, 0]  # CLS token

        features_reduced = self.project_to_hash_space(features)
        hash_binary = (features_reduced > 0).cpu().numpy().astype(np.uint8)
        return hash_binary.flatten()

    def project_to_hash_space(self, features):
        """Project high-dimensional features to hash space."""
        if not hasattr(self, 'projection_matrix'):
            self.projection_matrix = torch.randn(
                features.shape[-1], self.hash_bits
            ) / np.sqrt(features.shape[-1])
        return features @ self.projection_matrix

    def hamming_distance(self, hash1, hash2):
        """Compute Hamming distance between two hashes."""
        return np.sum(hash1 != hash2)

    def are_duplicates(self, hash1, hash2, threshold=5):
        """Check if two hashes represent near-duplicates."""
        return self.hamming_distance(hash1, hash2) <= threshold

---

Classical Perceptual Hashing

dHash (Difference Hash) - Fastest

from PIL import Image
import numpy as np

def compute_dhash(image, hash_size=8):
    """
    Compute dHash (Difference Hash).
    Fast, good for exact duplicates and minor edits.
    """
    image = image.convert('L')
    image = image.resize((hash_size + 1, hash_size), Image.LANCZOS)
    pixels = np.array(image)

    diff = pixels[:, 1:] > pixels[:, :-1]

    hash_value = 0
    for bit in diff.flatten():
        hash_value = (hash_value << 1) | int(bit)
    return hash_value


def dhash_hamming_distance(hash1, hash2):
    """Hamming distance between two dHashes."""
    return bin(hash1 ^ hash2).count('1')

pHash (Perceptual Hash) - More Robust

import cv2
from scipy.fftpack import dct

def compute_phash(image, hash_size=8):
    """
    Compute pHash using DCT.
    Better for near-duplicates with brightness/contrast changes.
    """
    image = image.convert('L')
    image = image.resize((hash_size * 4, hash_size * 4), Image.LANCZOS)
    pixels = np.array(image, dtype=np.float32)

    dct_coeff = dct(dct(pixels.T).T)
    dct_low = dct_coeff[:hash_size, :hash_size]
    median = np.median(dct_low)
    hash_binary = dct_low > median

    hash_value = 0
    for bit in hash_binary.flatten():
        hash_value = (hash_value << 1) | int(bit)
    return hash_value

---

Hybrid Duplicate Detection Pipeline

Strategy: Use fast classical hashing for filtering, deep learning for refinement.

class HybridDuplicateDetector:
    """
    Hybrid near-duplicate detection pipeline.
    Stage 1: Fast pHash filtering (eliminates obvious non-duplicates)
    Stage 2: DINOHash refinement (accurate near-duplicate detection)
    Stage 3: Siamese ViT verification (final confirmation)
    """

    def __init__(self):
        self.phash_index = {}
        self.dinohash_index = {}
        self.dino_hasher = DINOHasher()

    def add_photo(self, photo_id, image):
        """Add photo to index."""
        self.phash_index[photo_id] = compute_phash(image)
        self.dinohash_index[photo_id] = self.dino_hasher.compute_hash(image)

    def find_duplicates(self, aggressive=False):
        """Find all near-duplicate groups."""
        # Stage 1: Fast pHash pre-filtering
        phash_candidates = []
        photo_ids = list(self.phash_index.keys())

        for i in range(len(photo_ids)):
            for j in range(i + 1, len(photo_ids)):
                id1, id2 = photo_ids[i], photo_ids[j]
                distance = bin(self.phash_index[id1] ^ self.phash_index[id2]).count('1')
                if distance <= (10 if aggressive else 5):
                    phash_candidates.append((id1, id2, distance))

        # Stage 2: DINOHash refinement
        dino_duplicates = []
        for id1, id2, phash_dist in phash_candidates:
            dino_distance = self.dino_hasher.hamming_distance(
                self.dinohash_index[id1], self.dinohash_index[id2]
            )
            if dino_distance <= (10 if aggressive else 5):
                dino_duplicates.append((id1, id2, dino_distance))

        return self.cluster_duplicates(dino_duplicates)

    def cluster_duplicates(self, duplicate_pairs):
        """Cluster duplicate pairs into groups using union-find."""
        parent = {}

        def find(x):
            if x not in parent:
                parent[x] = x
            if parent[x] != x:
                parent[x] = find(parent[x])
            return parent[x]

        def union(x, y):
            root_x, root_y = find(x), find(y)
            if root_x != root_y:
                parent[root_x] = root_y

        for id1, id2, _ in duplicate_pairs:
            union(id1, id2)

        groups = {}
        for photo_id in set(id for pair in duplicate_pairs for id in pair[:2]):
            root = find(photo_id)
            groups.setdefault(root, []).append(photo_id)

        return list(groups.values())

---

BK-Tree for Efficient Search (100K+ Photos)

class BKTree:
    """
    Burkhard-Keller tree for efficient Hamming distance search.
    Enables O(log N) average-case search for perceptual hashes.
    """

    class Node:
        def __init__(self, hash_value, photo_id):
            self.hash = hash_value
            self.photo_id = photo_id
            self.children = {}

    def __init__(self):
        self.root = None

    def insert(self, photo_id, hash_value):
        """Insert photo hash into tree."""
        if self.root is None:
            self.root = self.Node(hash_value, photo_id)
        else:
            self._insert_recursive(self.root, photo_id, hash_value)

    def _insert_recursive(self, node, photo_id, hash_value):
        distance = self.hamming_distance(node.hash, hash_value)
        if distance in node.children:
            self._insert_recursive(node.children[distance], photo_id, hash_value)
        else:
            node.children[distance] = self.Node(hash_value, photo_id)

    def search(self, query_hash, threshold):
        """Find all photos within Hamming distance threshold."""
        if self.root is None:
            return []
        return self._search_recursive(self.root, query_hash, threshold)

    def _search_recursive(self, node, query_hash, threshold):
        results = []
        distance = self.hamming_distance(node.hash, query_hash)

        if distance <= threshold:
            results.append((node.photo_id, distance))

        for child_dist in range(max(0, distance - threshold),
                                distance + threshold + 1):
            if child_dist in node.children:
                results.extend(
                    self._search_recursive(node.children[child_dist],
                                         query_hash, threshold)
                )
        return results

    @staticmethod
    def hamming_distance(hash1, hash2):
        if isinstance(hash1, np.ndarray):
            return np.sum(hash1 != hash2)
        return bin(hash1 ^ hash2).count('1')

---

Performance

O(N²) for pHash comparison, but with early termination. For 10K photos:

• Stage 1 (pHash): ~5 seconds
• Stage 2 (DINOHash on candidates): ~2 seconds
• Total: ~7 seconds for full duplicate detection

Method Comparison

References

1. "DINOHash: Adversarially Fine-Tuned DINOv2 Features" (2025) 2. Neal Krawetz: dHash development 3. DCT-based pHash algorithms

Photo Indexing Pipeline Reference

Quick Indexing Before First Collage

Goal: Efficiently index 10K+ photos before creating first collage.

Strategy: Pipeline with caching, batching, and GPU acceleration.

class QuickPhotoIndexer:
    """
    Fast photo indexing pipeline for large libraries.

    Extracts:
    - Perceptual hashes (de-duplication)
    - Face embeddings (people clustering)
    - CLIP embeddings (semantic search)
    - Color palettes
    - Aesthetic scores

    Optimized for 10K photos in < 5 minutes.
    """

    def __init__(self, cache_dir='./photo_cache'):
        self.cache_dir = Path(cache_dir)
        self.cache_dir.mkdir(exist_ok=True)

        self.dino_hasher = DINOHasher()
        self.face_extractor = FaceEmbeddingExtractor()
        self.clip_model = CLIPModel.from_pretrained("openai/clip-vit-base-patch32")
        self.aesthetic_scorer = NIMAPredictor()

    def index_photo_library(self, photo_paths, batch_size=32):
        """Index entire photo library."""
        index = PhotoIndex()

        # Check cache
        cached_index = self.load_cache()
        if cached_index:
            print(f"Loaded {len(cached_index)} photos from cache")
            index = cached_index

        # Find new photos
        new_photos = [p for p in photo_paths if p not in index.photos]

        if not new_photos:
            return index

        print(f"Indexing {len(new_photos)} new photos...")

        # Process in batches
        for batch_start in range(0, len(new_photos), batch_size):
            batch_paths = new_photos[batch_start:batch_start + batch_size]
            batch_images = [Image.open(p).convert('RGB') for p in batch_paths]

            # BATCHED FEATURE EXTRACTION
            hashes = [self.dino_hasher.compute_hash(img) for img in batch_images]
            clip_embeddings = self.extract_clip_batch(batch_images)
            faces_batch = [self.face_extractor.extract_faces(img) for img in batch_images]
            palettes = [extract_palette(img) for img in batch_images]
            aesthetic_scores = self.aesthetic_scorer.predict_batch(batch_images)

            # Store in index
            for i, photo_path in enumerate(batch_paths):
                index.add_photo(
                    photo_id=str(photo_path),
                    perceptual_hash=hashes[i],
                    clip_embedding=clip_embeddings[i],
                    faces=faces_batch[i],
                    color_palette=palettes[i],
                    aesthetic_score=aesthetic_scores[i]
                )

            print(f"Indexed {batch_start + len(batch_paths)}/{len(new_photos)}")

        self.save_cache(index)

        # Post-processing
        print("Clustering faces...")
        index.cluster_faces()

        print("Detecting duplicates...")
        index.detect_duplicates()

        print("Detecting events...")
        index.detect_events()

        return index

    def extract_clip_batch(self, images):
        """Extract CLIP embeddings in batch (GPU-accelerated)."""
        from transformers import CLIPProcessor

        processor = CLIPProcessor.from_pretrained("openai/clip-vit-base-patch32")
        inputs = processor(images=images, return_tensors="pt", padding=True)
        inputs = {k: v.to(self.dino_hasher.device) for k, v in inputs.items()}

        with torch.no_grad():
            embeddings = self.clip_model.get_image_features(**inputs)

        return embeddings.cpu().numpy()

    def save_cache(self, index):
        cache_path = self.cache_dir / 'photo_index.pkl'
        with open(cache_path, 'wb') as f:
            pickle.dump(index, f)

    def load_cache(self):
        cache_path = self.cache_dir / 'photo_index.pkl'
        if cache_path.exists():
            with open(cache_path, 'rb') as f:
                return pickle.load(f)
        return None

---

PhotoIndex Container

@dataclass
class PhotoIndex:
    """Container for all photo features."""

    photos: Dict[str, Dict] = field(default_factory=dict)
    face_clusters: Dict = field(default_factory=dict)
    duplicate_groups: List = field(default_factory=list)
    events: List = field(default_factory=list)

    def add_photo(self, photo_id, **features):
        self.photos[photo_id] = features

    def cluster_faces(self):
        """Cluster all faces using HDBSCAN."""
        all_faces = []
        for photo_id, features in self.photos.items():
            for face in features.get('faces', []):
                all_faces.append({
                    'photo_id': photo_id,
                    'embedding': face['embedding']
                })

        if len(all_faces) < 3:
            return

        embeddings = [f['embedding'] for f in all_faces]
        clusterer = HDBSCANFaceClustering(min_cluster_size=3)
        labels = clusterer.cluster_faces(embeddings)

        for face, label in zip(all_faces, labels):
            if label == -1:
                continue
            self.face_clusters.setdefault(label, []).append(face)

    def detect_duplicates(self):
        """Detect duplicate photo groups."""
        detector = HybridDuplicateDetector()

        for photo_id, features in self.photos.items():
            detector.phash_index[photo_id] = features['perceptual_hash']
            detector.dinohash_index[photo_id] = features['perceptual_hash']

        self.duplicate_groups = detector.find_duplicates()

    def detect_events(self):
        """Detect temporal events (requires timestamps + GPS)."""
        # Use ST-DBSCAN from event-detection-temporal-intelligence-expert
        pass

---

Complete Curation Pipeline

def curate_photos_for_collage(photo_library_path, target_count=100):
    """
    Complete curation pipeline.

    Steps:
    1. Index all photos (quick indexing)
    2. Filter inappropriate (NSFW, screenshots, mundane)
    3. De-duplicate (keep best from each group)
    4. Cluster by person (prioritize important people)
    5. Detect events (prioritize significant events)
    6. Select diverse set
    """
    # 1. QUICK INDEXING
    indexer = QuickPhotoIndexer()
    photo_paths = list(Path(photo_library_path).glob('**/*.jpg'))
    index = indexer.index_photo_library(photo_paths)

    # 2. FILTERING
    filtered_photos = []
    for photo_id, features in index.photos.items():
        if features.get('is_nsfw', False):
            continue
        if features.get('is_screenshot', False):
            continue
        if features['aesthetic_score'] < 0.3:
            continue
        filtered_photos.append(photo_id)

    # 3. DE-DUPLICATION
    duplicates_removed = set()
    for dup_group in index.duplicate_groups:
        if len(dup_group) < 2:
            continue
        best = max(dup_group, key=lambda pid: index.photos[pid]['aesthetic_score'])
        for pid in dup_group:
            if pid != best:
                duplicates_removed.add(pid)

    filtered_photos = [p for p in filtered_photos if p not in duplicates_removed]

    # 4. PRIORITIZE IMPORTANT PEOPLE
    person_importance = {}
    for cluster_id, faces in index.face_clusters.items():
        importance = min(1.0, len(faces) / 100)
        person_importance[cluster_id] = importance

    for photo_id in filtered_photos:
        faces = index.photos[photo_id].get('faces', [])
        for face in faces:
            for cluster_id, cluster_faces in index.face_clusters.items():
                if any(f['photo_id'] == photo_id for f in cluster_faces):
                    boost = person_importance.get(cluster_id, 0) * 0.2
                    index.photos[photo_id]['aesthetic_score'] += boost

    # 5. EVENT-AWARE SELECTION
    # Use event-detection-temporal-intelligence-expert

    # 6. FINAL SELECTION
    filtered_photos.sort(
        key=lambda pid: index.photos[pid]['aesthetic_score'],
        reverse=True
    )

    return filtered_photos[:target_count]

---

Performance Benchmarks

Target Performance (Swift/Metal/Core ML):

---

Integration Points

• event-detection-temporal-intelligence-expert: Temporal event clustering
• color-theory-palette-harmony-expert: Color extraction
• collage-layout-expert: Photo selection for collages
• clip-aware-embeddings: Semantic search and similarity