Synthetic Dataset Pipeline for Object Detection¶

import torch
from transformers import CLIPModel, CLIPProcessor
from sklearn.metrics.pairwise import cosine_similarity
import numpy as np

model = CLIPModel.from_pretrained("openai/clip-vit-large-patch14")
processor = CLIPProcessor.from_pretrained("openai/clip-vit-large-patch14")

def get_embeddings(image_paths, batch_size=64):
    embeddings = []
    for i in range(0, len(image_paths), batch_size):
        batch = [Image.open(p) for p in image_paths[i:i+batch_size]]
        inputs = processor(images=batch, return_tensors="pt", padding=True)
        with torch.no_grad():
            feats = model.get_image_features(**inputs)
        embeddings.append(feats.cpu().numpy())
    return np.vstack(embeddings)

# Find duplicates
sim_matrix = cosine_similarity(embeddings)
np.fill_diagonal(sim_matrix, 0)
dup_pairs = np.argwhere(sim_matrix > 0.95)  # threshold configurable
# Keep highest-quality from each pair (resolution × file_size as proxy)

from brisque import BRISQUE
import cv2

brisque = BRISQUE()

def quality_filter(image_path, min_resolution=512, max_brisque=40):
    img = cv2.imread(image_path)
    h, w = img.shape[:2]
    if min(h, w) < min_resolution:
        return False
    score = brisque.score(img)
    return score < max_brisque  # lower score = better quality

# Alternative: JPEG artifact detection via DCT block analysis
def has_jpeg_artifacts(img, threshold=5.0):
    gray = cv2.cvtColor(img, cv2.COLOR_BGR2GRAY).astype(float)
    blocks = [gray[r:r+8, c:c+8] for r in range(0,gray.shape[0],8) 
              for c in range(0,gray.shape[1],8) if gray[r:r+8,c:c+8].shape == (8,8)]
    dct_blocks = [cv2.dct(b.astype(np.float32)) for b in blocks]
    block_noise = np.mean([np.std(b[4:, 4:]) for b in dct_blocks])
    return block_noise < threshold

from sklearn.cluster import KMeans
import cv2

def analyze_object_colors(image, mask, n_clusters=8):
    """Extract dominant colors from masked region."""
    masked_pixels = image[mask > 0]
    lab_pixels = cv2.cvtColor(
        masked_pixels.reshape(1, -1, 3).astype(np.uint8),
        cv2.COLOR_RGB2LAB
    )[0]

    kmeans = KMeans(n_clusters=n_clusters)
    kmeans.fit(lab_pixels)

    # Map cluster centers to prompt vocabulary
    return cluster_to_prompt_map(kmeans.cluster_centers_)

# SAM3 Pipeline Conditioning (per-image)
def annotate_with_sam3(image_path, prompts, pipeline):
    """
    prompts: list of text prompts from color_prompt_map.json
    pipeline: ComfyUI SAM3 pipeline
    """
    results = []
    for prompt in prompts:
        boxes, masks, scores = pipeline.run(
            image=image_path,
            prompt=prompt,
            output_format="json"
        )
        results.extend(zip(boxes, masks, scores))

    # Keep predictions above confidence threshold
    return [(box, mask, score) for box, mask, score in results if score > 0.85]

from transformers import AutoProcessor, AutoModelForCausalLM
from sam2.build_sam import build_sam2
from sam2.sam2_image_predictor import SAM2ImagePredictor

# Florence-2: open-vocabulary object detection
florence = AutoModelForCausalLM.from_pretrained("microsoft/Florence-2-large")
boxes = florence_detect(image, task="<OD>", prompt="gemstone")

# SAM2.1: boxes → pixel-precise masks
sam2 = SAM2ImagePredictor(build_sam2("sam2.1_hiera_large.pt"))
masks = sam2.predict(image, box=boxes, multimask_output=False)

# Generate descriptions → extract new prompts → re-annotate
captions = florence_caption(image, task="<CAPTION>")
new_prompts = parse_object_vocabulary(captions)  # NLP extraction
# A/B test new vs existing prompts on held-out images