Deploy YOLOv11 on Raspberry Pi 5 in 30 Minutes

# Raspberry Pi 5 (Bookworm OS)
sudo apt update && sudo apt install -y \
  python3-pip python3-opencv \
  libopenblas-dev libjpeg-dev

# Create virtual environment
python3 -m venv ~/robot-vision
source ~/robot-vision/bin/activate

# Install optimized PyTorch (critical for performance)
pip install torch torchvision --index-url https://download.pytorch.org/whl/cpu

# Install Ultralytics with dependencies
pip install ultralytics opencv-python-headless picamera2

# Use NVIDIA's JetPack-optimized wheels
pip install --no-cache-dir torch torchvision torchaudio --index-url https://developer.download.nvidia.com/compute/redist/jp/v60
pip install ultralytics

# export_model.py
from ultralytics import YOLO

# Download and convert to optimized format
model = YOLO("yolov11n.pt")  # 5.9 MB base model

# Export for CPU with quantization
model.export(
    format="onnx",           # OpenVINO-compatible
    int8=True,               # INT8 quantization (4x faster)
    dynamic=False,           # Fixed input size for edge
    simplify=True,           # Remove unused ops
    opset=12,                # Max compatibility
    imgsz=320                # Lower res = faster (vs 640 default)
)
print(f"Exported to: yolov11n_int8.onnx")

python export_model.py
# Copy yolov11n_int8.onnx to robot via scp

# robot_detector.py
import cv2
import numpy as np
from ultralytics import YOLO
from threading import Thread
from queue import Queue
from picamera2 import Picamera2
import time

class RobotVision:
    def __init__(self, model_path="yolov11n_int8.onnx", conf=0.45):
        # Load quantized model
        self.model = YOLO(model_path, task="detect")
        self.conf_threshold = conf
        
        # Threading for camera capture (prevents blocking)
        self.frame_queue = Queue(maxsize=2)  # Small buffer to stay current
        self.result_queue = Queue(maxsize=2)
        self.running = False
        
        # Initialize camera
        self.picam = Picamera2()
        config = self.picam.create_preview_configuration(
            main={"size": (640, 480), "format": "RGB888"},
            controls={"FrameRate": 30}  # Match inference rate
        )
        self.picam.configure(config)
        
    def start(self):
        self.running = True
        self.picam.start()
        
        # Separate threads for capture and inference
        Thread(target=self._capture_frames, daemon=True).start()
        Thread(target=self._process_frames, daemon=True).start()
        
    def _capture_frames(self):
        """Grab frames without blocking inference"""
        while self.running:
            frame = self.picam.capture_array()
            
            # Drop old frames if inference is slow
            if not self.frame_queue.full():
                self.frame_queue.put(frame)
                
    def _process_frames(self):
        """Run inference on separate thread"""
        while self.running:
            if not self.frame_queue.empty():
                frame = self.frame_queue.get()
                
                # Resize to model input (320x320 for speed)
                resized = cv2.resize(frame, (320, 320))
                
                # Run inference (non-blocking)
                results = self.model.predict(
                    resized,
                    conf=self.conf_threshold,
                    iou=0.5,
                    verbose=False,        # Disable logging
                    device="cpu",
                    half=False            # INT8 already quantized
                )[0]
                
                # Put results with original frame
                self.result_queue.put((frame, results))
                
    def get_detections(self):
        """Non-blocking get latest results"""
        if not self.result_queue.empty():
            return self.result_queue.get()
        return None, None
        
    def stop(self):
        self.running = False
        self.picam.stop()

# Usage example
if __name__ == "__main__":
    vision = RobotVision(conf=0.5)
    vision.start()
    
    fps_counter = []
    
    try:
        while True:
            start_time = time.time()
            
            frame, results = vision.get_detections()
            
            if frame is not None and results is not None:
                # Extract bounding boxes
                boxes = results.boxes.xyxy.cpu().numpy()  # x1,y1,x2,y2
                classes = results.boxes.cls.cpu().numpy()
                scores = results.boxes.conf.cpu().numpy()
                
                # Draw on frame
                for box, cls, score in zip(boxes, classes, scores):
                    x1, y1, x2, y2 = box.astype(int)
                    label = f"{results.names[int(cls)]}: {score:.2f}"
                    
                    cv2.rectangle(frame, (x1, y1), (x2, y2), (0, 255, 0), 2)
                    cv2.putText(frame, label, (x1, y1-10),
                               cv2.FONT_HERSHEY_SIMPLEX, 0.5, (0, 255, 0), 2)
                
                # Calculate FPS
                fps = 1.0 / (time.time() - start_time)
                fps_counter.append(fps)
                
                # Display (remove for headless robots)
                cv2.putText(frame, f"FPS: {fps:.1f}", (10, 30),
                           cv2.FONT_HERSHEY_SIMPLEX, 1, (0, 255, 0), 2)
                cv2.imshow("Robot Vision", frame)
                
                if cv2.waitKey(1) & 0xFF == ord('q'):
                    break
                    
    finally:
        vision.stop()
        cv2.destroyAllWindows()
        
        # Print performance stats
        avg_fps = np.mean(fps_counter[-100:])  # Last 100 frames
        print(f"\nAverage FPS: {avg_fps:.1f}")
        print(f"Average latency: {1000/avg_fps:.1f}ms")

# production_config.py
class ProductionVision(RobotVision):
    def __init__(self, **kwargs):
        super().__init__(**kwargs)
        
        # Thermal management (critical for sustained operation)
        self.check_temp_interval = 5.0  # seconds
        self.max_temp = 75  # Celsius, throttle at 80°C
        
    def _process_frames(self):
        """Enhanced with thermal monitoring"""
        last_temp_check = time.time()
        
        while self.running:
            # Check CPU temperature
            if time.time() - last_temp_check > self.check_temp_interval:
                temp = self._get_cpu_temp()
                if temp > self.max_temp:
                    print(f"WARNING: CPU at {temp}°C, throttling...")
                    time.sleep(0.5)  # Cool down
                last_temp_check = time.time()
                
            # Normal inference loop
            if not self.frame_queue.empty():
                frame = self.frame_queue.get()
                # ... rest of inference
                
    def _get_cpu_temp(self):
        """Read Raspberry Pi CPU temperature"""
        try:
            with open("/sys/class/thermal/thermal_zone0/temp", "r") as f:
                return float(f.read()) / 1000.0
        except:
            return 0.0

# Reduce inference rate when idle (battery-powered robots)
def _process_frames(self):
    idle_timeout = 2.0  # seconds
    last_detection = time.time()
    
    while self.running:
        if not self.frame_queue.empty():
            frame = self.frame_queue.get()
            results = self.model.predict(...)
            
            # Check if anything detected
            if len(results.boxes) > 0:
                last_detection = time.time()
            else:
                # No detections, reduce rate
                if time.time() - last_detection > idle_timeout:
                    time.sleep(0.1)  # 10 FPS when idle

# benchmark.py
import time
import numpy as np
from robot_detector import RobotVision

vision = RobotVision("yolov11n_int8.onnx")
vision.start()

fps_samples = []
latencies = []

for i in range(200):  # 200 frame test
    start = time.time()
    frame, results = vision.get_detections()
    
    if frame is not None:
        latency = (time.time() - start) * 1000  # ms
        latencies.append(latency)
        fps_samples.append(1000 / latency)
        
    time.sleep(0.01)  # Don't spin CPU

vision.stop()

print(f"\n=== Performance Report ===")
print(f"Average FPS: {np.mean(fps_samples):.1f}")
print(f"P50 Latency: {np.percentile(latencies, 50):.1f}ms")
print(f"P95 Latency: {np.percentile(latencies, 95):.1f}ms")
print(f"P99 Latency: {np.percentile(latencies, 99):.1f}ms")

# Increase GPU memory for better video performance
sudo nano /boot/firmware/config.txt
# Add: gpu_mem=256

# Enable performance governor (trade battery for speed)
echo performance | sudo tee /sys/devices/system/cpu/cpu*/cpufreq/scaling_governor

# Add heatsink + fan for sustained 25 FPS
# Without cooling: throttles to 15 FPS after 5 minutes

# Use TensorRT for 2x speedup
model.export(
    format="engine",         # TensorRT
    half=True,              # FP16 on Jetson GPU
    workspace=4,            # GB, increase for better optimization
    imgsz=416               # Higher res possible on Jetson
)

# Load TensorRT model
model = YOLO("yolov11n.engine", task="detect")

# ros2_yolo_node.py
import rclpy
from rclpy.node import Node
from sensor_msgs.msg import Image
from vision_msgs.msg import Detection2DArray, Detection2D
from cv_bridge import CvBridge

class YoloNode(Node):
    def __init__(self):
        super().__init__('yolo_detector')
        self.vision = RobotVision()
        self.vision.start()
        
        self.bridge = CvBridge()
        self.pub = self.create_publisher(Detection2DArray, '/detections', 10)
        
        # Process at 20Hz (below max FPS for stability)
        self.timer = self.create_timer(0.05, self.process_callback)
        
    def process_callback(self):
        frame, results = self.vision.get_detections()
        if results is not None:
            msg = Detection2DArray()
            msg.header.stamp = self.get_clock().now().to_msg()
            msg.header.frame_id = 'camera_link'
            
            for box, cls, score in zip(results.boxes.xyxy, 
                                       results.boxes.cls,
                                       results.boxes.conf):
                det = Detection2D()
                det.bbox.center.x = float((box[0] + box[2]) / 2)
                det.bbox.center.y = float((box[1] + box[3]) / 2)
                det.bbox.size_x = float(box[2] - box[0])
                det.bbox.size_y = float(box[3] - box[1])
                msg.detections.append(det)
                
            self.pub.publish(msg)