init

.gitattributes

@@ -29,3 +29,5 @@ saved_model/**/* filter=lfs diff=lfs merge=lfs -text
 *.zip filter=lfs diff=lfs merge=lfs -text
 *.zstandard filter=lfs diff=lfs merge=lfs -text
 *tfevents* filter=lfs diff=lfs merge=lfs -text
+*.png filter=lfs diff=lfs merge=lfs -text
+*.jpg filter=lfs diff=lfs merge=lfs -text

AffectNetClass.py

@@ -0,0 +1,130 @@
+from config import DatasetName, AffectnetConf, InputDataSize, LearningConfig, ExpressionCodesAffectnet
+from config import LearningConfig, InputDataSize, DatasetName, AffectnetConf, DatasetType
+
+import numpy as np
+import os
+import matplotlib.pyplot as plt
+import math
+from datetime import datetime
+from sklearn.utils import shuffle
+from sklearn.model_selection import train_test_split
+from numpy import save, load, asarray, savez_compressed, savez
+import csv
+from skimage.io import imread
+import pickle
+import csv
+from tqdm import tqdm
+from PIL import Image
+from skimage.transform import resize
+import tensorflow as tf
+import random
+import cv2
+from skimage.feature import hog
+from skimage import data, exposure
+from matplotlib.path import Path
+from scipy import ndimage, misc
+from data_helper import DataHelper
+from sklearn.metrics import accuracy_score
+from sklearn.metrics import confusion_matrix
+from dataset_class import CustomDataset
+from sklearn.metrics import precision_recall_fscore_support as score
+
+
+class AffectNet:
+    def __init__(self, ds_type):
+        """we set the parameters needed during the whole class:
+        """
+        self.ds_type = ds_type
+        if ds_type == DatasetType.train:
+            self.img_path = AffectnetConf.no_aug_train_img_path
+            self.anno_path = AffectnetConf.no_aug_train_annotation_path
+            self.img_path_aug = AffectnetConf.aug_train_img_path
+            self.masked_img_path = AffectnetConf.aug_train_masked_img_path
+            self.anno_path_aug = AffectnetConf.aug_train_annotation_path
+
+        elif ds_type == DatasetType.eval:
+            self.img_path_aug = AffectnetConf.eval_img_path
+            self.anno_path_aug = AffectnetConf.eval_annotation_path
+            self.img_path = AffectnetConf.eval_img_path
+            self.anno_path = AffectnetConf.eval_annotation_path
+            self.masked_img_path = AffectnetConf.eval_masked_img_path
+
+        elif ds_type == DatasetType.train_7:
+            self.img_path = AffectnetConf.no_aug_train_img_path_7
+            self.anno_path = AffectnetConf.no_aug_train_annotation_path_7
+            self.img_path_aug = AffectnetConf.aug_train_img_path_7
+            self.masked_img_path = AffectnetConf.aug_train_masked_img_path_7
+            self.anno_path_aug = AffectnetConf.aug_train_annotation_path_7
+
+        elif ds_type == DatasetType.eval_7:
+            self.img_path_aug = AffectnetConf.eval_img_path_7
+            self.anno_path_aug = AffectnetConf.eval_annotation_path_7
+            self.img_path = AffectnetConf.eval_img_path_7
+            self.anno_path = AffectnetConf.eval_annotation_path_7
+            self.masked_img_path = AffectnetConf.eval_masked_img_path_7
+
+    def test_accuracy(self, model, print_samples=False):
+        dhp = DataHelper()
+
+        batch_size = LearningConfig.batch_size
+        exp_pr_glob = []
+        exp_gt_glob = []
+        acc_per_label = []
+        '''create batches'''
+        img_filenames, exp_filenames = dhp.create_generator_full_path(
+            img_path=self.img_path,
+            annotation_path=self.anno_path, label=None)
+
+        print(len(img_filenames))
+        step_per_epoch = int(len(img_filenames) // batch_size)
+        exp_pr_lbl = []
+        exp_gt_lbl = []
+
+        cds = CustomDataset()
+        ds = cds.create_dataset(img_filenames=img_filenames,
+                                anno_names=exp_filenames,
+                                is_validation=True, ds=DatasetName.affectnet)
+        batch_index = 0
+        for img_batch, exp_gt_b in ds:
+            '''predict on batch'''
+            exp_gt_b = exp_gt_b[:, -1]
+            img_batch = img_batch[:, -1, :, :]
+
+            # probab_exp_pr_b, _ = model.predict_on_batch([img_batch])  # with embedding
+            pr_data = model.predict_on_batch([img_batch])
+            probab_exp_pr_b = pr_data[0]
+
+            scores_b = np.array([tf.nn.softmax(probab_exp_pr_b[i]) for i in range(len(probab_exp_pr_b))])
+            exp_pr_b = np.array([np.argmax(scores_b[i]) for i in range(len(probab_exp_pr_b))])
+
+            if print_samples:
+                for i in range(len(exp_pr_b)):
+                    dhp.test_image_print_exp(str(i) + str(batch_index + 1), np.array(img_batch[i]),
+                                             np.int8(exp_gt_b[i]), np.int8(exp_pr_b[i]))
+
+            exp_pr_lbl += np.array(exp_pr_b).tolist()
+            exp_gt_lbl += np.array(exp_gt_b).tolist()
+            batch_index += 1
+        exp_pr_lbl = np.int64(np.array(exp_pr_lbl))
+        exp_gt_lbl = np.int64(np.array(exp_gt_lbl))
+
+        global_accuracy = accuracy_score(exp_gt_lbl, exp_pr_lbl)
+
+        precision, recall, fscore, support = score(exp_gt_lbl, exp_pr_lbl)
+
+        conf_mat = confusion_matrix(exp_gt_lbl, exp_pr_lbl) / 500.0
+        # conf_mat = tf.math.confusion_matrix(exp_gt_lbl, exp_pr_lbl, num_classes=7)/500.0
+        avg_acc = np.mean([conf_mat[i, i] for i in range(7)])
+
+        ds = None
+        face_img_filenames = None
+        eyes_img_filenames = None
+        nose_img_filenames = None
+        mouth_img_filenames = None
+        exp_filenames = None
+        global_bunch = None
+        upper_bunch = None
+        middle_bunch = None
+        bottom_bunch = None
+
+        return global_accuracy, conf_mat, avg_acc, precision, recall, fscore, support

FerPlusClass.py

@@ -0,0 +1,143 @@
+from config import DatasetName, AffectnetConf, InputDataSize, LearningConfig, ExpressionCodesAffectnet
+from config import LearningConfig, InputDataSize, DatasetName, FerPlusConf, DatasetType
+
+import numpy as np
+import os
+import matplotlib.pyplot as plt
+import math
+from datetime import datetime
+from sklearn.utils import shuffle
+from sklearn.model_selection import train_test_split
+from numpy import save, load, asarray, savez_compressed
+import csv
+from skimage.io import imread
+import pickle
+import csv
+from tqdm import tqdm
+from PIL import Image
+from skimage.transform import resize
+import tensorflow as tf
+import random
+import cv2
+from skimage.feature import hog
+from skimage import data, exposure
+from matplotlib.path import Path
+from scipy import ndimage, misc
+from data_helper import DataHelper
+from sklearn.metrics import accuracy_score
+from sklearn.metrics import confusion_matrix
+from shutil import copyfile
+from dataset_class import CustomDataset
+from sklearn.metrics import precision_recall_fscore_support as score
+
+
+class FerPlus:
+
+    def __init__(self, ds_type):
+        """we set the parameters needed during the whole class:
+        """
+        self.ds_type = ds_type
+        if ds_type == DatasetType.train:
+            self.img_path = FerPlusConf.no_aug_train_img_path
+            self.anno_path = FerPlusConf.no_aug_train_annotation_path
+            self.img_path_aug = FerPlusConf.aug_train_img_path
+            self.anno_path_aug = FerPlusConf.aug_train_annotation_path
+            self.masked_img_path = FerPlusConf.aug_train_masked_img_path
+            self.orig_image_path = FerPlusConf.orig_image_path_train
+
+        elif ds_type == DatasetType.test:
+            self.img_path = FerPlusConf.test_img_path
+            self.anno_path = FerPlusConf.test_annotation_path
+            self.img_path_aug = FerPlusConf.test_img_path
+            self.anno_path_aug = FerPlusConf.test_annotation_path
+            self.masked_img_path = FerPlusConf.test_masked_img_path
+            self.orig_image_path = FerPlusConf.orig_image_path_train
+
+    def create_from_orig(self):
+        print('create_from_orig & relabel to affectNetLike--->')
+        """
+        labels are from 1-7, but we save them from 0 to 6
+        :param ds_type:
+        :return:
+        """
+
+        '''read the text file, and save exp, and image'''
+        dhl = DataHelper()
+
+        exp_affectnet_like_lbls = [6, 5, 4, 1, 0, 2, 3]
+        lbl_affectnet_like_lbls = ['angry/', 'disgust/', 'fear/', 'happy/', 'neutral/', 'sad/', 'surprise/']
+
+        for exp_index in range(len(lbl_affectnet_like_lbls)):
+            exp_prefix = lbl_affectnet_like_lbls[exp_index]
+            for i, file in tqdm(enumerate(os.listdir(self.orig_image_path + exp_prefix))):
+                if file.endswith(".jpg") or file.endswith(".png"):
+                    img_source_address = self.orig_image_path + exp_prefix + file
+                    img_dest_address = self.img_path + file
+                    exp_dest_address = self.anno_path + file[:-4]
+                    exp = exp_affectnet_like_lbls[exp_index]
+
+                    img = np.array(Image.open(img_source_address))
+                    res_img = resize(img, (InputDataSize.image_input_size, InputDataSize.image_input_size, 3),
+                                     anti_aliasing=True)
+
+                    im = Image.fromarray(np.round(res_img * 255.0).astype(np.uint8))
+                    '''save image'''
+                    im.save(img_dest_address)
+                    '''save annotation'''
+                    np.save(exp_dest_address + '_exp', exp)
+
+    def test_accuracy(self, model, print_samples=False):
+        print('FER: test_accuracy')
+        dhp = DataHelper()
+        '''create batches'''
+        img_filenames, exp_filenames = dhp.create_generator_full_path(
+            img_path=self.img_path,
+            annotation_path=self.anno_path, label=None)
+        print(len(img_filenames))
+        exp_pr_lbl = []
+        exp_gt_lbl = []
+
+        cds = CustomDataset()
+        ds = cds.create_dataset(img_filenames=img_filenames,
+                                anno_names=exp_filenames,
+                                is_validation=True,
+                                ds=DatasetName.fer2013)
+
+        batch_index = 0
+        print('FER: loading test ds')
+        for img_batch, exp_gt_b in tqdm(ds):
+            '''predict on batch'''
+            exp_gt_b = exp_gt_b[:, -1]
+            img_batch = img_batch[:, -1, :, :]
+
+            pr_data = model.predict_on_batch([img_batch])
+            probab_exp_pr_b = pr_data[0]
+
+            scores_b = np.array([tf.nn.softmax(probab_exp_pr_b[i]) for i in range(len(probab_exp_pr_b))])
+            exp_pr_b = np.array([np.argmax(scores_b[i]) for i in range(len(probab_exp_pr_b))])
+
+            if print_samples:
+                for i in range(len(exp_pr_b)):
+                    dhp.test_image_print_exp(str(i)+str(batch_index+1), np.array(img_batch[i]),
+                                             np.int8(exp_gt_b[i]), np.int8(exp_pr_b[i]))
+
+            exp_pr_lbl += np.array(exp_pr_b).tolist()
+            exp_gt_lbl += np.array(exp_gt_b).tolist()
+            batch_index += 1
+        exp_pr_lbl = np.float64(np.array(exp_pr_lbl))
+        exp_gt_lbl = np.float64(np.array(exp_gt_lbl))
+
+        global_accuracy = accuracy_score(exp_gt_lbl, exp_pr_lbl)
+        precision, recall, fscore, support = score(exp_gt_lbl, exp_pr_lbl)
+        conf_mat = confusion_matrix(exp_gt_lbl, exp_pr_lbl, normalize='true')
+        # conf_mat = tf.math.confusion_matrix(exp_gt_lbl, exp_pr_lbl, num_classes=7)
+        avg_acc = np.mean([conf_mat[i,i] for i in range(7)])
+
+        ds = None
+        face_img_filenames = None
+        eyes_img_filenames = None
+        nose_img_filenames = None
+        mouth_img_filenames = None
+        exp_filenames = None
+
+        return global_accuracy, conf_mat, avg_acc, precision, recall, fscore, support

LICENSE

@@ -0,0 +1,21 @@
+MIT License
+
+Copyright (c) 2022 Ali Pourramezan Fard
+
+Permission is hereby granted, free of charge, to any person obtaining a copy
+of this software and associated documentation files (the "Software"), to deal
+in the Software without restriction, including without limitation the rights
+to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
+copies of the Software, and to permit persons to whom the Software is
+furnished to do so, subject to the following conditions:
+
+The above copyright notice and this permission notice shall be included in all
+copies or substantial portions of the Software.
+
+THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
+IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
+FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
+AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
+LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
+OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
+SOFTWARE.

README.md

@@ -1,3 +1,97 @@
----
-license: mit
----
+# Ad-Corre
+Ad-Corre: Adaptive Correlation-Based Loss for Facial Expression Recognition in the Wild
+
+	
+  [![PWC](https://img.shields.io/endpoint.svg?url=https://paperswithcode.com/badge/ad-corre-adaptive-correlation-based-loss-for/facial-expression-recognition-on-raf-db)](https://paperswithcode.com/sota/facial-expression-recognition-on-raf-db?p=ad-corre-adaptive-correlation-based-loss-for)
+<!-- 
+[![PWC](https://img.shields.io/endpoint.svg?url=https://paperswithcode.com/badge/ad-corre-adaptive-correlation-based-loss-for/facial-expression-recognition-on-affectnet)](https://paperswithcode.com/sota/facial-expression-recognition-on-affectnet?p=ad-corre-adaptive-correlation-based-loss-for)
+
+[![PWC](https://img.shields.io/endpoint.svg?url=https://paperswithcode.com/badge/ad-corre-adaptive-correlation-based-loss-for/facial-expression-recognition-on-fer2013)](https://paperswithcode.com/sota/facial-expression-recognition-on-fer2013?p=ad-corre-adaptive-correlation-based-loss-for)
+ -->
+
+#### Link to the paper (open access):
+https://ieeexplore.ieee.org/document/9727163
+
+#### Link to the paperswithcode.com:
+https://paperswithcode.com/paper/ad-corre-adaptive-correlation-based-loss-for
+
+```
+Please cite this work as:
+
+@ARTICLE{9727163,
+  author={Fard, Ali Pourramezan and Mahoor, Mohammad H.},
+  journal={IEEE Access}, 
+  title={Ad-Corre: Adaptive Correlation-Based Loss for Facial Expression Recognition in the Wild}, 
+  year={2022},
+  volume={},
+  number={},
+  pages={1-1},
+  doi={10.1109/ACCESS.2022.3156598}}
+  
+```
+
+## Introduction
+
+Automated Facial Expression Recognition (FER) in the wild using deep neural networks is still challenging due to intra-class variations and inter-class similarities in facial images. Deep Metric Learning (DML) is among the widely used methods to deal with these issues by improving the discriminative power of the learned embedded features. This paper proposes an Adaptive Correlation (Ad-Corre) Loss to guide the network towards generating embedded feature vectors with high correlation for within-class samples and less correlation for between-class samples. Ad-Corre consists of 3 components called Feature Discriminator, Mean Discriminator, and Embedding Discriminator. We design the Feature Discriminator component to guide the network to create the embedded feature vectors to be highly correlated if they belong to a similar class, and less correlated if they belong to different classes. In addition, the Mean Discriminator component leads the network to make the mean embedded feature vectors of different classes to be less similar to each other.We use Xception network as the backbone of our model, and contrary to previous work, we propose an embedding feature space that contains k feature vectors. Then, the Embedding Discriminator component penalizes the network to generate the embedded feature vectors, which are dissimilar.We trained our model using the combination of our proposed loss functions called Ad-Corre Loss jointly with the cross-entropy loss. We achieved a very promising recognition accuracy on AffectNet, RAF-DB, and FER-2013. Our extensive experiments and ablation study indicate the power of our method to cope well with challenging FER tasks in the wild.
+
+
+## Evaluation and Samples
+The following samples are taken from the paper:
+
+![Samples](https://github.com/aliprf/Ad-Corre/blob/main/paper_graphical_items/samples.jpg?raw=true)
+
+
+----------------------------------------------------------------------------------------------------------------------------------
+## Installing the requirements
+In order to run the code you need to install python >= 3.5. 
+The requirements and the libraries needed to run the code can be installed using the following command:
+
+```
+  pip install -r requirements.txt
+```
+
+
+## Using the pre-trained models
+The pretrained models for Affectnet, RafDB, and Fer2013 are provided in the [Trained_Models](https://github.com/aliprf/Ad-Corre/tree/main/Trained_Models) folder. You can use the following code to predict the facial emotionn of a facial image:
+  
+```
+    tester = TestModels(h5_address='./trained_models/AffectNet_6336.h5')
+    tester.recognize_fer(img_path='./img.jpg')
+
+```
+plaese see the following [main.py](https://github.com/aliprf/Ad-Corre/tree/main/main.py) file.
+
+
+## Training Network from scratch
+The information and the code to train the model is provided in train.py .Plaese see the following [main.py](https://github.com/aliprf/Ad-Corre/tree/main/main.py) file:
+
+```
+    '''training part'''
+    trainer = TrainModel(dataset_name=DatasetName.affectnet, ds_type=DatasetType.train_7)
+    trainer.train(arch="xcp", weight_path="./")
+
+```
+
+
+### Preparing Data
+Data needs to be normalized and saved in npy format. 
+
+---------------------------------------------------------------
+
+```
+Please cite this work as:
+
+@ARTICLE{9727163,
+  author={Fard, Ali Pourramezan and Mahoor, Mohammad H.},
+  journal={IEEE Access}, 
+  title={Ad-Corre: Adaptive Correlation-Based Loss for Facial Expression Recognition in the Wild}, 
+  year={2022},
+  volume={},
+  number={},
+  pages={1-1},
+  doi={10.1109/ACCESS.2022.3156598}}
+
+```
+
+
+

RafdbClass.py

@@ -0,0 +1,109 @@
+from config import DatasetName, AffectnetConf, InputDataSize, LearningConfig, ExpressionCodesAffectnet
+from config import LearningConfig, InputDataSize, DatasetName, RafDBConf, DatasetType
+from sklearn.metrics import precision_recall_fscore_support as score
+
+import numpy as np
+import os
+import matplotlib.pyplot as plt
+import math
+from datetime import datetime
+from sklearn.utils import shuffle
+from sklearn.model_selection import train_test_split
+from numpy import save, load, asarray, savez_compressed
+import csv
+from skimage.io import imread
+import pickle
+import csv
+from tqdm import tqdm
+from PIL import Image
+from skimage.transform import resize
+import tensorflow as tf
+import random
+import cv2
+from skimage.feature import hog
+from skimage import data, exposure
+from matplotlib.path import Path
+from scipy import ndimage, misc
+from data_helper import DataHelper
+from sklearn.metrics import accuracy_score
+from sklearn.metrics import confusion_matrix
+from shutil import copyfile
+from dataset_class import CustomDataset
+
+
+class RafDB:
+    def __init__(self, ds_type):
+        """we set the parameters needed during the whole class:
+        """
+        self.ds_type = ds_type
+        if ds_type == DatasetType.train:
+            self.img_path = RafDBConf.no_aug_train_img_path
+            self.anno_path = RafDBConf.no_aug_train_annotation_path
+            self.img_path_aug = RafDBConf.aug_train_img_path
+            self.anno_path_aug = RafDBConf.aug_train_annotation_path
+            self.masked_img_path = RafDBConf.aug_train_masked_img_path
+
+        elif ds_type == DatasetType.test:
+            self.img_path = RafDBConf.test_img_path
+            self.anno_path = RafDBConf.test_annotation_path
+            self.img_path_aug = RafDBConf.test_img_path
+            self.anno_path_aug = RafDBConf.test_annotation_path
+            self.masked_img_path = RafDBConf.test_masked_img_path
+
+    def test_accuracy(self, model, print_samples=False):
+        dhp = DataHelper()
+        batch_size = LearningConfig.batch_size
+        # exp_pr_glob = []
+        # exp_gt_glob = []
+        # acc_per_label = []
+        '''create batches'''
+        img_filenames, exp_filenames = dhp.create_generator_full_path(
+            img_path=self.img_path,
+            annotation_path=self.anno_path, label=None)
+        print(len(img_filenames))
+        step_per_epoch = int(len(img_filenames) // batch_size)
+        exp_pr_lbl = []
+        exp_gt_lbl = []
+
+        cds = CustomDataset()
+        ds = cds.create_dataset(img_filenames=img_filenames,
+                                anno_names=exp_filenames,
+                                is_validation=True,
+                                ds=DatasetName.rafdb)
+
+        batch_index = 0
+        for img_batch, exp_gt_b in tqdm(ds):
+            '''predict on batch'''
+            exp_gt_b = exp_gt_b[:, -1]
+            img_batch = img_batch[:, -1, :, :]
+
+            pr_data = model.predict_on_batch([img_batch])
+
+            probab_exp_pr_b = pr_data[0]
+            exp_pr_b = np.array([np.argmax(probab_exp_pr_b[i]) for i in range(len(probab_exp_pr_b))])
+
+            if print_samples:
+                for i in range(len(exp_pr_b)):
+                    dhp.test_image_print_exp(str(i) + str(batch_index + 1), np.array(img_batch[i]),
+                                             np.int8(exp_gt_b[i]), np.int8(exp_pr_b[i]))
+
+            exp_pr_lbl += np.array(exp_pr_b).tolist()
+            exp_gt_lbl += np.array(exp_gt_b).tolist()
+            batch_index += 1
+        exp_pr_lbl = np.float64(np.array(exp_pr_lbl))
+        exp_gt_lbl = np.float64(np.array(exp_gt_lbl))
+
+        global_accuracy = accuracy_score(exp_gt_lbl, exp_pr_lbl)
+        precision, recall, fscore, support = score(exp_gt_lbl, exp_pr_lbl)
+
+        conf_mat = confusion_matrix(exp_gt_lbl, exp_pr_lbl, normalize='true')
+        avg_acc = np.mean([conf_mat[i, i] for i in range(7)])
+
+        ds = None
+        face_img_filenames = None
+        eyes_img_filenames = None
+        nose_img_filenames = None
+        mouth_img_filenames = None
+        exp_filenames = None
+
+        return global_accuracy, conf_mat, avg_acc, precision, recall, fscore, support

cnn_model.py

@@ -0,0 +1,116 @@
+from config import DatasetName, AffectnetConf, InputDataSize, LearningConfig
+# from hg_Class import HourglassNet
+
+import tensorflow as tf
+# from tensorflow import keras
+# from skimage.transform import resize
+from keras.models import Model
+
+from keras.applications import mobilenet_v2, mobilenet, resnet50, densenet, resnet
+from keras.layers import Dense, MaxPooling2D, Conv2D, Flatten, \
+    BatchNormalization, Activation, GlobalAveragePooling2D, DepthwiseConv2D, \
+    Dropout, ReLU, Concatenate, Input, GlobalMaxPool2D, LeakyReLU, Softmax, ELU
+
+class CNNModel:
+    def get_model(self, arch, num_of_classes, weights):
+        if arch == 'resnet':
+            model = self._create_resnetemb(num_of_classes,
+                                           num_of_embeddings=LearningConfig.num_embeddings,
+                                           input_shape=(InputDataSize.image_input_size,
+                                                        InputDataSize.image_input_size, 3),
+                                           weights=weights
+                                           )
+        if arch == 'xcp':
+            model = self._create_Xception_l2(num_of_classes,
+                                             num_of_embeddings=LearningConfig.num_embeddings,
+                                             input_shape=(InputDataSize.image_input_size,
+                                                          InputDataSize.image_input_size, 3),
+                                             weights=weights
+                                             )
+
+        return model
+
+    def _create_resnetemb(self, num_of_classes, input_shape, weights, num_of_embeddings):
+        resnet_model = resnet.ResNet50(
+            input_shape=input_shape,
+            include_top=True,
+            weights='imagenet',
+            # weights=None,
+            input_tensor=None,
+            pooling=None)
+        resnet_model.layers.pop()
+
+        avg_pool = resnet_model.get_layer('avg_pool').output  # 2048
+        ''''''
+        embeddings = []
+        for i in range(num_of_embeddings):
+            emb = tf.keras.layers.Dense(LearningConfig.embedding_size, activation=None)(avg_pool)
+            emb_l2 = tf.keras.layers.Lambda(lambda x: tf.math.l2_normalize(x, axis=1))(emb)
+            embeddings.append(emb_l2)
+
+        if num_of_embeddings > 1:
+            fused = tf.keras.layers.Concatenate(axis=1)([embeddings[i] for i in range(num_of_embeddings)])
+        else:
+            fused = embeddings[0]
+
+        fused = Dropout(rate=0.5)(fused)
+
+        '''out'''
+        out_categorical = Dense(num_of_classes,
+                                activation='softmax',
+                                name='out')(fused)
+
+        inp = [resnet_model.input]
+
+        revised_model = Model(inp, [out_categorical] + [embeddings[i] for i in range(num_of_embeddings)])
+        revised_model.summary()
+        '''save json'''
+        model_json = revised_model.to_json()
+
+        with open("./model_archs/resnetemb.json", "w") as json_file:
+            json_file.write(model_json)
+
+        return revised_model
+
+    def _create_Xception_l2(self, num_of_classes, num_of_embeddings, input_shape, weights):
+        xception_model = tf.keras.applications.Xception(
+            include_top=False,
+            # weights=None,
+            input_tensor=None,
+            weights='imagenet',
+            input_shape=input_shape,
+            pooling=None,
+            classes=num_of_classes
+        )
+
+        act_14 = xception_model.get_layer('block14_sepconv2_act').output
+        avg_pool = GlobalAveragePooling2D()(act_14)
+
+        embeddings = []
+        for i in range(num_of_embeddings):
+            emb = tf.keras.layers.Dense(LearningConfig.embedding_size, activation=None)(avg_pool)
+            emb_l2 = tf.keras.layers.Lambda(lambda x: tf.math.l2_normalize(x, axis=1))(emb)
+
+            embeddings.append(emb_l2)
+        if num_of_embeddings > 1:
+            fused = tf.keras.layers.Concatenate(axis=1)([embeddings[i] for i in range(num_of_embeddings)])
+        else:
+            fused = embeddings[0]
+        fused = Dropout(rate=0.5)(fused)
+
+        '''out'''
+        out_categorical = Dense(num_of_classes,
+                                activation='softmax',
+                                name='out')(fused)
+
+        inp = [xception_model.input]
+
+        revised_model = Model(inp, [out_categorical] + [embeddings[i] for i in range(num_of_embeddings)])
+        revised_model.summary()
+        '''save json'''
+        model_json = revised_model.to_json()
+
+        with open("./model_archs/xcp_embedding.json", "w") as json_file:
+            json_file.write(model_json)
+
+        return revised_model

config.py

@@ -0,0 +1,145 @@
+class DatasetName:
+    affectnet = 'affectnet'
+    rafdb = 'rafdb'
+    fer2013 = 'fer2013'
+
+
+class ExpressionCodesRafdb:
+    Surprise = 1
+    Fear = 2
+    Disgust = 3
+    Happiness = 4
+    Sadness = 5
+    Anger = 6
+    Neutral = 7
+
+class ExpressionCodesAffectnet:
+    neutral = 0
+    happy = 1
+    sad = 2
+    surprise = 3
+    fear = 4
+    disgust = 5
+    anger = 6
+    contempt = 7
+    none = 8
+    uncertain = 9
+    noface = 10
+
+class DatasetType:
+    train = 0
+    train_7 = 1
+    eval = 2
+    eval_7 = 3
+    test = 4
+
+
+class LearningConfig:
+    # batch_size = 100
+    batch_size = 50
+    # batch_size = 1
+    # batch_size = 5
+    virtual_batch_size = 5* batch_size
+    epochs = 300
+    embedding_size = 256 # we have 360 filters at the end
+    # embedding_size = 128 # we have 360 filters at the end
+    labels_history_frame = 1000
+    num_classes = 7
+    # num_embeddings = 7
+    num_embeddings = 10
+
+
+class InputDataSize:
+    image_input_size = 224
+
+
+class FerPlusConf:
+    _prefix_path = './FER_plus'  # --> zeue
+
+    orig_image_path_train = _prefix_path + '/orig/train/'
+    orig_image_path_test = _prefix_path + '/orig/test/'
+
+    '''only 7 labels'''
+    no_aug_train_img_path = _prefix_path + '/train_set/images/'
+    no_aug_train_annotation_path = _prefix_path + '/train_set/annotations/'
+
+    aug_train_img_path = _prefix_path + '/train_set_aug/images/'
+    aug_train_annotation_path = _prefix_path + '/train_set_aug/annotations/'
+    aug_train_masked_img_path = _prefix_path + '/train_set_aug/masked_images/'
+
+    weight_save_path = _prefix_path + '/weight_saving_path/'
+
+    '''both public&private test:'''
+    test_img_path = _prefix_path + '/test_set/images/'
+    test_annotation_path = _prefix_path + '/test_set/annotations/'
+    test_masked_img_path = _prefix_path + '/test_set/masked_images/'
+
+    '''private test:'''
+    private_test_img_path = _prefix_path + '/private_test_set/images/'
+    private_test_annotation_path = _prefix_path + '/private_test_set/annotations/'
+    '''public test-> Eval'''
+    public_test_img_path = _prefix_path + '/public_test_set/images/'
+    public_test_annotation_path = _prefix_path + '/public_test_set/annotations/'
+
+
+class RafDBConf:
+    _prefix_path = './RAF-DB'  #
+
+    orig_annotation_txt_path = _prefix_path + '/list_patition_label.txt'
+    orig_image_path = _prefix_path + '/original/'
+    orig_bounding_box = _prefix_path + '/boundingbox/'
+
+    '''only 7 labels'''
+    no_aug_train_img_path = _prefix_path + '/train_set/images/'
+    no_aug_train_annotation_path = _prefix_path + '/train_set/annotations/'
+
+    aug_train_img_path = _prefix_path + '/train_set_aug/images/'
+    aug_train_annotation_path = _prefix_path + '/train_set_aug/annotations/'
+    aug_train_masked_img_path = _prefix_path + '/train_set_aug/masked_images/'
+
+    test_img_path = _prefix_path + '/test_set/images/'
+    test_annotation_path = _prefix_path + '/test_set/annotations/'
+    test_masked_img_path = _prefix_path + '/test_set/masked_images/'
+
+    augmentation_factor = 5
+
+    weight_save_path = _prefix_path + '/weight_saving_path/'
+
+
+class AffectnetConf:
+    """"""
+    '''atlas'''
+    _prefix_path = './affectnet'  # --> Aq
+
+    orig_csv_train_path = _prefix_path + '/orig/training.csv'
+    orig_csv_evaluate_path = _prefix_path + '/orig/validation.csv'
+
+    '''8 labels'''
+    no_aug_train_img_path = _prefix_path + '/train_set/images/'
+    no_aug_train_annotation_path = _prefix_path + '/train_set/annotations/'
+
+    aug_train_img_path = _prefix_path + '/train_set_aug/images/'
+    aug_train_annotation_path = _prefix_path + '/train_set_aug/annotations/'
+    aug_train_masked_img_path = _prefix_path + '/train_set_aug/masked_images/'
+
+    eval_img_path = _prefix_path + '/eval_set/images/'
+    eval_annotation_path = _prefix_path + '/eval_set/annotations/'
+    eval_masked_img_path = _prefix_path + '/eval_set/masked_images/'
+
+    '''7 labels'''
+    no_aug_train_img_path_7 = _prefix_path + '/train_set_7/images/'
+    no_aug_train_annotation_path_7 = _prefix_path + '/train_set_7/annotations/'
+
+    aug_train_img_path_7 = _prefix_path + '/train_set_7_aug/images/'
+    aug_train_annotation_path_7 = _prefix_path + '/train_set_7_aug/annotations/'
+    aug_train_masked_img_path_7 = _prefix_path + '/train_set_7_aug/masked_images/'
+
+    eval_img_path_7 = _prefix_path + '/eval_set_7/images/'
+    eval_annotation_path_7 = _prefix_path + '/eval_set_7/annotations/'
+    eval_masked_img_path_7 = _prefix_path + '/eval_set_7/masked_images/'
+
+    weight_save_path = _prefix_path + '/weight_saving_path/'
+
+    num_of_samples_train = 2420940
+    num_of_samples_train_7 = 0
+    num_of_samples_eval = 3999

custom_loss.py

@@ -0,0 +1,218 @@
+import tensorflow as tf
+import numpy as np
+from config import DatasetName, ExpressionCodesRafdb, ExpressionCodesAffectnet
+from keras import backend as K
+from sklearn.metrics import confusion_matrix
+import time
+from config import LearningConfig
+
+
+class CustomLosses:
+
+    def embedding_loss_distance(self, embeddings):
+
+        """
+        for each item in batch: calculate the correlation between all the embeddings
+        :param embeddings:
+        :return:
+        """
+        '''correlation'''
+        emb_len = len(embeddings)
+        ''' emb_num, bs, emb_size: '''
+        embeddings = tf.cast(embeddings, dtype=tf.dtypes.float32)
+        loss = tf.cast([np.corrcoef(embeddings[:, i, :]) for i in range(LearningConfig.batch_size)],
+                       dtype=tf.dtypes.float32)
+        embedding_similarity_loss = tf.reduce_mean((1 - tf.eye(emb_len)) *  # ignore the affect of the diagonal
+                                                   (1 + np.array(loss)))  # the loss -> more correlation, the better
+        return embedding_similarity_loss
+
+    def mean_embedding_loss_distance(self, embeddings, exp_gt_vec, exp_pr_vec, num_of_classes):
+        """
+        calculate the mean distribution for each class, and force the mean_embedding to be the same
+        :param embedding: bs * embedding_size
+        :param exp_gt_vec: bs
+        :param exp_pr_vec: bs * num_of_classes
+        :param num_of_classes:
+        :return:
+        """
+        # bs_size = tf.shape(exp_pr_vec, out_type=tf.dtypes.int64)[0]
+        #  7 * bs: for each class - which we have 7 class, not the embeddings -, is the sample belongs to it we put 1
+        # else we put 0 .  THE SAME FOR ALL embeddings
+        c_map = np.array([tf.cast(tf.where(exp_gt_vec == i, 1.0, K.epsilon()), dtype=tf.dtypes.float32)
+                          for i in range(num_of_classes)])  # 7 * bs
+
+        # calculate class-related mean embedding: num_of_classes * embedding_size
+        # 7 embedding, and 7 classes and each class has an embedding mean of 256 -> 7,7,256
+        mean_embeddings = np.array([[np.average(embeddings[k], axis=0, weights=c_map[i, :])
+                                     for i in range(num_of_classes)]
+                                    for k in range(len(embeddings))])  # 7:embedding,7:class, 256:size
+
+        #  the correlation between each mean_embedding should be low:
+        mean_emb_correlation_loss = tf.reduce_mean([(1 - tf.eye(num_of_classes)) *  # zero the diagonal
+                                                    (1 + tf.cast(np.corrcoef(mean_embeddings[k, :, :]),
+                                                                 dtype=tf.dtypes.float32))
+                                                    for k in range(len(embeddings))])
+
+        # the correlation between each mean_embedding should be low:
+        return mean_emb_correlation_loss
+
+    def mean_embedding_loss(self, embedding, exp_gt_vec, exp_pr_vec, num_of_classes):
+        """
+        calculate the mean distribution for each class, and force the mean_embedding to be the same
+        :param embedding: bs * embedding_size
+        :param exp_gt_vec: bs
+        :param exp_pr_vec: bs * num_of_classes
+        :param num_of_classes:
+        :return:
+        """
+        kl = tf.keras.losses.KLDivergence()
+        bs_size = tf.shape(exp_pr_vec, out_type=tf.dtypes.int64)[0]
+        # calculate class maps: num_of_classes * bs
+        c_map = np.array([tf.cast(tf.where(exp_gt_vec == i, 1, 0), dtype=tf.dtypes.int8)
+                          for i in range(num_of_classes)])  # 7 * bs
+        # calculate class-related mean embedding: num_of_classes * embedding_size
+        mean_embeddings = np.array([np.average(embedding, axis=0, weights=c_map[i, :])
+                                    if np.sum(c_map[i, :]) > 0 else np.zeros(LearningConfig.embedding_size)
+                                    for i in range(num_of_classes)]) + \
+                          K.epsilon()  # added as a bias to get ride of zeros
+
+        # calculate loss:
+        #   1 -> the correlation between each mean_embedding should be low:
+        mean_emb_correlation_loss = tf.reduce_mean((1 - tf.eye(num_of_classes)) *
+                                                   (1 + tf.cast(np.corrcoef(mean_embeddings), dtype=tf.dtypes.float32)))
+        #   2 -> the KL-divergence between the mean distribution of each class and the related
+        #   embeddings should be low. Accordingly, we lead the network towards learning the mean distribution
+        mean_emb_batch = tf.cast([np.array(mean_embeddings)[i] for i in np.argmax(c_map.T, axis=1)],
+                                 dtype=tf.dtypes.float32)
+        emb_kl_loss = kl(y_true=mean_emb_batch, y_pred=embedding)
+
+        return mean_emb_correlation_loss, emb_kl_loss
+
+    def variance_embedding_loss(self, embedding, exp_gt_vec, exp_pr_vec, num_of_classes):
+        """
+        calculate the variance of the distribution for each class, and force the mean_embedding to be the same
+        :param embedding:
+        :param exp_gt_vec:
+        :param exp_pr_vec:
+        :param num_of_classes:
+        :return:
+        """
+        kl = tf.keras.losses.KLDivergence()
+        bs_size = tf.shape(exp_pr_vec, out_type=tf.dtypes.int64)[0]
+        #
+        c_map = np.array([tf.cast(tf.where(exp_gt_vec == i, 1, 0), dtype=tf.dtypes.int8)
+                          for i in range(num_of_classes)])  # 7 * bs
+        # calculate class-related var embedding: num_of_classes * embedding_size
+        var_embeddings = np.array([tf.math.reduce_std(tf.math.multiply(embedding,
+                                                                       tf.repeat(tf.expand_dims(
+                                                                           tf.cast(c_map[i, :],
+                                                                                   dtype=tf.dtypes.float32), -1),
+                                                                           LearningConfig.embedding_size, axis=-1, )),
+                                                      axis=0)
+                                   for i in range(num_of_classes)]) \
+                         + K.epsilon()  # added as a bias to get ride of zeros
+
+        # calculate loss:
+        #   1 -> the correlation between each mean_embedding should be low:
+        var_emb_correlation_loss = tf.reduce_mean((1.0 - tf.eye(num_of_classes)) *
+                                                  (1.0 + tf.cast(np.cov(var_embeddings), dtype=tf.dtypes.float32)))
+        #   embeddings should be low. Accordingly, we lead the network towards learning the mean distribution
+        var_emb_batch = tf.cast([np.array(var_embeddings)[i] for i in np.argmax(c_map.T, axis=1)],
+                                dtype=tf.dtypes.float32)
+        emb_kl_loss = abs(kl(y_true=var_emb_batch, y_pred=embedding))
+        return var_emb_correlation_loss, emb_kl_loss
+
+    def correlation_loss(self, embedding, exp_gt_vec, exp_pr_vec, tr_conf_matrix):
+        bs_size = tf.shape(exp_pr_vec, out_type=tf.dtypes.int64)[0]
+        # convert from sigmoid to labels to real classes:
+        exp_pr = tf.constant([np.argmax(exp_pr_vec[i]) for i in range(bs_size)], dtype=tf.dtypes.int64)
+        # Cov matrix
+        phi_correlation_matrix = tf.cast(np.corrcoef(embedding), dtype=tf.dtypes.float32)  # bs * bs
+
+        elems_col = tf.repeat(tf.expand_dims(exp_gt_vec, 0), repeats=[bs_size], axis=0)
+        elems_row = tf.repeat(tf.expand_dims(exp_gt_vec, -1), repeats=[bs_size], axis=-1)
+        delta = elems_row - elems_col
+        omega_matrix = tf.cast(tf.where(delta == 0, 1, -1), dtype=tf.dtypes.float32)
+        # creating the adaptive weights
+        adaptive_weight = self._create_adaptive_correlation_weights(bs_size=bs_size,
+                                                                    exp_gt_vec=exp_gt_vec,  # real labels
+                                                                    exp_pr=exp_pr,  # real labels
+                                                                    conf_mat=tr_conf_matrix)
+        # calculate correlation loss
+        cor_loss = tf.reduce_mean(adaptive_weight * tf.abs(omega_matrix - phi_correlation_matrix))
+        return cor_loss
+
+    def correlation_loss_multi(self, embeddings, exp_gt_vec, exp_pr_vec, tr_conf_matrix):
+        """
+        here, we consider only one embedding and so want to make the embeddings of the same classes be similar
+        while the ones from different classes are different.
+        :param embeddings:
+        :param exp_gt_vec:
+        :param exp_pr_vec:
+        :param tr_conf_matrix:
+        :return:
+        """
+        bs_size = tf.shape(exp_pr_vec, out_type=tf.dtypes.int64)[0]
+        exp_pr = tf.constant([np.argmax(exp_pr_vec[i]) for i in range(bs_size)], dtype=tf.dtypes.int64)
+
+        phi_correlation_matrices = [tf.cast(np.corrcoef(embeddings[i]), dtype=tf.dtypes.float32)
+                                    for i in range(len(embeddings))]  # cls * bs * bs
+        #
+        elems_col = tf.repeat(tf.expand_dims(exp_gt_vec, 0), repeats=[bs_size], axis=0)
+        elems_row = tf.repeat(tf.expand_dims(exp_gt_vec, -1), repeats=[bs_size], axis=-1)
+        delta = elems_row - elems_col
+        omega_matrix = tf.repeat(tf.expand_dims(tf.cast(tf.where(delta == 0, 1, -1),
+                                                        dtype=tf.dtypes.float32), axis=0),
+                                 repeats=len(embeddings), axis=0)
+        cor_loss = tf.reduce_mean(tf.abs(omega_matrix - phi_correlation_matrices))
+
+        return cor_loss
+
+    def _create_adaptive_correlation_weights(self, bs_size, exp_gt_vec, exp_pr, conf_mat):
+        """
+        creating the weights
+        :param exp_gt_vec: real int labels
+        :param exp_pr_vec: one_hot labels
+        :param conf_mat: confusion matrix which is normalized over the rows(
+                        ground-truths with respect to the number of corresponding classes)
+        :return: a bath_size * bath_size matrix containing weights. The diameter of the matrix is zero
+        """
+        '''new'''
+        tf_identity = tf.eye(bs_size)
+        # weight based on the correct section of the conf_matrix
+        '''
+        1 : - conf_mat[exp_gt_vec[i], exp_gt_vec[i]] : sum of all the missed values=> the better the performance of
+        the model on a label, the smaller the weight  
+        '''
+        correct_row_base_weight = tf.repeat(tf.expand_dims(
+            tf.map_fn(fn=lambda i: 1 - conf_mat[i, i], elems=exp_gt_vec)  # map
+            , 0),  # expand_dims
+            repeats=[bs_size], axis=0)  # repeat
+
+        correct_col_base_weight = tf.einsum('ab->ba', correct_row_base_weight)
+        correct_weight = correct_row_base_weight + correct_col_base_weight
+        adaptive_weight = tf.cast((correct_weight), dtype=tf.dtypes.float32)
+        adaptive_weight = 1 + adaptive_weight  # we don't want the weights to be zero (correct prediction)
+        adaptive_weight = (1 - tf_identity) * adaptive_weight  # remove the main diagon
+        return adaptive_weight
+
+    def update_confusion_matrix(self, exp_gt_vec, exp_pr,
+                                all_gt_exp, all_pr_exp):
+        # adding to the previous predicted items:
+        all_pr_exp += np.array(exp_pr).tolist()
+        all_gt_exp += np.array(exp_gt_vec).tolist()
+        # calculate confusion matrix:
+        conf_mat = confusion_matrix(y_true=all_gt_exp, y_pred=all_pr_exp, normalize='true',
+                                    labels=[0, 1, 2, 3, 4, 5, 6])
+        return conf_mat, all_gt_exp, all_pr_exp
+
+    def cross_entropy_loss(self, y_gt, y_pr, num_classes, ds_name):
+        y_gt_oh = tf.one_hot(y_gt, depth=num_classes)
+        ''' manual weighted CE'''
+        y_pred = y_pr
+        y_pred /= tf.reduce_sum(y_pred, axis=-1, keepdims=True)
+        y_pred = K.clip(y_pred, K.epsilon(), 1)
+        loss = -tf.reduce_mean(y_gt_oh * tf.math.log(y_pred))
+        '''accuracy'''
+        accuracy = tf.reduce_mean(tf.keras.metrics.categorical_accuracy(y_pr, y_gt_oh))
+        return loss, accuracy

data_helper.py

@@ -0,0 +1,80 @@
+from config import DatasetName, AffectnetConf, InputDataSize, LearningConfig, ExpressionCodesAffectnet
+from config import LearningConfig, InputDataSize, DatasetName, AffectnetConf, DatasetType
+
+import numpy as np
+import os
+import matplotlib.pyplot as plt
+import math
+from datetime import datetime
+from sklearn.utils import shuffle
+from sklearn.model_selection import train_test_split
+from numpy import save, load, asarray
+import csv
+from skimage.io import imread
+import pickle
+import csv
+from tqdm import tqdm
+from PIL import Image
+from skimage.transform import resize
+from skimage import transform
+from skimage.transform import resize
+import tensorflow as tf
+import random
+import cv2
+from skimage.feature import hog
+from skimage import data, exposure
+from matplotlib.path import Path
+from scipy import ndimage, misc
+from skimage.transform import SimilarityTransform, AffineTransform
+from skimage.draw import rectangle
+from skimage.draw import line, set_color
+
+
+class DataHelper:
+
+    def create_generator_full_path(self, img_path, annotation_path, label=None):
+        img_filenames, exp_filenames = self._create_image_and_labels_name_full_path(img_path=img_path,
+                                                                                    annotation_path=annotation_path,
+                                                                                    label=label)
+        '''shuffle'''
+        img_filenames, exp_filenames = shuffle(img_filenames, exp_filenames)
+        return img_filenames, exp_filenames
+
+    def _create_image_and_labels_name_full_path(self, img_path, annotation_path, label):
+        img_filenames = []
+        exp_filenames = []
+
+        print('reading list -->')
+        file_names = tqdm(os.listdir(img_path))
+        print('<-')
+
+        for file in file_names:
+            if file.endswith(".jpg") or file.endswith(".png"):
+                exp_lbl_file = str(file)[:-4] + "_exp.npy"  # just name
+
+                if os.path.exists(annotation_path + exp_lbl_file):
+                    if label is not None:
+                        exp = np.load(annotation_path + exp_lbl_file)
+                        if label is not None and exp != label:
+                            continue
+
+                    img_filenames.append(img_path + str(file))
+                    exp_filenames.append(annotation_path + exp_lbl_file)
+
+        return np.array(img_filenames), np.array(exp_filenames)
+
+    def relabel_ds(self, labels):
+        new_labels = np.copy(labels)
+
+        index_src = [0, 1, 2, 3, 4, 5, 6, 7, 17, 18, 19, 20, 21, 31, 32, 36, 37, 38, 39, 40, 41, 48, 49, 50,
+                     60, 61, 67, 59, 58]
+        index_dst = [16, 15, 14, 13, 12, 11, 10, 9, 26, 25, 24, 23, 22, 35, 34, 45, 44, 43, 42, 47, 46, 54, 53, 52,
+                     64, 63, 65, 55, 56]
+
+        for i in range(len(index_src)):
+            new_labels[index_src[i] * 2] = labels[index_dst[i] * 2]
+            new_labels[index_src[i] * 2 + 1] = labels[index_dst[i] * 2 + 1]
+
+            new_labels[index_dst[i] * 2] = labels[index_src[i] * 2]
+            new_labels[index_dst[i] * 2 + 1] = labels[index_src[i] * 2 + 1]
+        return new_labels

dataset_class.py

@@ -0,0 +1,71 @@
+from config import DatasetName, AffectnetConf, InputDataSize, LearningConfig, ExpressionCodesAffectnet
+from config import LearningConfig, InputDataSize, DatasetName, AffectnetConf, DatasetType
+
+import numpy as np
+import os
+import matplotlib.pyplot as plt
+import math
+from datetime import datetime
+from sklearn.utils import shuffle
+from sklearn.model_selection import train_test_split
+from numpy import save, load, asarray
+import csv
+from skimage.io import imread
+import pickle
+import csv
+from tqdm import tqdm
+from PIL import Image
+from skimage.transform import resize
+from skimage import transform
+from skimage.transform import resize
+import tensorflow as tf
+import random
+import cv2
+from skimage.feature import hog
+from skimage import data, exposure
+from matplotlib.path import Path
+from scipy import ndimage, misc
+from skimage.transform import SimilarityTransform, AffineTransform
+from skimage.draw import rectangle
+from skimage.draw import line, set_color
+
+
+class CustomDataset:
+
+    def create_dataset(self, img_filenames, anno_names, is_validation=False, ds=DatasetName.affectnet):
+        def get_img(file_name):
+            path = bytes.decode(file_name)
+            image_raw = tf.io.read_file(path)
+            img = tf.image.decode_image(image_raw, channels=3)
+            img = tf.cast(img, tf.float32) / 255.0
+            '''augmentation'''
+            # if not (is_validation):# or tf.random.uniform([]) <= 0.5):
+            #     img = self._do_augment(img)
+            # ''''''
+            return img
+
+        def get_lbl(anno_name):
+            path = bytes.decode(anno_name)
+            lbl = load(path)
+            return lbl
+
+        def wrap_get_img(img_filename, anno_name):
+            img = tf.numpy_function(get_img, [img_filename], [tf.float32])
+            if is_validation and ds == DatasetName.affectnet:
+                lbl = tf.numpy_function(get_lbl, [anno_name], [tf.string])
+            else:
+                lbl = tf.numpy_function(get_lbl, [anno_name], [tf.int64])
+
+            return img, lbl
+
+        epoch_size = len(img_filenames)
+
+        img_filenames = tf.convert_to_tensor(img_filenames, dtype=tf.string)
+        anno_names = tf.convert_to_tensor(anno_names)
+
+        dataset = tf.data.Dataset.from_tensor_slices((img_filenames, anno_names))
+        dataset = dataset.shuffle(epoch_size)
+        dataset = dataset.map(wrap_get_img, num_parallel_calls=32) \
+            .batch(LearningConfig.batch_size, drop_remainder=True) \
+            .prefetch(10)
+        return dataset

main.py

@@ -0,0 +1,12 @@
+from test_model import TestModels
+from train import TrainModel
+from config import DatasetName, DatasetType
+if __name__ == "__main__":
+
+    '''testing the pre-trained models'''
+    tester = TestModels(h5_address='./trained_models/AffectNet_6336.h5')
+    tester.recognize_fer(img_path='./img.jpg')
+
+    '''training part'''
+    trainer = TrainModel(dataset_name=DatasetName.affectnet, ds_type=DatasetType.train_7)
+    trainer.train(arch="xcp", weight_path="./")

requirements.txt

@@ -0,0 +1,13 @@
+bottleneck
+tensorflow-gpu==2.4
+keras
+numpy
+matplotlib
+opencv-python
+opencv-contrib-python
+scipy
+scikit-learn
+scikit-image
+Pillow
+tqdm
+tensorboard

test_model.py

@@ -0,0 +1,47 @@
+
+import tensorflow as tf
+import tensorflow.keras as keras
+import numpy as np
+import matplotlib.pyplot as plt
+import math
+from datetime import datetime
+from numpy import save, load, asarray
+import csv
+from skimage.io import imread
+from skimage.transform import resize
+import pickle
+from PIL import Image
+
+import os
+
+# tf.test.is_gpu_available()
+
+class TestModels:
+    def __init__(self, h5_address:str, GPU=True):
+        self.exps = ['Neutral', 'Happy', 'Sad', 'Surprise', 'Fear', 'Disgust', 'Anger']
+
+        if GPU:
+            physical_devices = tf.config.list_physical_devices('GPU')
+            tf.config.experimental.set_memory_growth(physical_devices[0], True)
+
+        self.model = self.load_model(h5_address=h5_address)
+
+    def load_model(self, h5_address: str):
+        "load weight file and create the model once"
+        model = tf.keras.models.load_model(h5_address, custom_objects={'tf': tf})
+        return model
+
+    def recognize_fer(self, img_path:str):
+        "create and image from the path and recognize expression"
+        img = imread(img_path)
+        # resize img to 1*224*224*3
+        img = resize(img, (224, 224,3))
+        img = np.expand_dims(img, axis=0)
+        #
+        prediction = self.model.predict_on_batch([img])
+        exp = np.array(prediction[0])
+        '''in case you need the embeddings'''
+        # embeddings = prediction[1:]
+        print(self.exps[np.argmax(exp)])
+
+

train.py

@@ -0,0 +1,349 @@
+
+import tensorflow as tf
+import numpy as np
+import matplotlib.pyplot as plt
+import math
+from datetime import datetime
+from sklearn.utils import shuffle
+from sklearn.model_selection import train_test_split
+from numpy import save, load, asarray
+import csv
+from skimage.io import imread
+import pickle
+from sklearn.metrics import accuracy_score
+import os
+import time
+
+from AffectNetClass import AffectNet
+from RafdbClass import RafDB
+from FerPlusClass import FerPlus
+
+from config import DatasetName, AffectnetConf, InputDataSize, LearningConfig, DatasetType, RafDBConf, FerPlusConf
+from cnn_model import CNNModel
+from custom_loss import CustomLosses
+from data_helper import DataHelper
+from dataset_class import CustomDataset
+
+
+class TrainModel:
+    def __init__(self, dataset_name, ds_type, weights='imagenet', lr=1e-3, aug=True):
+        self.dataset_name = dataset_name
+        self.ds_type = ds_type
+        self.weights = weights
+        self.lr = lr
+
+        self.base_lr = 1e-5
+        self.max_lr = 5e-4
+        if dataset_name == DatasetName.fer2013:
+            self.drop = 0.1
+            self.epochs_drop = 5
+            if aug:
+                self.img_path = FerPlusConf.aug_train_img_path
+                self.annotation_path = FerPlusConf.aug_train_annotation_path
+                self.masked_img_path = FerPlusConf.aug_train_masked_img_path
+            else:
+                self.img_path = FerPlusConf.no_aug_train_img_path
+                self.annotation_path = FerPlusConf.no_aug_train_annotation_path
+
+            self.val_img_path = FerPlusConf.test_img_path
+            self.val_annotation_path = FerPlusConf.test_annotation_path
+            self.eval_masked_img_path = FerPlusConf.test_masked_img_path
+            self.num_of_classes = 7
+            self.num_of_samples = None
+
+        elif dataset_name == DatasetName.rafdb:
+            self.drop = 0.1
+            self.epochs_drop = 5
+
+            if aug:
+                self.img_path = RafDBConf.aug_train_img_path
+                self.annotation_path = RafDBConf.aug_train_annotation_path
+                self.masked_img_path = RafDBConf.aug_train_masked_img_path
+            else:
+                self.img_path = RafDBConf.no_aug_train_img_path
+                self.annotation_path = RafDBConf.no_aug_train_annotation_path
+
+            self.val_img_path = RafDBConf.test_img_path
+            self.val_annotation_path = RafDBConf.test_annotation_path
+            self.eval_masked_img_path = RafDBConf.test_masked_img_path
+            self.num_of_classes = 7
+            self.num_of_samples = None
+
+        elif dataset_name == DatasetName.affectnet:
+            self.drop = 0.1
+            self.epochs_drop = 5
+
+            if ds_type == DatasetType.train:
+                self.img_path = AffectnetConf.aug_train_img_path
+                self.annotation_path = AffectnetConf.aug_train_annotation_path
+                self.masked_img_path = AffectnetConf.aug_train_masked_img_path
+                self.val_img_path = AffectnetConf.eval_img_path
+                self.val_annotation_path = AffectnetConf.eval_annotation_path
+                self.eval_masked_img_path = AffectnetConf.eval_masked_img_path
+                self.num_of_classes = 8
+                self.num_of_samples = AffectnetConf.num_of_samples_train
+            elif ds_type == DatasetType.train_7:
+                if aug:
+                    self.img_path = AffectnetConf.aug_train_img_path_7
+                    self.annotation_path = AffectnetConf.aug_train_annotation_path_7
+                    self.masked_img_path = AffectnetConf.aug_train_masked_img_path_7
+                else:
+                    self.img_path = AffectnetConf.no_aug_train_img_path_7
+                    self.annotation_path = AffectnetConf.no_aug_train_annotation_path_7
+
+                self.val_img_path = AffectnetConf.eval_img_path_7
+                self.val_annotation_path = AffectnetConf.eval_annotation_path_7
+                self.eval_masked_img_path = AffectnetConf.eval_masked_img_path_7
+                self.num_of_classes = 7
+                self.num_of_samples = AffectnetConf.num_of_samples_train_7
+
+    def train(self, arch, weight_path):
+        """"""
+
+        '''create loss'''
+        c_loss = CustomLosses()
+
+        '''create summary writer'''
+        summary_writer = tf.summary.create_file_writer(
+            "./train_logs/fit/" + datetime.now().strftime("%Y%m%d-%H%M%S"))
+        start_train_date = datetime.now().strftime("%Y%m%d-%H%M%S")
+
+        '''making models'''
+        model = self.make_model(arch=arch, w_path=weight_path)
+        '''create save path'''
+        if self.dataset_name == DatasetName.affectnet:
+            save_path = AffectnetConf.weight_save_path + start_train_date + '/'
+        elif self.dataset_name == DatasetName.rafdb:
+            save_path = RafDBConf.weight_save_path + start_train_date + '/'
+        elif self.dataset_name == DatasetName.fer2013:
+            save_path = FerPlusConf.weight_save_path + start_train_date + '/'
+        if not os.path.exists(save_path):
+            os.makedirs(save_path)
+
+        '''create sample generator'''
+        dhp = DataHelper()
+
+        '''     Train   Generator'''
+        img_filenames, exp_filenames = dhp.create_generator_full_path(img_path=self.img_path,
+                                                                      annotation_path=self.annotation_path)
+        '''create dataset'''
+        cds = CustomDataset()
+        ds = cds.create_dataset(img_filenames=img_filenames,
+                                anno_names=exp_filenames,
+                                is_validation=False)
+
+        '''create train configuration'''
+        step_per_epoch = len(img_filenames) // LearningConfig.batch_size
+        gradients = None
+        virtual_step_per_epoch = LearningConfig.virtual_batch_size // LearningConfig.batch_size
+
+        '''create optimizer'''
+        optimizer = tf.keras.optimizers.Adam(self.lr, decay=1e-5)
+
+        '''start train:'''
+        all_gt_exp = []
+        all_pr_exp = []
+
+        for epoch in range(LearningConfig.epochs):
+            ce_weight = 2
+            batch_index = 0
+
+            for img_batch, exp_batch in ds:
+                '''since the calculation of the confusion matrix will be time-consuming,
+                    we only save 1000 labels each time. Moreover, this help us to be more qiuck on updates
+                    '''
+                all_gt_exp, all_pr_exp = self._update_all_labels_arrays(all_gt_exp, all_pr_exp)
+                '''load annotation and images'''
+                '''squeeze'''
+                exp_batch = exp_batch[:, -1]
+                img_batch = img_batch[:, -1, :, :]
+
+                '''train step'''
+                step_gradients, all_gt_exp, all_pr_exp = self.train_step(epoch=epoch, step=batch_index,
+                                                                         total_steps=step_per_epoch,
+                                                                         img_batch=img_batch,
+                                                                         anno_exp=exp_batch,
+                                                                         model=model, optimizer=optimizer,
+                                                                         c_loss=c_loss,
+                                                                         ce_weight=ce_weight,
+                                                                         summary_writer=summary_writer,
+                                                                         all_gt_exp=all_gt_exp,
+                                                                         all_pr_exp=all_pr_exp)
+                batch_index += 1
+
+            '''evaluating part'''
+            global_accuracy, conf_mat, avg_acc = self._eval_model(model=model)
+            '''save weights'''
+            save_name = save_path + '_' + str(epoch) + '_' + self.dataset_name + '_AC_' + str(global_accuracy)
+            model.save(save_name + '.h5')
+            self._save_confusion_matrix(conf_mat, save_name + '.txt')
+
+    def train_step(self, epoch, step, total_steps, model, ce_weight,
+                   img_batch, anno_exp, optimizer, summary_writer, c_loss, all_gt_exp, all_pr_exp):
+        with tf.GradientTape() as tape:
+            pr_data = model([img_batch], training=True)
+            exp_pr_vec = pr_data[0]
+            embeddings = pr_data[1:]
+
+            bs_size = tf.shape(exp_pr_vec, out_type=tf.dtypes.int64)[0]
+
+            loss_exp, accuracy = c_loss.cross_entropy_loss(y_pr=exp_pr_vec, y_gt=anno_exp,
+                                                           num_classes=self.num_of_classes,
+                                                           ds_name=self.dataset_name)
+
+            '''Feature difference loss'''
+            # embedding_similarity_loss = 0
+            embedding_similarity_loss = c_loss.embedding_loss_distance(embeddings=embeddings)
+
+            '''update confusion matrix'''
+            exp_pr = tf.constant([np.argmax(exp_pr_vec[i]) for i in range(bs_size)], dtype=tf.dtypes.int64)
+            tr_conf_matrix, all_gt_exp, all_pr_exp = c_loss.update_confusion_matrix(anno_exp,  # real labels
+                                                                                    exp_pr,  # real labels
+                                                                                    all_gt_exp,
+                                                                                    all_pr_exp)
+            ''' correlation between the embeddings'''
+            correlation_loss = c_loss.correlation_loss_multi(embeddings=embeddings,
+                                                             exp_gt_vec=anno_exp,
+                                                             exp_pr_vec=exp_pr_vec,
+                                                             tr_conf_matrix=tr_conf_matrix)
+            '''mean loss'''
+            mean_correlation_loss = c_loss.mean_embedding_loss_distance(embeddings=embeddings,
+                                                                        exp_gt_vec=anno_exp,
+                                                                        exp_pr_vec=exp_pr_vec,
+                                                                        num_of_classes=self.num_of_classes)
+
+            lamda_param = 50
+            loss_total = lamda_param * loss_exp + \
+                         embedding_similarity_loss + \
+                         correlation_loss + \
+                         mean_correlation_loss
+
+            # '''calculate gradient'''
+        gradients_of_model = tape.gradient(loss_total, model.trainable_variables)
+        # '''apply Gradients:'''
+        optimizer.apply_gradients(zip(gradients_of_model, model.trainable_variables))
+        # '''printing loss Values: '''
+        tf.print("->EPOCH: ", str(epoch), "->STEP: ", str(step) + '/' + str(total_steps),
+                 ' -> : accuracy: ', accuracy,
+                 ' -> : loss_total: ', loss_total,
+                 ' -> : loss_exp: ', loss_exp,
+                 ' -> : embedding_similarity_loss: ', embedding_similarity_loss,
+                 ' -> : correlation_loss: ', correlation_loss,
+                 ' -> : mean_correlation_loss: ', mean_correlation_loss)
+        with summary_writer.as_default():
+            tf.summary.scalar('loss_total', loss_total, step=epoch)
+            tf.summary.scalar('loss_exp', loss_exp, step=epoch)
+            tf.summary.scalar('correlation_loss', correlation_loss, step=epoch)
+            tf.summary.scalar('mean_correlation_loss', mean_correlation_loss, step=epoch)
+            tf.summary.scalar('embedding_similarity_loss', embedding_similarity_loss, step=epoch)
+        return gradients_of_model, all_gt_exp, all_pr_exp
+
+    def train_step_old(self, epoch, step, total_steps, model, ce_weight,
+                       img_batch, anno_exp, optimizer, summary_writer, c_loss, all_gt_exp, all_pr_exp):
+        with tf.GradientTape() as tape:
+            # '''create annotation_predicted'''
+            # exp_pr, embedding = model([img_batch], training=True)
+            exp_pr_vec, embedding_class, embedding_mean, embedding_var = model([img_batch], training=True)
+
+            bs_size = tf.shape(exp_pr_vec, out_type=tf.dtypes.int64)[0]
+            # # '''CE loss'''
+            loss_exp, accuracy = c_loss.cross_entropy_loss(y_pr=exp_pr_vec, y_gt=anno_exp,
+                                                           num_classes=self.num_of_classes,
+                                                           ds_name=self.dataset_name)
+            #
+            loss_cls_mean, loss_cls_var, loss_mean_var = c_loss.embedding_loss_distance(
+                embedding_class=embedding_class,
+                embedding_mean=embedding_mean,
+                embedding_var=embedding_var,
+                bs_size=bs_size)
+            feature_diff_loss = loss_cls_mean + loss_cls_var + loss_mean_var
+
+            # correlation between the class_embeddings
+            cor_loss, all_gt_exp, all_pr_exp = c_loss.correlation_loss(embedding=embedding_class,  # distribution
+                                                                       exp_gt_vec=anno_exp,
+                                                                       exp_pr_vec=exp_pr_vec,
+                                                                       num_of_classes=self.num_of_classes,
+                                                                       all_gt_exp=all_gt_exp,
+                                                                       all_pr_exp=all_pr_exp)
+            # correlation between the mean_emb_cor_loss
+            mean_emb_cor_loss, mean_emb_kl_loss = c_loss.mean_embedding_loss(embedding=embedding_mean,
+                                                                             exp_gt_vec=anno_exp,
+                                                                             exp_pr_vec=exp_pr_vec,
+                                                                             num_of_classes=self.num_of_classes)
+            mean_loss = mean_emb_cor_loss + 10 * mean_emb_kl_loss
+
+            var_emb_cor_loss, var_emb_kl_loss = c_loss.variance_embedding_loss(embedding=embedding_var,
+                                                                               exp_gt_vec=anno_exp,
+                                                                               exp_pr_vec=exp_pr_vec,
+                                                                               num_of_classes=self.num_of_classes)
+            var_loss = var_emb_cor_loss + 10 * var_emb_kl_loss
+            # '''total:'''
+            loss_total = 100 * loss_exp + cor_loss + 10 * feature_diff_loss + mean_loss + var_loss
+
+        # '''calculate gradient'''
+        gradients_of_model = tape.gradient(loss_total, model.trainable_variables)
+        # '''apply Gradients:'''
+        optimizer.apply_gradients(zip(gradients_of_model, model.trainable_variables))
+        # '''printing loss Values: '''
+        tf.print("->EPOCH: ", str(epoch), "->STEP: ", str(step) + '/' + str(total_steps),
+                 ' -> : accuracy: ', accuracy,
+                 ' -> : loss_total: ', loss_total,
+                 ' -> : loss_exp: ', loss_exp,
+                 ' -> : cor_loss: ', cor_loss,
+                 ' -> : feature_loss: ', feature_diff_loss,
+                 ' -> : mean_loss: ', mean_loss,
+                 ' -> : var_loss: ', var_loss)
+
+        with summary_writer.as_default():
+            tf.summary.scalar('loss_total', loss_total, step=epoch)
+            tf.summary.scalar('loss_exp', loss_exp, step=epoch)
+            tf.summary.scalar('loss_correlation', cor_loss, step=epoch)
+        return gradients_of_model, all_gt_exp, all_pr_exp
+
+    def _eval_model(self, model):
+        """"""
+        '''first we need to create the 4 bunch here: '''
+
+        '''for Affectnet, we need to calculate accuracy of each label and then total avg accuracy:'''
+        global_accuracy = 0
+        avg_acc = 0
+        conf_mat = []
+        if self.dataset_name == DatasetName.affectnet:
+            if self.ds_type == DatasetType.train:
+                affn = AffectNet(ds_type=DatasetType.eval)
+            else:
+                affn = AffectNet(ds_type=DatasetType.eval_7)
+            global_accuracy, conf_mat, avg_acc, precision, recall, fscore, support = \
+                affn.test_accuracy(model=model)
+        elif self.dataset_name == DatasetName.rafdb:
+            rafdb = RafDB(ds_type=DatasetType.test)
+            global_accuracy, conf_mat, avg_acc, precision, recall, fscore, support = rafdb.test_accuracy(model=model)
+        elif self.dataset_name == DatasetName.fer2013:
+            ferplus = FerPlus(ds_type=DatasetType.test)
+            global_accuracy, conf_mat, avg_acc, precision, recall, fscore, support = ferplus.test_accuracy(model=model)
+        print("================== global_accuracy =====================")
+        print(global_accuracy)
+        print("================== Average Accuracy =====================")
+        print(avg_acc)
+        print("================== Confusion Matrix =====================")
+        print(conf_mat)
+        return global_accuracy, conf_mat, avg_acc
+
+    def make_model(self, arch, w_path):
+        cnn = CNNModel()
+        model = cnn.get_model(arch=arch, num_of_classes=LearningConfig.num_classes, weights=self.weights)
+        if w_path is not None:
+            model.load_weights(w_path)
+        return model
+
+    def _save_confusion_matrix(self, conf_mat, save_name):
+        f = open(save_name, "a")
+        print(save_name)
+        f.write(np.array_str(conf_mat))
+        f.close()
+
+    def _update_all_labels_arrays(self, all_gt_exp, all_pr_exp):
+        if len(all_gt_exp) < LearningConfig.labels_history_frame:
+            return all_gt_exp, all_pr_exp
+        else:  # remove the first batch:
+            return all_gt_exp[LearningConfig.batch_size:], all_pr_exp[LearningConfig.batch_size:]

trained_models/AffectNet_6336.h5

@@ -0,0 +1,3 @@
+version https://git-lfs.github.com/spec/v1
+oid sha256:1f78e385a7c780a5f95e9f0772ec6ca18564b776875b1269d2858ad168df7956
+size 98617720

trained_models/Fer2013_7203.h5

@@ -0,0 +1,3 @@
+version https://git-lfs.github.com/spec/v1
+oid sha256:60caf5d2b95d84225c8197950d973ddd35523f50a577157f0f4885df2633d28d
+size 98617752

trained_models/RafDB_8696.h5

@@ -0,0 +1,3 @@
+version https://git-lfs.github.com/spec/v1
+oid sha256:5dfe5e381ccce39e922479aa4a99bdcf5460dab0b421d99cc1d76a79d4c6b4d9
+size 98617728