如何用训练好的TensorFlow SSD MobileNet V2模型实现多摄像头推理

Alright, let's get your SSD MobileNet V2 model set up for multi-camera object detection. I’ve tackled this exact scenario before, so here’s a practical, step-by-step guide to make it work smoothly:

1. Handle Multi-Camera Stream Capture #

First, you need to read frames from multiple cameras without blocking the main process. Single-threaded capture will cause lag, so multi-threading is the way to go:

• Use cv2.VideoCapture for each camera, assigning the correct index (0, 1, 2... depending on your hardware).
• Spin up a separate thread for each camera to handle frame reading independently—this prevents one camera's lag from affecting others.

2. Adapt the Model for Batch Input #

Your single-camera code feeds one frame at a time, but multi-camera works better with batch processing (especially if you’re using a GPU for parallelism):

• Keep your preprocessing logic identical to training (resize to 300x300, normalize pixel values, adjust color channels if needed—remember OpenCV uses BGR, but most TF models expect RGB).
• Stack preprocessed frames into a batch tensor with shape [batch_size, height, width, channels] before feeding to the model.

Example Code Implementation #

import cv2
import numpy as np
import tensorflow as tf
import threading

# Load your exported SSD MobileNet V2 model
model = tf.saved_model.load('path/to/your/saved_model')
detect_fn = model.signatures['serving_default']

# Preprocessing function (match your training pipeline!)
def preprocess_frame(frame):
    # Convert BGR to RGB if your model was trained on RGB
    frame_rgb = cv2.cvtColor(frame, cv2.COLOR_BGR2RGB)
    # Resize to model's input size
    resized_frame = cv2.resize(frame_rgb, (300, 300))
    # Normalize (adjust based on your training preprocessing)
    input_tensor = resized_frame / 255.0
    # Add batch dimension for single frame
    input_tensor = tf.convert_to_tensor(input_tensor, dtype=tf.float32)
    return input_tensor[tf.newaxis, ...]

# Thread class for independent camera frame capture
class CameraCaptureThread(threading.Thread):
    def __init__(self, cam_index):
        super().__init__()
        self.cam_index = cam_index
        self.cap = cv2.VideoCapture(cam_index)
        self.current_frame = None
        self.running = True

    def run(self):
        while self.running:
            ret, frame = self.cap.read()
            if ret:
                self.current_frame = frame

    def stop(self):
        self.running = False
        self.cap.release()

# Initialize threads for your cameras (adjust count as needed)
camera_threads = [CameraCaptureThread(0), CameraCaptureThread(1)]
for thread in camera_threads:
    thread.start()

try:
    while True:
        # Collect valid frames from all active cameras
        valid_frames = []
        valid_thread_indices = []
        for idx, thread in enumerate(camera_threads):
            if thread.current_frame is not None:
                valid_frames.append(thread.current_frame.copy())
                valid_thread_indices.append(idx)
        
        if not valid_frames:
            continue

        # Prepare batch input
        batch_input = tf.concat([preprocess_frame(frame) for frame in valid_frames], axis=0)
        
        # Run detection on the batch
        detections = detect_fn(batch_input)

        # Process and display results for each frame
        for frame_idx, original_frame in enumerate(valid_frames):
            h, w, _ = original_frame.shape
            # Extract detection results for this frame
            boxes = detections['detection_boxes'][frame_idx].numpy()
            scores = detections['detection_scores'][frame_idx].numpy()
            classes = detections['detection_classes'][frame_idx].numpy()

            # Draw bounding boxes (filter by confidence threshold)
            for box, score, cls in zip(boxes, scores, classes):
                if score > 0.5:  # Adjust threshold to your needs
                    y1, x1, y2, x2 = box
                    # Convert normalized coordinates to pixel values
                    y1, x1, y2, x2 = int(y1*h), int(x1*w), int(y2*h), int(x2*w)
                    cv2.rectangle(original_frame, (x1, y1), (x2, y2), (0, 255, 0), 2)
                    cv2.putText(original_frame, f"Class {int(cls)}: {score:.2f}", 
                                (x1, y1-10), cv2.FONT_HERSHEY_SIMPLEX, 0.5, (0,255,0), 2)
            
            # Show the result for this camera
            cv2.imshow(f"Camera {camera_threads[valid_thread_indices[frame_idx]].cam_index}", original_frame)

        # Exit on 'q' press
        if cv2.waitKey(1) & 0xFF == ord('q'):
            break
finally:
    # Clean up threads and windows
    for thread in camera_threads:
        thread.stop()
    cv2.destroyAllWindows()

3. Performance Optimization Tips #

• GPU Batch Processing: If you have a GPU, increase the batch size (within your VRAM limits) to maximize parallelism—this is way faster than processing frames one by one.
• Resource Limits: On low-power devices, consider frame downsampling or reducing the model's input size (tradeoff with detection accuracy) to keep latency low.
• Thread Safety: If you need to share frame data across threads, use threading.Lock() to avoid race conditions (the example above avoids this by keeping frames per-thread).

4. Critical Checks #

• Double-check that your preprocessing matches exactly what you used during training (color space, normalization, input size)—mismatches will break detection accuracy.
• Test each camera individually first to confirm they work with your existing single-camera code, then add multi-threading.

内容的提问来源于stack exchange，提问作者abdoalii37