How to choose a duplicate detection method¶

import time

import cv2
import matplotlib.pyplot as plt
import numpy as np

from dataeval import config
from dataeval.extractors._bovw import BoVWExtractor
from dataeval.flags import ImageStats
from dataeval.quality import Duplicates

config.set_seed(67)  # six seven

def create_textured_image(seed: int, size: int = 128) -> np.ndarray:
    """Create an image with texture patterns that SIFT can detect."""
    rng = np.random.default_rng(seed)

    # Create base with gradient and noise
    x = np.linspace(0, 4 * np.pi, size)
    y = np.linspace(0, 4 * np.pi, size)
    xx, yy = np.meshgrid(x, y)

    # Create pattern with multiple frequency components
    pattern = (
        np.sin(xx * (1 + seed % 3)) * np.cos(yy * (2 + seed % 2))
        + np.sin(xx * 3 + seed) * 0.5
        + rng.random((size, size)) * 0.3
    )

    # Normalize to 0-255
    pattern = ((pattern - pattern.min()) / (pattern.max() - pattern.min()) * 255).astype(np.uint8)

    # Create RGB image (same pattern in all channels with slight variation)
    img = np.stack(
        [pattern, np.roll(pattern, seed % 10, axis=0), np.roll(pattern, seed % 7, axis=1)], axis=0
    )  # Shape: (3, H, W) - channels first

    return img.astype(np.uint8)

def rotate_image(img: np.ndarray, angle: float) -> np.ndarray:
    """Rotates image by any angle"""
    angle = angle % 360

    if angle == 0:
        return img

    # Transpose to HWC for OpenCV
    img_hwc = np.transpose(img, (1, 2, 0))

    # Orthogonal rotations (90, 180, 270)
    if angle % 90 == 0:
        rotate_code = {1: cv2.ROTATE_90_COUNTERCLOCKWISE, 2: cv2.ROTATE_180, 3: cv2.ROTATE_90_CLOCKWISE}
        rotated = cv2.rotate(img_hwc, rotate_code[int((angle // 90) % 4)])

    # Affine rotation (Diagonal)
    else:
        h, w = img_hwc.shape[:2]
        center = (w // 2, h // 2)
        matrix = cv2.getRotationMatrix2D(center, angle, 1.0)

        cos, sin = np.abs(matrix[0, 0]), np.abs(matrix[0, 1])
        new_w, new_h = int(h * sin + w * cos), int(h * cos + w * sin)

        matrix[0, 2] += (new_w - w) / 2
        matrix[1, 2] += (new_h - h) / 2

        rotated = cv2.warpAffine(img_hwc, matrix, (new_w, new_h), borderValue=(128, 128, 128))

    # Transpose back to CHW
    return np.transpose(rotated, (2, 0, 1))

def flip_image(img: np.ndarray, direction: str) -> np.ndarray:
    """Flip image horizontally or vertically"""
    img_hwc = np.transpose(img, (1, 2, 0))
    flipped = cv2.flip(img_hwc, 1) if direction == "horizontal" else cv2.flip(img_hwc, 0)
    return np.transpose(flipped, (2, 0, 1))

images, labels, group_info = [], [], []

experiments = [
    (
        42,
        "Group 1: Orthogonal (D4 Detectable)",
        [(rotate_image, 90, "Rot 90°"), (rotate_image, 180, "Rot 180°"), (flip_image, "horizontal", "Flip H")],
    ),
    (
        63,
        "Group 2: Diagonal (BoVW Only)",
        [(rotate_image, 45, "Rot 45°"), (rotate_image, 135, "Rot 135°"), (rotate_image, 30, "Rot 30°")],
    ),
    (
        89,
        "Group 3: Mixed Rotations",
        [(rotate_image, 90, "Rot 90°"), (rotate_image, 60, "Rot 60°"), (flip_image, "vertical", "Flip V")],
    ),
]

for seed, desc, transforms in experiments:
    start_idx = len(images)
    base_img = create_textured_image(seed=seed)

    images.append(base_img)
    labels.append(f"Original (Seed {seed})")

    for func, arg, suffix in transforms:
        images.append(func(base_img, arg))
        labels.append(f"Seed {seed} - {suffix}")

    group_info.append((desc, start_idx, len(images) - 1))

start_idx = len(images)
unique_seeds = [777, 888, 999]

for i, seed in enumerate(unique_seeds):
    images.append(create_textured_image(seed=seed))
    labels.append(f"Unique {i + 1}")

group_info.append(("Group 4: Unique Images", start_idx, len(images) - 1))

print(f"Created {len(images)} test images\n" + "=" * 60)
for desc, start, end in group_info:
    print(f"{desc:<40} (indices {start}-{end})")
print("=" * 60)

for i, label in enumerate(labels):
    print(f"  [{i:2d}] {label}")

# Visualize the test images
fig, axes = plt.subplots(3, 5, figsize=(12, 6))
axes = axes.flatten()

for i, (img, label) in enumerate(zip(images, labels)):
    if i < len(axes):
        # Convert CHW to HWC for display
        img_display = np.transpose(img, (1, 2, 0))
        axes[i].imshow(img_display)
        axes[i].set_title(f"[{i}] {label}", fontsize=8)
        axes[i].axis("off")

# Hide empty subplots
for i in range(len(images), len(axes)):
    axes[i].axis("off")

plt.tight_layout()
plt.suptitle("Test Images: 90° rotations (Group 1) vs Diagonal rotations (Group 2)", y=1.02, fontsize=14)
plt.show()

# Run D4 hash-based detection
start_time = time.time()

d4_detector = Duplicates(flags=ImageStats.HASH_DUPLICATES_D4)
d4_results = d4_detector.evaluate(images)

d4_time = time.time() - start_time
print(f"D4 Hash Detection completed in {d4_time:.3f} seconds")

print("\n=== D4 Hash Results ===")
print("\nNear duplicates (perceptual similarity):")
if d4_results.items.near:
    for group in d4_results.items.near:
        indices = list(group.indices)
        methods = sorted(group.methods)
        print(f"  Indices: {indices}")
        print(f"    Methods: {methods}")
        print(f"    Labels: {[labels[int(i)] for i in indices]}")
        print()

# Run BoVW-based detection
bovw_extractor = BoVWExtractor(vocab_size=128)  # Smaller vocab for small dataset

start_time = time.time()

bovw_detector = Duplicates(
    flags=ImageStats.NONE,  # Skip hash computation
    extractor=bovw_extractor,
    cluster_threshold=1.25,
)
bovw_results = bovw_detector.evaluate(images)

bovw_time = time.time() - start_time
print(f"BoVW Detection completed in {bovw_time:.3f} seconds")

print("\n=== BoVW Results ===")
print("\nNear duplicates (embedding similarity):")
if bovw_results.items.near:
    for group in bovw_results.items.near:
        indices = list(group.indices)
        methods = sorted(group.methods)
        print(f"  Indices: {indices}")
        print(f"    Methods: {methods}")
        print(f"    Labels: {[labels[int(i)] for i in indices]}")
        print()
else:
    print("  No near duplicates found")

print("\n=== Performance Summary ===")
print("\nExecution Time:")
print(f"  D4 Hashes:  {d4_time:.3f}s")
print(f"  BoVW:       {bovw_time:.3f}s ({bovw_time / d4_time:.1f}x slower)")

print("\nDetection Results:")
print(f"  {'Method':<15} {'Exact':<10} {'Near Groups':<15}")
print(f"  {'-' * 40}")

d4_exact = len(d4_results.items.exact) if d4_results.items.exact else 0
d4_near = len(d4_results.items.near) if d4_results.items.near else 0
print(f"  {'D4 Hashes':<15} {d4_exact:<10} {d4_near:<15}")

bovw_exact = len(bovw_results.items.exact) if bovw_results.items.exact else 0
bovw_near = len(bovw_results.items.near) if bovw_results.items.near else 0
print(f"  {'BoVW':<15} {bovw_exact:<10} {bovw_near:<15}")