CLIP for Robotics: Open-Vocabulary Object Detection in 45 Minutes

# Create isolated environment
python3 -m venv clip_robotics
source clip_robotics/bin/activate

# Core vision stack
pip install torch torchvision --index-url https://download.pytorch.org/whl/cu121
pip install transformers pillow numpy opencv-python

# Robotics integration (optional for deployment)
pip install rclpy sensor-msgs-py  # ROS2 Humble

# clip_detector.py
import torch
import torch.nn.functional as F
from transformers import CLIPProcessor, CLIPModel
from PIL import Image
import numpy as np
from typing import List, Tuple
import cv2

class OpenVocabularyDetector:
    """CLIP-based object detector with spatial localization for robotics"""
    
    def __init__(self, model_name: str = "openai/clip-vit-base-patch32"):
        self.device = "cuda" if torch.cuda.is_available() else "cpu"
        
        # Load CLIP model - base version balances speed/accuracy for robotics
        self.model = CLIPModel.from_pretrained(model_name).to(self.device)
        self.processor = CLIPProcessor.from_pretrained(model_name)
        
        # Sliding window params for spatial grounding
        self.window_sizes = [(224, 224), (336, 336)]  # Multi-scale detection
        self.stride = 112  # 50% overlap for better localization
        
    def detect(
        self, 
        image: np.ndarray,  # BGR format from OpenCV/ROS
        text_queries: List[str],
        confidence_threshold: float = 0.25
    ) -> List[dict]:
        """
        Detect objects matching text descriptions with bounding boxes.
        
        Args:
            image: Input image in BGR format (H, W, 3)
            text_queries: List of object descriptions, e.g., ["red toolbox", "safety helmet"]
            confidence_threshold: Minimum similarity score (0-1)
            
        Returns:
            List of detections: [{"bbox": (x, y, w, h), "label": str, "confidence": float}]
        """
        # Convert BGR to RGB for CLIP
        image_rgb = cv2.cvtColor(image, cv2.COLOR_BGR2RGB)
        h, w = image_rgb.shape[:2]
        
        detections = []
        
        # Encode text queries once (efficient for multiple windows)
        text_inputs = self.processor(
            text=text_queries, 
            return_tensors="pt", 
            padding=True
        ).to(self.device)
        
        with torch.no_grad():
            text_features = self.model.get_text_features(**text_inputs)
            text_features = F.normalize(text_features, dim=-1)
        
        # Sliding window detection for spatial grounding
        for window_size in self.window_sizes:
            win_h, win_w = window_size
            
            for y in range(0, h - win_h + 1, self.stride):
                for x in range(0, w - win_w + 1, self.stride):
                    # Extract window
                    window = image_rgb[y:y+win_h, x:x+win_w]
                    
                    # Encode image window
                    image_input = self.processor(
                        images=Image.fromarray(window),
                        return_tensors="pt"
                    ).to(self.device)
                    
                    with torch.no_grad():
                        image_features = self.model.get_image_features(**image_input)
                        image_features = F.normalize(image_features, dim=-1)
                    
                    # Compute similarity scores
                    similarities = (image_features @ text_features.T).squeeze(0)
                    
                    # Check each query
                    for idx, score in enumerate(similarities):
                        if score.item() > confidence_threshold:
                            detections.append({
                                "bbox": (x, y, win_w, win_h),
                                "label": text_queries[idx],
                                "confidence": score.item(),
                                "window_size": window_size
                            })
        
        # Non-maximum suppression to remove duplicate detections
        detections = self._nms(detections, iou_threshold=0.5)
        
        return detections
    
    def _nms(self, detections: List[dict], iou_threshold: float = 0.5) -> List[dict]:
        """Non-maximum suppression to filter overlapping boxes"""
        if not detections:
            return []
        
        # Group by label (only suppress same-label detections)
        label_groups = {}
        for det in detections:
            label = det["label"]
            if label not in label_groups:
                label_groups[label] = []
            label_groups[label].append(det)
        
        filtered = []
        for label, group in label_groups.items():
            # Sort by confidence
            group.sort(key=lambda x: x["confidence"], reverse=True)
            
            keep = []
            while group:
                best = group.pop(0)
                keep.append(best)
                
                # Remove overlapping detections
                group = [
                    det for det in group 
                    if self._iou(best["bbox"], det["bbox"]) < iou_threshold
                ]
            
            filtered.extend(keep)
        
        return filtered
    
    def _iou(self, box1: Tuple[int, int, int, int], 
             box2: Tuple[int, int, int, int]) -> float:
        """Calculate Intersection over Union for two bounding boxes"""
        x1, y1, w1, h1 = box1
        x2, y2, w2, h2 = box2
        
        # Intersection coordinates
        xi1 = max(x1, x2)
        yi1 = max(y1, y2)
        xi2 = min(x1 + w1, x2 + w2)
        yi2 = min(y1 + h1, y2 + h2)
        
        if xi2 <= xi1 or yi2 <= yi1:
            return 0.0
        
        intersection = (xi2 - xi1) * (yi2 - yi1)
        union = w1 * h1 + w2 * h2 - intersection
        
        return intersection / union if union > 0 else 0.0

    def visualize(self, image: np.ndarray, detections: List[dict]) -> np.ndarray:
        """Draw bounding boxes on image for debugging"""
        vis_image = image.copy()
        
        for det in detections:
            x, y, w, h = det["bbox"]
            conf = det["confidence"]
            label = det["label"]
            
            # Draw box
            color = (0, 255, 0)  # Green
            cv2.rectangle(vis_image, (x, y), (x+w, y+h), color, 2)
            
            # Draw label with confidence
            text = f"{label}: {conf:.2f}"
            cv2.putText(
                vis_image, text, (x, y-10),
                cv2.FONT_HERSHEY_SIMPLEX, 0.5, color, 2
            )
        
        return vis_image

# test_warehouse_scenario.py
import cv2
from clip_detector import OpenVocabularyDetector

def main():
    # Initialize detector
    detector = OpenVocabularyDetector()
    
    # Warehouse pick-and-place scenario
    queries = [
        "cardboard box",
        "red toolbox",
        "safety helmet",
        "metal wrench"
    ]
    
    # Capture from robot camera (or test image)
    # In production: integrate with ROS2 image topic
    image = cv2.imread("warehouse_scene.jpg")
    
    if image is None:
        print("Error: Could not load image")
        return
    
    # Run detection
    print(f"Searching for: {', '.join(queries)}")
    detections = detector.detect(image, queries, confidence_threshold=0.28)
    
    print(f"\nFound {len(detections)} objects:")
    for det in detections:
        print(f"  - {det['label']}: {det['confidence']:.3f} at {det['bbox']}")
    
    # Visualize results
    vis_image = detector.visualize(image, detections)
    cv2.imwrite("detections_output.jpg", vis_image)
    print("\nVisualization saved to detections_output.jpg")
    
    # For robot control: get highest-confidence target
    if detections:
        target = max(detections, key=lambda d: d["confidence"])
        x, y, w, h = target["bbox"]
        center_x = x + w // 2
        center_y = y + h // 2
        print(f"\nTarget for gripper: {target['label']} at pixel ({center_x}, {center_y})")

if __name__ == "__main__":
    main()

# clip_detector_node.py
import rclpy
from rclpy.node import Node
from sensor_msgs.msg import Image
from vision_msgs.msg import Detection2DArray, Detection2D, ObjectHypothesisWithPose
from cv_bridge import CvBridge
import numpy as np
from clip_detector import OpenVocabularyDetector

class CLIPDetectorNode(Node):
    """ROS2 node for CLIP-based object detection"""
    
    def __init__(self):
        super().__init__('clip_detector_node')
        
        # Parameters
        self.declare_parameter('camera_topic', '/camera/image_raw')
        self.declare_parameter('queries', ['cardboard box', 'toolbox'])
        self.declare_parameter('confidence_threshold', 0.28)
        self.declare_parameter('publish_rate', 5.0)  # Hz - balance speed/CPU
        
        camera_topic = self.get_parameter('camera_topic').value
        self.queries = self.get_parameter('queries').value
        self.conf_threshold = self.get_parameter('confidence_threshold').value
        rate = self.get_parameter('publish_rate').value
        
        # Initialize detector
        self.detector = OpenVocabularyDetector()
        self.bridge = CvBridge()
        
        # ROS2 interfaces
        self.image_sub = self.create_subscription(
            Image, camera_topic, self.image_callback, 10
        )
        self.detection_pub = self.create_publisher(
            Detection2DArray, '/clip/detections', 10
        )
        
        # Rate limiting for expensive CLIP inference
        self.last_process_time = self.get_clock().now()
        self.process_interval = 1.0 / rate  # seconds
        
        self.get_logger().info(
            f"CLIP Detector started. Searching for: {self.queries}"
        )
    
    def image_callback(self, msg: Image):
        """Process incoming camera images"""
        current_time = self.get_clock().now()
        elapsed = (current_time - self.last_process_time).nanoseconds / 1e9
        
        # Rate limit processing
        if elapsed < self.process_interval:
            return
        
        self.last_process_time = current_time
        
        # Convert ROS Image to OpenCV
        cv_image = self.bridge.imgmsg_to_cv2(msg, desired_encoding='bgr8')
        
        # Run detection
        detections = self.detector.detect(
            cv_image, self.queries, self.conf_threshold
        )
        
        # Publish detections
        if detections:
            self.publish_detections(detections, msg.header)
            self.get_logger().info(
                f"Detected {len(detections)} objects", throttle_duration_sec=2.0
            )
    
    def publish_detections(self, detections: list, header):
        """Convert detections to ROS2 Detection2DArray message"""
        msg = Detection2DArray()
        msg.header = header
        
        for det in detections:
            detection = Detection2D()
            detection.header = header
            
            # Bounding box
            x, y, w, h = det["bbox"]
            detection.bbox.center.position.x = float(x + w / 2)
            detection.bbox.center.position.y = float(y + h / 2)
            detection.bbox.size_x = float(w)
            detection.bbox.size_y = float(h)
            
            # Hypothesis (classification result)
            hypothesis = ObjectHypothesisWithPose()
            hypothesis.hypothesis.class_id = det["label"]
            hypothesis.hypothesis.score = det["confidence"]
            detection.results.append(hypothesis)
            
            msg.detections.append(detection)
        
        self.detection_pub.publish(msg)

def main(args=None):
    rclpy.init(args=args)
    node = CLIPDetectorNode()
    
    try:
        rclpy.spin(node)
    except KeyboardInterrupt:
        pass
    finally:
        node.destroy_node()
        rclpy.shutdown()

if __name__ == '__main__':
    main()

ros2 run clip_detector clip_detector_node \
  --ros-args \
  -p camera_topic:=/zed/rgb/image_rect_color \
  -p queries:="['damaged package', 'QR code', 'yellow safety vest']" \
  -p confidence_threshold:=0.30 \
  -p publish_rate:=5.0

# optimized_clip_detector.py
import torch
from transformers import CLIPModel, CLIPProcessor
import cv2
import numpy as np
from typing import List
import time

class ProductionCLIPDetector:
    """Optimized CLIP detector for real-time robotics"""
    
    def __init__(self, model_name: str = "openai/clip-vit-base-patch32"):
        self.device = "cuda" if torch.cuda.is_available() else "cpu"
        
        # Load model
        self.model = CLIPModel.from_pretrained(model_name).to(self.device)
        self.processor = CLIPProcessor.from_pretrained(model_name)
        
        # Enable half-precision on GPU for 2x speedup
        if self.device == "cuda":
            self.model = self.model.half()
            print("Using FP16 precision for faster inference")
        
        # Compile model with torch.compile for PyTorch 2.0+ (30% faster)
        try:
            self.model = torch.compile(self.model, mode="reduce-overhead")
            print("Model compiled with torch.compile")
        except Exception as e:
            print(f"torch.compile not available: {e}")
        
        # Cache text embeddings (queries rarely change in production)
        self.text_cache = {}
        
        # Adaptive window sizing based on image resolution
        self.base_stride = 112
        
    def detect_with_caching(
        self,
        image: np.ndarray,
        text_queries: List[str],
        confidence_threshold: float = 0.28
    ) -> List[dict]:
        """Detection with text embedding caching"""
        
        # Check cache for text embeddings
        cache_key = tuple(text_queries)
        if cache_key in self.text_cache:
            text_features = self.text_cache[cache_key]
        else:
            # Encode and cache
            text_inputs = self.processor(
                text=text_queries,
                return_tensors="pt",
                padding=True
            ).to(self.device)
            
            with torch.no_grad():
                text_features = self.model.get_text_features(**text_inputs)
                text_features = torch.nn.functional.normalize(text_features, dim=-1)
                
                # Keep in FP16 if model is FP16
                if self.device == "cuda":
                    text_features = text_features.half()
            
            self.text_cache[cache_key] = text_features
        
        # Adaptive window sizing based on image size
        h, w = image.shape[:2]
        if max(h, w) > 1080:
            # High-res images: use larger stride to maintain speed
            stride = int(self.base_stride * 1.5)
        else:
            stride = self.base_stride
        
        # Rest of detection pipeline...
        # (sliding window code from Step 2, using cached text_features)
        
        return detections
    
    def benchmark(self, image: np.ndarray, queries: List[str], num_runs: int = 10):
        """Measure inference speed"""
        times = []
        
        # Warmup
        for _ in range(3):
            self.detect_with_caching(image, queries)
        
        # Benchmark
        for _ in range(num_runs):
            start = time.perf_counter()
            detections = self.detect_with_caching(image, queries)
            elapsed = time.perf_counter() - start
            times.append(elapsed)
        
        avg_time = np.mean(times)
        fps = 1.0 / avg_time
        
        print(f"\nBenchmark Results ({num_runs} runs):")
        print(f"  Average time: {avg_time*1000:.1f} ms")
        print(f"  FPS: {fps:.1f}")
        print(f"  Min time: {min(times)*1000:.1f} ms")
        print(f"  Max time: {max(times)*1000:.1f} ms")

# 1. Test with sample image
python test_warehouse_scenario.py

# 2. Benchmark performance
python -c "
from optimized_clip_detector import ProductionCLIPDetector
import cv2
detector = ProductionCLIPDetector()
image = cv2.imread('warehouse_scene.jpg')
detector.benchmark(image, ['cardboard box', 'toolbox'], num_runs=20)
"

# 3. ROS2 integration test (if using ROS)
ros2 run clip_detector clip_detector_node &
ros2 topic hz /clip/detections  # Should show ~5 Hz

# Dynamic inventory handling
queries = [
    "damaged package",    # Quality control
    "package with red label",  # Sorting
    "small cardboard box",  # Size-based routing
]

# Defect detection without training examples
queries = [
    "scratched metal surface",
    "missing screw hole",
    "misaligned component",
]

# Natural language object references
queries = [
    "the blue container on the left",
    "small Allen wrench",
    "safety goggles",
]