[tensorflow] 面部情绪识别-基于 young-affectnet-hq 数据集的面部情绪识别模 型构建与分析

import os
import numpy as np
import cv2
import glob
import tensorflow as tf
from sklearn.preprocessing import LabelEncoder
from tensorflow.keras.utils import to_categorical
from tensorflow.keras.callbacks import EarlyStopping, ReduceLROnPlateau
from tensorflow.keras.applications import EfficientNetB0
from tensorflow.keras.layers import GlobalAveragePooling2D, Dense, Dropout, BatchNormalization
from tensorflow.keras.optimizers import Adam
from tensorflow.keras.regularizers import l2
import matplotlib.pyplot as plt
from sklearn.model_selection import train_test_split
from sklearn.metrics import confusion_matrix, classification_report
import seaborn as sns

# 定义基础目录
base_dir = "./"
os.makedirs(base_dir, exist_ok=True)

# 定义图像大小
image_size = 224

# 从指定目录加载图像的函数
def load_images_from_directory(directory):
    """
    从给定目录加载图像，返回图像数组和对应的标签数组。
    """
    images = []
    labels = []
    for emotion_folder in os.listdir(directory):
        emotion_path = os.path.join(directory, emotion_folder)
        if os.path.isdir(emotion_path):
            for image_path in glob.glob(os.path.join(emotion_path, '*')):
                img = cv2.imread(image_path)
                if img is None:
                    continue
                img = cv2.resize(img, (image_size, image_size))
                img = cv2.cvtColor(img, cv2.COLOR_BGR2RGB)
                images.append(img)
                labels.append(emotion_folder)
    return np.array(images), np.array(labels)

# 缓存图像和标签的函数
def cache_images(directory, cache_filename):
    """
    加载指定目录的图像和标签，并将其缓存为压缩的 numpy 文件。
    """
    images, labels = load_images_from_directory(directory)
    np.savez_compressed(cache_filename, images=images, labels=labels)
    print(f'将数据缓存到 {cache_filename}')

# 加载缓存图像和标签的函数
def load_cached_images(cache_filename):
    """
    加载缓存的图像和标签数据。
    """
    data = np.load(cache_filename)
    images = data['images']
    labels = data['labels']
    return images, labels

# 定义缓存文件名
train_cache_filename = os.path.join(base_dir, 'train_data.npz')
test_cache_filename = os.path.join(base_dir, 'test_data.npz')

# 检查缓存文件是否存在，若不存在则创建
if not os.path.exists(train_cache_filename) or not os.path.exists(test_cache_filename):
    cache_images(os.path.join(base_dir, 'data/train'), train_cache_filename)
    cache_images(os.path.join(base_dir, 'data/test'), test_cache_filename)

# 从缓存加载图像
train_images, train_labels = load_cached_images(train_cache_filename)
test_images, test_labels = load_cached_images(test_cache_filename)

# 合并训练集和测试集后重新划分
all_images = np.concatenate([train_images, test_images], axis=0)
all_labels = np.concatenate([train_labels, test_labels], axis=0)

# 将数据划分为训练集、验证集和测试集
train_images, temp_images, train_labels, temp_labels = train_test_split(all_images, all_labels, test_size=0.3, random_state=42)
val_images, test_images, val_labels, test_labels = train_test_split(temp_images, temp_labels, test_size=0.5, random_state=42)

# 检查并打印标签数量的函数
def print_label_counts(labels, set_name):
    """
    打印给定数据集的标签数量信息。
    """
    unique_labels, label_counts = np.unique(labels, return_counts=True)
    print(f"{set_name} 数据集的标签数量：{dict(zip(unique_labels, label_counts))}")

print_label_counts(train_labels, "训练集")
print_label_counts(val_labels, "验证集")
print_label_counts(test_labels, "测试集")

# 归一化图像
train_images = train_images.astype('float32') / 255.0
val_images = val_images.astype('float32') / 255.0
test_images = test_images.astype('float32') / 255.0

# 编码标签
label_encoder = LabelEncoder()
encoded_train_labels = label_encoder.fit_transform(train_labels)
encoded_val_labels = label_encoder.transform(val_labels)
encoded_test_labels = label_encoder.transform(test_labels)

# 独热编码标签
train_labels_one_hot = to_categorical(encoded_train_labels)
val_labels_one_hot = to_categorical(encoded_val_labels)
test_labels_one_hot = to_categorical(encoded_test_labels)

# 增强数据的函数
def augment_image(image, label):
    """
    对图像进行数据增强操作，包括随机翻转、亮度、对比度、饱和度、色调调整以及随机裁剪等。
    """
    image = tf.image.random_flip_left_right(image)
    image = tf.image.random_flip_up_down(image)
    image = tf.image.random_brightness(image, max_delta=0.2)
    image = tf.image.random_contrast(image, lower=0.8, upper=1.2)
    image = tf.image.random_saturation(image, lower=0.8, upper=1.2)
    image = tf.image.random_hue(image, max_delta=0.2)
    image = tf.image.random_crop(image, size=[image_size-20, image_size-20, 3])
    image = tf.image.resize(image, [image_size, image_size])
    return image, label

# 创建 TensorFlow 数据集的函数
def create_tf_dataset(images, labels, batch_size=32, is_train=False):
    """
    从图像和标签数组创建 TensorFlow 数据集，可根据是否为训练集进行打乱和数据增强操作。
    """
    dataset = tf.data.Dataset.from_tensor_slices((images, labels))
    if is_train:
        dataset = dataset.shuffle(buffer_size=len(images))
        dataset = dataset.map(augment_image, num_parallel_calls=tf.data.AUTOTUNE)
    dataset = dataset.batch(batch_size)
    dataset = dataset.prefetch(buffer_size=tf.data.AUTOTUNE)
    return dataset

# 创建 TensorFlow 数据集
train_dataset = create_tf_dataset(train_images, train_labels_one_hot, is_train=True)
val_dataset = create_tf_dataset(val_images, val_labels_one_hot, is_train=False)
test_dataset = create_tf_dataset(test_images, test_labels_one_hot, is_train=False)

# 加载预训练的 EfficientNetB0 模型
base_model = EfficientNetB0(weights='imagenet', include_top=False, input_shape=(image_size, image_size, 3))

# 解冻基础模型的部分层
for layer in base_model.layers[-20:]:
    layer.trainable = True

# 定义模型架构
model = tf.keras.Sequential([
    base_model,
    GlobalAveragePooling2D(),
    BatchNormalization(),
    Dense(512, activation='relu', kernel_regularizer=l2(0.01)),
    Dropout(0.5),
    BatchNormalization(),
    Dense(256, activation='relu', kernel_regularizer=l2(0.01)),
    Dropout(0.5),
    Dense(len(label_encoder.classes_), activation='softmax')
])

# 编译模型
initial_learning_rate = 0.001
optimizer = Adam(learning_rate=initial_learning_rate)
model.compile(optimizer=optimizer, loss='categorical_crossentropy', metrics=['accuracy'])

# 定义回调函数
early_stopping = EarlyStopping(monitor='val_loss', patience=15, restore_best_weights=True)
reduce_lr = ReduceLROnPlateau(monitor='val_loss', factor=0.2, patience=5, min_lr=1e-6)

# 训练模型
history = model.fit(
    train_dataset,
    epochs=50,
    validation_data=val_dataset,
    callbacks=[early_stopping, reduce_lr]
)

# 评估模型
test_loss, test_accuracy = model.evaluate(test_dataset)
print(f'测试准确率：{test_accuracy}')

# 绘制训练历史
plt.figure(figsize=(12, 4))
plt.subplot(1, 2, 1)
plt.plot(history.history['accuracy'], label='训练准确率')
plt.plot(history.history['val_accuracy'], label='验证准确率')
plt.title('模型准确率')
plt.xlabel('轮次')
plt.ylabel('准确率')
plt.legend()

plt.subplot(1, 2, 2)
plt.plot(history.history['loss'], label='训练损失')
plt.plot(history.history['val_loss'], label='验证损失')
plt.title('模型损失')
plt.xlabel('轮次')
plt.ylabel('损失')
plt.legend()

plt.tight_layout()
plt.savefig(os.path.join(base_dir, 'training_history.png'))
plt.close()

# 保存模型
model.save(os.path.join(base_dir, 'emotion_recognition_model_v2.h5'))
print("模型已保存为 'emotion_recognition_model_v2.h5'")

# 混淆矩阵
y_pred = model.predict(test_dataset)
y_pred_classes = np.argmax(y_pred, axis=1)
y_true = np.argmax(test_labels_one_hot, axis=1)

cm = confusion_matrix(y_true, y_pred_classes)
plt.figure(figsize=(10, 8))
sns.heatmap(cm, annot=True, fmt='d', cmap='Blues')
plt.title('混淆矩阵')
plt.xlabel('预测值')
plt.ylabel('真实值')
plt.savefig(os.path.join(base_dir, 'confusion_matrix.png'))
plt.close()

# 分类报告
print("\n分类报告:")
print(classification_report(y_true, y_pred_classes, target_names=label_encoder.classes_))

import os
import cv2
import numpy as np
import tensorflow as tf
from flask import Flask, request, Response, render_template

app = Flask(__name__)

# 加载人脸检测级联分类器
face_cascade = cv2.CascadeClassifier(cv2.data.haarcascades + 'haarcascade_frontalface_default.xml')

# 加载人脸情绪识别模型
model = tf.keras.models.load_model('./emotion_detection_model.h5')
emotion_labels = ['angry', 'disgust', 'fear', 'happy', 'neutral', 'sad', 'surprise']
input_size = 48

@app.route('/')
def index():
    return render_template('upload.html')

@app.route('/detect_emotion', methods=['POST'])
def detect_emotion():
    if 'image' not in request.files:
        return "No image file provided", 400
    image_file = request.files['image']
    if image_file.filename == '':
        return "No selected image file", 400
    # 保存上传的图片到临时文件
    temp_image_path = "temp_image.jpg"
    image_file.save(temp_image_path)
    # 使用 OpenCV 和模型进行情绪检测
    image = cv2.imread(temp_image_path)
    gray = cv2.cvtColor(image, cv2.COLOR_BGR2GRAY)
    faces = face_cascade.detectMultiScale(gray, scaleFactor=1.1, minNeighbors=5, minSize=(30, 30))
    for (x, y, w, h) in faces:
        cv2.rectangle(image, (x, y), (x + w, y + h), (255, 0, 0), 2)
        face_img = gray[y:y + h, x:x + w]
        resized_face = cv2.resize(face_img, (input_size, input_size))
        normalized_face = resized_face / 255.0
        expanded_face = np.expand_dims(normalized_face, axis=-1)
        expanded_face = np.expand_dims(expanded_face, axis=0)
        prediction = model.predict(expanded_face)
        emotion_label = np.argmax(prediction)
        cv2.putText(image, emotion_labels[emotion_label], (x, y - 10), cv2.FONT_HERSHEY_SIMPLEX, 0.9, (255, 0, 0), 2)
    # 保存标注后的图片为临时文件
    result_image_path = "result_image.jpg"
    cv2.imwrite(result_image_path, image)
    # 删除上传的临时图片
    os.remove(temp_image_path)
    # 返回标注后的图片
    with open(result_image_path, 'rb') as f:
        result_image = f.read()
    os.remove(result_image_path)
    return Response(result_image, mimetype='image/jpeg')

if __name__ == '__main__':
    app.run(host="0.0.0.0",debug=True)