YOLO11顔検出システムの実装

実験の目的

YOLO を用いて顔検出を実現できるようになること。

応用シーン

リアルタイム顔監視システム：防犯監視、人数カウント、スマート入退室管理などの用途で、Web インターフェースを通じてリアルタイム映像ストリームと顔検出結果を可視化して提供します。

機能要件

顔を動的に検出し、顔をマーキングします。

ハードウェアの準備

回路接続

ハードウェアモジュール

1.1ハードウェアインターフェース で RDK X5 のインターフェースを簡単に確認できます

実機の配線

下図を参考に接続し、その後ディスプレイ、キーボード・マウスを接続します（またはリモートアクセスを行います）。

ソフトウェア設計

設計フローチャート

ソースコードのダウンロード

ソースコードファイル：face.rar

（リンクをクリックしてオンラインの説明を見ることもできます）（HumansX サポートより提供）

プログラムコード

コードの構成：

具体的なコード：

完全なプロジェクトコードは（ソースコードファイル）をクリックしてダウンロードしてください。

#!/usr/bin/env python3
"""
人脸检测系统主程序 - Flask版本
基于Flask的人脸检测Web应用
"""

import logging
import signal
import sys
import time
import cv2
import numpy as np
import threading
from pathlib import Path
from typing import Optional

from flask import Flask, Response, request, jsonify, render_template

# 添加项目根目录到Python路径
sys.path.insert(0, str(Path(__file__).parent))

from config import get_config, get_model_path, FaceDetectionAppConfig
from app import FaceDetector, CameraManager

# 全局变量
face_detector: Optional[FaceDetector] = None
camera_manager: Optional[CameraManager] = None
app_config: Optional[FaceDetectionAppConfig] = None

# 应用状态管理
shutdown_requested = False
app_running = True

# 系统统计信息
system_stats = {
    'start_time': time.time(),
    'total_requests': 0,
    'detection_stats': {
        'total_frames': 0,
        'total_detections': 0,
        'fps': 0.0,
        'avg_inference_time': 0.0,
        'face_count': 0
    },
    'camera_stats': {
        'fps_actual': 0.0,
        'total_frames': 0
    }
}

# 当前设置（可动态调整）
current_settings = {
    'show_detection_boxes': True,
    'show_center_point': True,
    'show_stats': True,
    'confidence_threshold': 0.6
}

def init_logging_and_config():
    """初始化配置和日志"""
    global app_config
    app_config = get_config()

    # 配置日志
    logging.basicConfig(
        level=getattr(logging, app_config.logging.level),
        format=app_config.logging.format
    )

init_logging_and_config()
logger = logging.getLogger("FaceDetectionApp")

def init_components():
    """初始化系统组件"""
    global face_detector, camera_manager, app_config

    logger.info("🚀 正在启动人脸检测系统...")

    try:
        # 获取模型路径
        model_path = get_model_path()
        logger.info(f"模型路径: {model_path}")

        # 初始化人脸检测器
        face_detector = FaceDetector(
            model_path=str(model_path),
            conf_threshold=app_config.model.confidence_threshold,
            iou_threshold=app_config.model.iou_threshold
        )
        logger.info("✅ 人脸检测器初始化完成")

        # 初始化摄像头管理器
        camera_manager = CameraManager(
            device_path=app_config.camera.device_path,
            width=app_config.camera.width,
            height=app_config.camera.height,
            fps=app_config.camera.fps
        )

        # 启动异步捕获线程（如果配置启用）
        if app_config.camera.async_capture:
            camera_manager.start_capture_thread()

        logger.info("✅ 摄像头管理器初始化完成")

        # 更新初始设置
        current_settings.update({
            'show_detection_boxes': app_config.web.show_detection_boxes,
            'show_center_point': app_config.web.show_center_point,
            'show_stats': app_config.web.show_stats,
            'confidence_threshold': app_config.model.confidence_threshold
        })

        logger.info("🎉 人脸检测系统启动完成!")
        return True

    except Exception as e:
        logger.error(f"❌ 系统启动失败: {e}", exc_info=True)
        return False

def cleanup_resources():
    """清理系统资源"""
    global face_detector, camera_manager, shutdown_requested, app_running

    logger.info("🔄 正在关闭人脸检测系统...")
    shutdown_requested = True
    app_running = False

    try:
        if camera_manager:
            camera_manager.release()
            logger.info("✅ 摄像头资源已释放")

        # 重置全局变量
        face_detector = None
        camera_manager = None

        logger.info("✅ 人脸检测系统已关闭")

    except Exception as e:
        logger.error(f"❌ 关闭系统时发生错误: {e}", exc_info=True)

def generate_frames():
    """生成视频流帧"""
    global system_stats, shutdown_requested, app_running
    fps_counter = 0
    fps_start_time = time.time()

    while app_running and not shutdown_requested:
        try:
            # 检查摄像头状态
            if not camera_manager or not camera_manager.is_opened:
                break

            # 获取帧
            if app_config.camera.async_capture and camera_manager:
                success, frame, _ = camera_manager.get_latest_frame()
            elif camera_manager:
                success, frame = camera_manager.read_frame()
            else:
                success, frame = False, None

            if not success or frame is None:
                # 生成错误帧
                error_frame = np.zeros((480, 640, 3), dtype=np.uint8)
                cv2.putText(error_frame, "Camera Error", (200, 240), 
                           cv2.FONT_HERSHEY_SIMPLEX, 1, (0, 0, 255), 2)
                frame = error_frame

            # 执行人脸检测
            if face_detector:
                detection_result = face_detector.detect(frame)

                if detection_result['success']:
                    # 更新统计信息
                    system_stats['detection_stats']['face_count'] = detection_result['face_count']
                    system_stats['detection_stats']['total_frames'] += 1

                    detector_stats = face_detector.get_stats()
                    if detector_stats:
                        system_stats['detection_stats'].update(detector_stats)
                        if detector_stats.get('total_detections', 0) > 0:
                            system_stats['detection_stats']['avg_inference_time'] = (
                                detector_stats['total_inference_time'] / 
                                detector_stats['total_detections']
                            )

                # 绘制检测结果
                frame = face_detector.draw_results(
                    frame, 
                    detection_result,
                    show_boxes=current_settings['show_detection_boxes'],
                    show_center=current_settings['show_center_point'],
                    show_stats=current_settings['show_stats']
                )

            # 更新FPS计算
            fps_counter += 1
            current_time = time.time()
            if current_time - fps_start_time >= 1.0:
                system_stats['detection_stats']['fps'] = fps_counter / (current_time - fps_start_time)
                fps_counter = 0
                fps_start_time = current_time

            # 编码帧为JPEG
            encode_params = [cv2.IMWRITE_JPEG_QUALITY, app_config.web.video_quality]
            _, buffer = cv2.imencode('.jpg', frame, encode_params)
            frame_bytes = buffer.tobytes()

            # 生成multipart响应
            yield (b'--frame\r\n'
                   b'Content-Type: image/jpeg\r\n\r\n' + frame_bytes + b'\r\n')

            # 控制帧率
            if app_config.web.stream_fps > 0:
                time.sleep(max(0, 1.0/app_config.web.stream_fps - 0.001))

        except Exception as e:
            if app_running and not shutdown_requested:
                logger.error(f"视频流生成错误: {e}")
            time.sleep(0.1)  # 避免错误循环过快

    logger.info("视频流生成器已停止")

# 创建Flask应用
app = Flask(__name__)

# 设置模板目录（如果存在的话）
template_dir = Path(__file__).parent / "templates"
if template_dir.exists():
    app.template_folder = str(template_dir)

@app.route("/")
def root():
    """主页面"""
    system_stats['total_requests'] += 1
    # 如果有模板文件，使用模板；否则返回简单HTML
    try:
        return render_template("index.html")
    except:
        return """
        <!DOCTYPE html>
        <html>
        <head>
            <title>人脸检测系统</title>
            <meta charset="utf-8">
        </head>
        <body>
            <h1>人脸检测系统</h1>
            <img src="/video_feed" width="640" height="480" style="border: 1px solid #ccc;">
            <br>
            <p><a href="/api/health">健康检查</a> | <a href="/api/stats">系统统计</a> | <a href="/api/info">系统信息</a></p>
        </body>
        </html>
        """

@app.route("/video_feed")
def video_feed():
    """视频流端点"""
    system_stats['total_requests'] += 1
    return Response(
        generate_frames(),
        mimetype="multipart/x-mixed-replace; boundary=frame"
    )

@app.route("/api/health")
def health_check():
    """健康检查"""
    status = {
        "status": "healthy",
        "face_detector": face_detector is not None,
        "camera_manager": camera_manager is not None and camera_manager.is_opened,
        "config_loaded": app_config is not None,
        "uptime": time.time() - system_stats['start_time']
    }

    if not all([status["face_detector"], status["camera_manager"], status["config_loaded"]]):
        status["status"] = "unhealthy"
        return jsonify(status), 503

    return jsonify(status)

@app.route("/api/stats")
def get_stats():
    """获取系统统计信息"""
    stats = system_stats.copy()

    # 添加摄像头统计
    if camera_manager:
        camera_stats = camera_manager.get_stats()
        stats['camera_stats'].update(camera_stats)

    return jsonify(stats)

@app.route("/api/info")
def get_system_info():
    """获取系统信息"""
    info = {
        "app": {
            "name": app_config.app.name,
            "version": app_config.app.version,
            "environment": app_config.app.environment
        },
        "model": {
            "name": "YOLO11",
            "path": str(get_model_path()),
            "confidence_threshold": current_settings['confidence_threshold'],
            "iou_threshold": app_config.model.iou_threshold
        },
        "camera": camera_manager.get_camera_info() if camera_manager else {},
        "settings": current_settings
    }

    return jsonify(info)

@app.route("/api/settings", methods=["POST"])
def update_settings():
    """更新设置"""
    global current_settings

    data = request.get_json()
    if not data:
        return jsonify({"error": "无效的JSON数据"}), 400

    updated = {}
    for key, value in data.items():
        if key in current_settings:
            current_settings[key] = value
            updated[key] = value
            logger.info(f"设置已更新: {key} = {value}")

    # 如果更新了检测阈值，需要重新初始化检测器
    if 'confidence_threshold' in updated:
        try:
            global face_detector
            model_path = get_model_path()
            face_detector = FaceDetector(
                model_path=str(model_path),
                conf_threshold=current_settings['confidence_threshold'],
                iou_threshold=app_config.model.iou_threshold
            )
            logger.info("检测器已根据新阈值重新初始化")
        except Exception as e:
            logger.error(f"重新初始化检测器失败: {e}")
            return jsonify({"error": "更新检测参数失败"}), 500

    return jsonify({"status": "success", "updated_settings": updated})

@app.route("/api/settings/reset", methods=["POST"])
def reset_settings():
    """重置设置到默认值"""
    global current_settings

    default_settings = {
        'show_detection_boxes': app_config.web.show_detection_boxes,
        'show_center_point': app_config.web.show_center_point,
        'show_stats': app_config.web.show_stats,
        'confidence_threshold': app_config.model.confidence_threshold
    }

    current_settings.update(default_settings)

    # 重新初始化检测器
    try:
        global face_detector
        model_path = get_model_path()
        face_detector = FaceDetector(
            model_path=str(model_path),
            conf_threshold=current_settings['confidence_threshold'],
            iou_threshold=app_config.model.iou_threshold
        )
        logger.info("设置已重置，检测器已重新初始化")
    except Exception as e:
        logger.error(f"重置时重新初始化检测器失败: {e}")

    return jsonify({"status": "success", "settings": current_settings})

def signal_handler(signum, frame):
    """信号处理器"""
    global shutdown_requested, app_running

    logger.info(f"收到信号 {signum}，正在关闭应用...")
    shutdown_requested = True
    app_running = False

    # 强制中断可能的阻塞操作
    if camera_manager:
        camera_manager._capture_running = False

    # 立即开始资源清理
    cleanup_resources()

    # 强制退出程序
    import os
    os._exit(0)

def main():
    """主函数"""
    global app_config

    # 设置信号处理器
    signal.signal(signal.SIGINT, signal_handler)
    signal.signal(signal.SIGTERM, signal_handler)

    logger.info(f"🌟 启动人脸检测系统")
    logger.info(f"📍 服务地址: http://{app_config.server.host}:{app_config.server.port}")

    # 初始化系统组件
    if not init_components():
        logger.error("系统初始化失败，退出")
        sys.exit(1)

    try:
        # 启动Flask服务器
        app.run(
            host=app_config.server.host,
            port=app_config.server.port,
            debug=False,  # 禁用调试模式以避免重载问题
            threaded=True,  # 启用多线程处理
            use_reloader=False  # 禁用自动重载
        )
    except KeyboardInterrupt:
        logger.info("收到键盘中断信号")
    except Exception as e:
        logger.error(f"服务器运行错误: {e}")
    finally:
        cleanup_resources()

if __name__ == "__main__":
    main()

操作手順

操作の画面録画

リンク

具体的な操作（テキスト版）：

回路接続が完了したら、リモートでターミナルにアクセスします：

1. ターミナルにリモートアクセスします。ターミナルで次のコマンドを入力し、Enter キーを押します

ssh sunrise@192.168.8.111

パスワードは sunrise です。パスワードを入力して Enter キーを押すとログインが完了します。

2. サービスを起動します。face フォルダに移動し、メインプログラム python main.py を起動します。

Enter キーを押して正常に読み込まれると、アクセスアドレス http://192.168.8.111:8003 が表示されます

アドレスをクリックするか、ブラウザにコピーすると顔検出システムが表示されます。顔が検出されると face とマーキングされます。

実験結果

顔を動的に検出し、マーキングします。

リンク

コミュニティ

資料のダウンロード・技術サポートについては、HumansX株式会社までお問い合わせください。