使用 Flask 和 Picamera2 打造 YOLO 即時物件偵測網頁介面

前言

隨著物聯網（IoT）和人工智慧（AI）的普及，即時物件偵測已成為許多應用中的核心功能。今天，我將帶你在 Raspberry Pi 上使用 Flask 和 Picamera2 模組，搭配 YOLO 模型，實現即時影像偵測，並在網頁上顯示影像流和檢測資訊。

透過這篇教學，你將學會：

如何安裝 Flask 與 YOLO 環境。
使用 Picamera2 從 Raspberry Pi 相機捕捉影像。
將即時影像與 YOLO 偵測結果透過 Flask 提供網頁串流。
在網頁上加入按鈕來暫停/恢復影像流，以及顯示檢測資訊。

準備工作

硬體需求：

Raspberry Pi 4 或 5
Raspberry Pi 相機（支援 Picamera2 模組）
網路連接（本地網路）

軟體需求：

Raspberry Pi OS（推薦最新版）
Python 3（已預裝）
Flask、OpenCV 和 YOLO 模型相關套件

步驟一：安裝必要的套件

首先，確保 Raspberry Pi 已更新系統軟體，並安裝所需的 Python 套件。

更新系統與套件：

sudo apt update
sudo apt upgrade -y

啟用虛擬環境

source venv/bin/activate

安裝 Python 套件：

pip install flask opencv-python numpy ultralytics picamera2

下載 YOLO 模型：

我們使用輕量級的 YOLO 模型 yolov8n.pt。從 Ultralytics 官方下載，並將模型放在專案目錄中。

步驟二：撰寫 Flask 和 YOLO 應用程式

我們的 Flask 應用會：

使用 Picamera2 從攝像頭捕捉影像。
使用 YOLO 模型進行物件偵測。
提供一個網頁介面顯示即時影像流。
顯示偵測資訊並提供按鈕來控制暫停和恢復功能。

完整程式碼：

from flask import Flask, Response, render_template_string, jsonify
import cv2
import numpy as np
from ultralytics import YOLO
from picamera2 import Picamera2
import threading
import time

app = Flask(__name__)

# Load the YOLOv8 model
model = YOLO("yolov8n.pt")

# Global variables for control and information sharing
is_paused = False
lock = threading.Lock()
latest_info = []

# Function to initialize and start the camera
def initialize_camera():
    picam2 = Picamera2()
    picam2.configure(picam2.create_preview_configuration(main={"size": (640, 480)}))
    picam2.start()
    return picam2

# Generate video frames for streaming
def generate_frames():
    global is_paused, latest_info
    while True:
        with lock:
            if is_paused:
                time.sleep(0.1)  # Sleep briefly to prevent tight loop when paused
                continue
        try:
            with initialize_camera() as camera:
                frame = camera.capture_array()

                # Check if the image has 4 channels and convert to 3 channels if necessary
                if frame.shape[2] == 4:
                    frame = cv2.cvtColor(frame, cv2.COLOR_BGRA2BGR)

                results = model.predict(source=frame, stream=True)
                temp_info = []
                for result in results:
                    boxes = result.boxes.xyxy.numpy()
                    for box, cls, conf in zip(boxes, result.boxes.cls.numpy(), result.boxes.conf.numpy()):
                        x1, y1, x2, y2 = map(int, box)
                        label = f"{model.names[int(cls)]}: {conf:.2f}"
                        temp_info.append(label)
                        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)
                with lock:
                    latest_info = temp_info
                _, buffer = cv2.imencode('.jpg', frame)
                yield (b'--frame\r\n' b'Content-Type: image/jpeg\r\n\r\n' + buffer.tobytes() + b'\r\n')
        except RuntimeError as e:
            print(f"RuntimeError: {e}")
            time.sleep(1)  # Wait before retrying to avoid rapid repeated failures

# Main page route
@app.route('/')
def index():
    return render_template_string('''
        <!doctype html>
        <html>
        <head>
            <title>Live YOLOv8 Object Detection</title>
            <script>
                setInterval(() => {
                    fetch('/latest_info')
                        .then(response => response.json())
                        .then(data => {
                            document.getElementById('info').innerText = "Detected: " + data.join(', ');
                        });
                }, 1000);

                function pause() {
                    fetch('/pause', { method: 'POST' }).then(() => alert('Paused'));
                }

                function resume() {
                    fetch('/resume', { method: 'POST' }).then(() => alert('Resumed'));
                }
            </script>
        </head>
        <body>
            <h1>Live YOLOv8 Object Detection</h1>
            <img src="/video_feed" width="640" height="480">
            <div id="info">Detection Info: </div>
            <button onclick="pause()">Pause</button>
            <button onclick="resume()">Resume</button>
        </body>
        </html>
    ''')

# Video feed route
@app.route('/video_feed')
def video_feed():
    return Response(generate_frames(), mimetype='multipart/x-mixed-replace; boundary=frame')

# Pause route
@app.route('/pause', methods=['POST'])
def pause():
    global is_paused
    with lock:
        is_paused = True
    return jsonify({"status": "paused"})

# Resume route
@app.route('/resume', methods=['POST'])
def resume():
    global is_paused
    with lock:
        is_paused = False
    return jsonify({"status": "resumed"})

# Latest info route
@app.route('/latest_info')
def latest_info_route():
    with lock:
        return jsonify(latest_info)

# Run the Flask app
if __name__ == "__main__":
    app.run(host="0.0.0.0", port=5000, debug=True)