讯飞AI开发者大赛-人脸情绪识别Baseline分享

各框架Baseline地址

1. Kears and Pytorch：https://gitee.com/coggle/competition-baseline/tree/master/competition/科大讯飞AI开发者大赛2021/人脸情绪识别挑战赛
2. Paddle: https://aistudio.baidu.com/aistudio/projectdetail/2194725?shared=1

本项目给出Paddle在组网时应用Paddlehub模型。目前网上均未找到此种方法，所以这里给出本人所采用的一种方法。

本项目基于阿水大佬的代码，采用Paddle实现。

Paddle代码解析

加载数据，处理

%cd /home/aistudio
!unzip ./data/data100388/Datawhale_人脸情绪识别_数据集.zip
%cd Datawhale_人脸情绪识别_数据集
!nohup unzip train.zip
!nohup unzip test.zip

导入模块

import paddle
import paddle.nn as nn
import cv2
from PIL import  Image
import numpy as np
import pandas as pd
import glob
from sklearn.model_selection import  train_test_split, StratifiedKFold, KFold
import time
import paddle.vision.transforms as transforms
import warnings
import paddlehub
# 忽略警告输出
warnings.filterwarnings("ignore")
paddle.set_device('gpu')

设置Dataset，自定义读取数据集

train_jpg = glob.glob('/home/aistudio/Datawhale_人脸情绪识别_数据集/train/*/*')
np.random.shuffle(train_jpg)
train_jpg = np.array(train_jpg)
class QRDataset(paddle.io.Dataset):
    def __init__(self, img_path, transform=None):
        self.img_path = img_path
        if transform is not None:
            self.transform = transform
        else:
            self.transform = None
    
    def __getitem__(self, index):
        start_time = time.time()
        img = Image.open(self.img_path[index]).convert('RGB')
        
        lbl_dict = {'angry': 0,
             'disgusted': 1,
             'fearful': 2,
             'happy': 3,
             'neutral': 4,
             'sad': 5,
             'surprised': 6}
        if self.transform is not None:
            img = self.transform(img)
        
        if 'test' in self.img_path[index]:
            return img, paddle.to_tensor(np.array(0))
        else:
            lbl_int = lbl_dict[self.img_path[index].split('/')[-2]]
            return img, paddle.to_tensor(np.array(lbl_int))
    
    def __len__(self):
        return len(self.img_path)

模型组网

from EfficientNet_B1 import EfficientNet_B1

class XunFeiNet(nn.Layer):
    def __init__(self):
        super(XunFeiNet, self).__init__()
        #efmodel = paddlehub.Module(name = "efficientnetb1_imagenet")
        self.model = EfficientNet_B1()
        self.fc = nn.Linear(1000, 7)
    def forward(self, img):
        out = self.model(img)
        out = self.fc(out)
        return out

对于EfficientNet_B1模块，是在本地建一个Python文件，具体需要的到下列Github路径中寻找：https://github.com/PaddlePaddle/PaddleHub/blob/release/v2.1/modules/image/classification/efficientnetb1_imagenet/module.py

设置验证、预测、训练代码，利用基础API

def validate(val_loader, model, criterion):
    model.eval()
    acc1 = []
    with paddle.no_grad():
        end = time.time()
        for i, (input_, target) in enumerate(val_loader):   
            output = model(input_)
            target = paddle.unsqueeze(paddle.to_tensor(target), 1)
            loss = criterion(output, target)
            #loss = nn.functional.sigmoid_focal_loss(output, target)
            acc1.append(paddle.metric.accuracy(output, target))

        print(' * Val [email protected] {0}'.format(np.mean(acc1)))
        return np.mean(acc1)

def predict(test_loader, model, tta=10):
    model.eval()
    
    test_pred_tta = None
    for _ in range(tta):
        test_pred = []
        with paddle.no_grad():
            end = time.time()
            for i, (input_, target) in enumerate(test_loader):
                output = model(input_)
                target = paddle.unsqueeze(paddle.to_tensor(target), 1)
                output = output.numpy()

                test_pred.append(output)
        test_pred = np.vstack(test_pred)
    
        if test_pred_tta is None:
            test_pred_tta = test_pred
        else:
            test_pred_tta += test_pred
    
    return test_pred_tta

def train(train_loader, model, criterion, optimizer, epoch):
    model.train()

    end = time.time()
    acc1 = []
    for i, (input_, target) in enumerate(train_loader):
        output = model(input_)
        target = paddle.unsqueeze(paddle.to_tensor(target), 1)
        loss = criterion(output, target)
        #loss = nn.functional.sigmoid_focal_loss(output, target)
        acc1.append(paddle.metric.accuracy(output, target))
        
        loss.backward()
        optimizer.step()
        optimizer.clear_grad()
        
        if i % 100 == 0:
            print('Train: {0}'.format(np.mean(acc1)))

设置交叉验证训练，此处只用一折

skf = KFold(n_splits=5, random_state=233, shuffle=True)
for flod_idx, (train_idx, val_idx) in enumerate(skf.split(train_jpg, train_jpg)):
    train_loader = paddle.io.DataLoader(
        QRDataset(train_jpg[train_idx][:],
                transforms.Compose([
                            transforms.ColorJitter(hue=.05, saturation=.05),
                            transforms.RandomHorizontalFlip(),
                            transforms.RandomVerticalFlip(),
                            transforms.Resize((196, 196)),
                            transforms.ToTensor(),
                            transforms.Normalize([0.485, 0.456, 0.406], [0.229, 0.224, 0.225])
                            
            ])
        ), batch_size=128, shuffle=True, num_workers=0, use_buffer_reader=True
    )
    
    val_loader = paddle.io.DataLoader(
        QRDataset(train_jpg[val_idx][:1000],
                transforms.Compose([
                            transforms.Resize((196, 196)),
                            # transforms.Resize((124, 124)),
                            # transforms.RandomCrop((88, 88)),
                            transforms.ToTensor(),
                            transforms.Normalize([0.485, 0.456, 0.406], [0.229, 0.224, 0.225])
            ])
        ), batch_size=48, shuffle=False, num_workers=0,use_buffer_reader=True
    )
        
    model = XunFeiNet()
    '''
    model = paddlehub.Module(
        name='resnet50_vd_imagenet_ssld',
        label_list=["fearful", "surprised", "happy", "disgusted", "angry", "sad", "neutral"],
        load_checkpoint=None)
    '''
    criterion = nn.CrossEntropyLoss()
    optimizer = paddle.optimizer.SGD(0.01, parameters = model.parameters())
    best_acc = 0.0
    for epoch in range(100):
        print('\nEpoch: ', epoch)

        train(train_loader, model, criterion, optimizer, epoch)
        val_acc = validate(val_loader, model, criterion)
        
        if val_acc > best_acc:
            best_acc = val_acc
            paddle.save(model.state_dict(), './resnet18_fold{0}.pdparams'.format(flod_idx))
            
    break

预测生成结果

import glob
import numpy as np
import paddle
import time
from PIL import  Image
import paddle.vision.transforms as transforms
import paddlehub
import pandas as pd

class QRDataset(paddle.io.Dataset):
    def __init__(self, img_path, transform=None):
        self.img_path = img_path
        if transform is not None:
            self.transform = transform
        else:
            self.transform = None
    
    def __getitem__(self, index):
        start_time = time.time()
        img = Image.open(self.img_path[index]).convert('RGB')
        
        lbl_dict = {'angry': 0,
             'disgusted': 1,
             'fearful': 2,
             'happy': 3,
             'neutral': 4,
             'sad': 5,
             'surprised': 6}
        if self.transform is not None:
            img = self.transform(img)
        
        if 'test' in self.img_path[index]:
            return img, paddle.to_tensor(np.array(0))
        else:
            lbl_int = lbl_dict[self.img_path[index].split('/')[-2]]
            return img, paddle.to_tensor(np.array(lbl_int))
    
    def __len__(self):
        return len(self.img_path)

def predict(test_loader, model, tta=10):
    model.eval()
    
    test_pred_tta = None
    for _ in range(tta):
        test_pred = []
        with paddle.no_grad():
            end = time.time()
            for i, (input_, target) in enumerate(test_loader):
                output = model(input_)[0]
                target = paddle.unsqueeze(paddle.to_tensor(target), 1)
                output = output.numpy()

                test_pred.append(output)
        test_pred = np.vstack(test_pred)
    
        if test_pred_tta is None:
            test_pred_tta = test_pred
        else:
            test_pred_tta += test_pred
    
    return test_pred_tta


test_jpg = glob.glob('./Datawhale_人脸情绪识别_数据集/test/*')
test_jpg = np.array(test_jpg)
test_jpg.sort()

test_loader = paddle.io.DataLoader(
        QRDataset(test_jpg,
                transforms.Compose([
                            transforms.RandomHorizontalFlip(),
                            transforms.RandomVerticalFlip(),
                            transforms.ToTensor(),
                            transforms.Normalize([0.485, 0.456, 0.406], [0.229, 0.224, 0.225])
            ])
        ), batch_size=50, shuffle=False, num_workers=5
)
        
model = paddlehub.Module(
        name='resnet50_vd_imagenet_ssld',
        label_list=["fearful", "surprised", "happy", "disgusted", "angry", "sad", "neutral"],
        load_checkpoint='resnet18_fold0.pdparams')
test_pred = predict(test_loader, model, 5)

cls_name = np.array(['angry', 'disgusted', 'fearful', 'happy','neutral', 'sad', 'surprised'])
submit_df = pd.DataFrame({'name': test_jpg, 'label': cls_name[test_pred.argmax(1)]})
submit_df['name'] = submit_df['name'].apply(lambda x: x.split('/')[-1])
submit_df = submit_df.sort_values(by='name')
submit_df.to_csv('paddle_submit.csv', index=None)

常见问题：

1. GPU和显存利用率过低，但是还提示显存爆炸。解决：设置DataLoader，num_works=0。
2. 。。。。暂无