Summary

.flake8

@@ -0,0 +1,5 @@
+; https://www.flake8rules.com/
+
+[flake8]
+max-line-length = 120
+ignore = E203, E402, E741, W503

.gitignore

@@ -0,0 +1,172 @@
+# Compiled source #
+###################
+*.com
+*.class
+*.dll
+*.exe
+*.o
+*.so
+*.pyc
+
+# Packages #
+############
+# it's better to unpack these files and commit the raw source
+# git has its own built in compression methods
+*.7z
+*.dmg
+*.gz
+*.iso
+*.rar
+#*.tar
+*.zip
+
+# Logs and databases #
+######################
+*.log
+*.sqlite
+
+# OS generated files #
+######################
+.DS_Store
+ehthumbs.db
+Icon
+Thumbs.db
+.tmtags
+.idea
+.vscode
+tags
+vendor.tags
+tmtagsHistory
+*.sublime-project
+*.sublime-workspace
+.bundle
+
+# Byte-compiled / optimized / DLL files
+__pycache__/
+*.py[cod]
+*$py.class
+
+# C extensions
+*.so
+
+# Distribution / packaging
+.Python
+build/
+develop-eggs/
+dist/
+downloads/
+eggs/
+.eggs/
+lib/
+lib64/
+parts/
+sdist/
+var/
+wheels/
+pip-wheel-metadata/
+share/python-wheels/
+*.egg-info/
+.installed.cfg
+*.egg
+MANIFEST
+node_modules/
+
+# PyInstaller
+#  Usually these files are written by a python script from a template
+#  before PyInstaller builds the exe, so as to inject date/other infos into it.
+*.manifest
+*.spec
+
+# Installer logs
+pip-log.txt
+pip-delete-this-directory.txt
+
+# Unit test / coverage reports
+htmlcov/
+.tox/
+.nox/
+.coverage
+.coverage.*
+.cache
+nosetests.xml
+coverage.xml
+*.cover
+.hypothesis/
+.pytest_cache/
+
+# Translations
+*.mo
+*.pot
+
+# Django stuff:
+*.log
+local_settings.py
+db.sqlite3
+db.sqlite3-journal
+
+# Flask stuff:
+instance/
+.webassets-cache
+
+# Scrapy stuff:
+.scrapy
+
+# Sphinx documentation
+docs/_build/
+
+# PyBuilder
+target/
+
+# Jupyter Notebook
+.ipynb_checkpoints
+
+# IPython
+profile_default/
+ipython_config.py
+
+# pyenv
+.python-version
+
+# pipenv
+#   According to pypa/pipenv#598, it is recommended to include Pipfile.lock in version control.
+#   However, in case of collaboration, if having platform-specific dependencies or dependencies
+#   having no cross-platform support, pipenv may install dependencies that don't work, or not
+#   install all needed dependencies.
+#Pipfile.lock
+
+# celery beat schedule file
+celerybeat-schedule
+
+# SageMath parsed files
+*.sage.py
+
+# Environments
+.env
+.venv
+env/
+venv/
+ENV/
+env.bak/
+venv.bak/
+
+# Spyder project settings
+.spyderproject
+.spyproject
+
+# Rope project settings
+.ropeproject
+
+# mkdocs documentation
+/site
+
+# mypy
+.mypy_cache/
+.dmypy.json
+dmypy.json
+
+# Pyre type checker
+.pyre/
+
+# Custom
+*.pth
+*.pt

CODE_OF_CONDUCT.md

@@ -0,0 +1,80 @@
+# Code of Conduct
+
+## Our Pledge
+
+In the interest of fostering an open and welcoming environment, we as
+contributors and maintainers pledge to make participation in our project and
+our community a harassment-free experience for everyone, regardless of age, body
+size, disability, ethnicity, sex characteristics, gender identity and expression,
+level of experience, education, socio-economic status, nationality, personal
+appearance, race, religion, or sexual identity and orientation.
+
+## Our Standards
+
+Examples of behavior that contributes to creating a positive environment
+include:
+
+* Using welcoming and inclusive language
+* Being respectful of differing viewpoints and experiences
+* Gracefully accepting constructive criticism
+* Focusing on what is best for the community
+* Showing empathy towards other community members
+
+Examples of unacceptable behavior by participants include:
+
+* The use of sexualized language or imagery and unwelcome sexual attention or
+advances
+* Trolling, insulting/derogatory comments, and personal or political attacks
+* Public or private harassment
+* Publishing others' private information, such as a physical or electronic
+address, without explicit permission
+* Other conduct which could reasonably be considered inappropriate in a
+professional setting
+
+## Our Responsibilities
+
+Project maintainers are responsible for clarifying the standards of acceptable
+behavior and are expected to take appropriate and fair corrective action in
+response to any instances of unacceptable behavior.
+
+Project maintainers have the right and responsibility to remove, edit, or
+reject comments, commits, code, wiki edits, issues, and other contributions
+that are not aligned to this Code of Conduct, or to ban temporarily or
+permanently any contributor for other behaviors that they deem inappropriate,
+threatening, offensive, or harmful.
+
+## Scope
+
+This Code of Conduct applies within all project spaces, and it also applies when
+an individual is representing the project or its community in public spaces.
+Examples of representing a project or community include using an official
+project e-mail address, posting via an official social media account, or acting
+as an appointed representative at an online or offline event. Representation of
+a project may be further defined and clarified by project maintainers.
+
+This Code of Conduct also applies outside the project spaces when there is a
+reasonable belief that an individual's behavior may have a negative impact on
+the project or its community.
+
+## Enforcement
+
+Instances of abusive, harassing, or otherwise unacceptable behavior may be
+reported by contacting the project team at <[email protected]>. All
+complaints will be reviewed and investigated and will result in a response that
+is deemed necessary and appropriate to the circumstances. The project team is
+obligated to maintain confidentiality with regard to the reporter of an incident.
+Further details of specific enforcement policies may be posted separately.
+
+Project maintainers who do not follow or enforce the Code of Conduct in good
+faith may face temporary or permanent repercussions as determined by other
+members of the project's leadership.
+
+## Attribution
+
+This Code of Conduct is adapted from the [Contributor Covenant][homepage], version 1.4,
+available at <https://www.contributor-covenant.org/version/1/4/code-of-conduct.html>
+
+[homepage]: https://www.contributor-covenant.org
+
+For answers to common questions about this code of conduct, see
+<https://www.contributor-covenant.org/faq>

LICENSE

@@ -0,0 +1,21 @@
+MIT License
+
+Copyright (c) 2024 Elena Ryumina
+
+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,13 +1,14 @@
 ---
-title: AVCER
-emoji: 👀
-colorFrom: indigo
-colorTo: blue
+title: Audio-visual compound expression recognition
+emoji: 😀😲😐😥🥴😱😡
+colorFrom: blue
+colorTo: pink
 sdk: gradio
-sdk_version: 4.36.0
+sdk_version: 4.24.0
 app_file: app.py
 pinned: false
 license: mit
+short_description: A tool to detect audio-visual compound expressions
 ---
 
 Check out the configuration reference at https://huggingface.co/docs/hub/spaces-config-reference

app.css

@@ -0,0 +1,116 @@
+div.app-flex-container {
+    display: flex;
+    align-items: left;
+}
+
+div.app-flex-container > a {
+    margin-left: 6px;
+}
+
+div.dl1 div.upload-container {
+    height: 350px;
+    max-height: 350px;
+}
+
+div.dl2 {
+    max-height: 200px;
+}
+
+div.dl2 img {
+    max-height: 200px;
+}
+
+div.dl5 {
+    max-height: 200px;
+}
+
+div.dl5 img {
+    max-height: 200px;
+}
+
+div.video1 div.video-container {
+    height: 500px;
+}
+
+div.video2 {
+    height: 200px;
+}
+
+div.video3 {
+    height: 200px;
+}
+
+div.video4 {
+    height: 200px;
+}
+
+div.stat {
+    height: 350px;
+}
+
+div.audio {
+    height: 120px;
+}
+
+div.pred {
+    height: 65px;
+}
+
+div.video {
+    height: 65px;
+}
+
+div.img {
+    height: 120px;
+}
+
+div.settings-wrapper {
+    display: none;
+}
+
+.submit {
+    display: inline-block;
+    padding: 10px 20px;
+    font-size: 16px;
+    font-weight: bold;
+    text-align: center;
+    text-decoration: none;
+    cursor: pointer;
+    border: var(--button-border-width) solid var(--button-primary-border-color);
+    background: var(--button-primary-background-fill);
+    color: var(--button-primary-text-color);
+    border-radius: 8px;
+    transition: all 0.3s ease;
+}
+
+.submit[disabled] {
+    cursor: not-allowed;
+    opacity: 0.6;
+}
+
+.submit:hover:not([disabled]) {
+    border-color: var(--button-primary-border-color-hover);
+    background: var(--button-primary-background-fill-hover);
+    color: var(--button-primary-text-color-hover);
+}
+
+.clear {
+    display: inline-block;
+    padding: 10px 20px;
+    font-size: 16px;
+    font-weight: bold;
+    text-align: center;
+    text-decoration: none;
+    cursor: pointer;
+    border-radius: 8px;
+    transition: all 0.3s ease;
+}
+
+.clear[disabled] {
+    cursor: not-allowed;
+    opacity: 0.6;
+}
+
+.submit:active:not([disabled]), .clear:active:not([disabled]) {
+    transform: scale(0.98);
+}

app.py

@@ -0,0 +1,115 @@
+"""
+File: app.py
+Author: Elena Ryumina and Dmitry Ryumin
+Description: Description: Main application file for Facial_Expression_Recognition.
+             The file defines the Gradio interface, sets up the main blocks,
+             and includes event handlers for various components.
+License: MIT License
+"""
+
+import gradio as gr
+
+# Importing necessary components for the Gradio app
+from app.description import DESCRIPTION_DYNAMIC
+from app.authors import AUTHORS
+from app.app_utils import preprocess_video_and_predict
+
+
+def clear_static_info():
+    return (
+        gr.Image(value=None, type="pil"),
+        gr.Image(value=None, scale=1, elem_classes="dl5"),
+        gr.Image(value=None, scale=1, elem_classes="dl2"),
+        gr.Label(value=None, num_top_classes=3, scale=1, elem_classes="dl3"),
+    )
+
+def clear_dynamic_info():
+    return (
+        gr.Video(value=None),
+        gr.Plot(value=None),
+        gr.Plot(value=None),
+        gr.Plot(value=None),
+        gr.Textbox(value=None),
+        # gr.HTML(value=None),
+        gr.File(value=None),
+        gr.File(value=None),
+    )
+
+with gr.Blocks(css="app.css") as demo:
+    with gr.Tab("AVCER App"):
+        gr.Markdown(value=DESCRIPTION_DYNAMIC)
+        with gr.Row():
+            with gr.Column(scale=2):
+                input_video = gr.Video(elem_classes="video1")
+                with gr.Row():
+                    clear_btn_dynamic = gr.Button(
+                        value="Clear", interactive=True, scale=1
+                    )
+                    submit_dynamic = gr.Button(
+                        value="Submit", interactive=True, scale=1, elem_classes="submit"
+                    )
+                text = gr.Textbox(label="Result", info='Positive state includes Happiness, Surprise, Happily Surprised, and Happily Disgusted emotions. Negative state includes other emotions and Surprise.')
+                # question_mark = gr.HTML(tooltip_html)
+            with gr.Column(scale=2, elem_classes="dl4"):
+                output_face = gr.Plot(label="Face images", elem_classes="img")
+                output_heatmaps = gr.Plot(label="Waveform", elem_classes="audio")
+                output_statistics = gr.Plot(label="Statistics of emotions", elem_classes="stat")
+                with gr.Row():
+                    output_video = gr.File(label="Original video", 
+                                            file_count="single", 
+                                            file_types=[".mp4"],
+                                            show_label=True,
+                                            interactive=False,
+                                            visible=True, 
+                                            elem_classes="video")
+                    prediction_file = gr.File(label="Prediction file", 
+                                            file_count="single", 
+                                            file_types=[".csv"],
+                                            show_label=True,
+                                            interactive=False,
+                                            visible=True, 
+                                            elem_classes="pred")
+        gr.Examples(
+            ["videos/video1.mp4",
+            "videos/video2.mp4",
+            "videos/video3.mp4",
+            "videos/video4.mp4",
+            ],
+            [input_video],
+        )
+    
+    with gr.Tab("Authors"):
+        gr.Markdown(value=AUTHORS)
+
+    submit_dynamic.click(
+        fn=preprocess_video_and_predict,
+        inputs=input_video,
+        outputs=[
+            output_face,
+            output_heatmaps, 
+            output_statistics,
+            text,
+            # question_mark,
+            output_video,
+            prediction_file,
+        ],
+        queue=True,
+    )
+    clear_btn_dynamic.click(
+        fn=clear_dynamic_info,
+        inputs=[],
+        outputs=[
+            input_video,
+            output_face,
+            output_heatmaps, 
+            output_statistics,
+            text,
+            # question_mark,
+            output_video,
+            prediction_file,
+        ],
+        queue=True,
+    )
+
+if __name__ == "__main__":
+    demo.queue(api_open=False).launch(share=False)

app/app_utils.py

@@ -0,0 +1,296 @@
+"""
+File: app_utils.py
+Author: Elena Ryumina and Dmitry Ryumin
+Description: This module contains utility functions for facial expression recognition application.
+License: MIT License
+"""
+
+import torch
+import numpy as np
+import mediapipe as mp
+import pandas as pd
+from PIL import Image
+import cv2
+
+# Importing necessary components for the Gradio app
+from app.model import (
+    pth_model_static, 
+    pth_model_dynamic,  
+    activations,
+    audio_processor,
+    audio_model,
+    device
+)
+
+from app.utils import (
+    convert_mp4_to_mp3,
+    pad_wav, 
+    pad_wav_zeros,
+    get_box,
+    pth_processing, 
+    convert_webm_to_mp4,
+    get_evenly_spaced_frame_indices,
+    get_c_expr_db_pred
+)
+
+from app.config import DICT_EMO_VIDEO, AV_WEIGHTS, NAME_EMO_AUDIO, DICT_PRED, config_data
+from app.plot import display_frame_info, plot_images
+from collections import Counter
+
+mp_face_mesh = mp.solutions.face_mesh
+
+class EmotionRecognition:
+    def __init__(
+        self,
+        step=2,
+        window=4,
+        sr=16000,
+        save_path="",
+        padding="",
+    ):
+        self.save_path = save_path
+        self.step = step
+        self.window = window
+        self.sr = sr
+        self.padding = padding
+
+    def predict_emotion(self, path, frame_indices, fps):
+        prob, plt = self.load_audio_features(path, frame_indices, fps)
+        return prob, plt
+
+    def load_audio_features(self, path, frame_indices, fps):
+
+        window_a = self.window * self.sr
+        step_a = int(self.step * self.sr)
+
+        wav, audio_plt = convert_mp4_to_mp3(path, frame_indices, fps, self.sr)
+
+        probs = []
+        framess = []
+
+        for start_a in range(0, len(wav) + 1, step_a):
+            end_a = min(start_a + window_a, len(wav))
+            a_fss_chunk = wav[start_a:end_a]
+            if self.padding == "mean" or self.padding == "constant":
+                a_fss = pad_wav_zeros(a_fss_chunk, window_a, mode=self.padding)
+            elif self.padding == "repeat":
+                a_fss = pad_wav(a_fss_chunk, window_a)
+            a_fss = torch.unsqueeze(a_fss, 0)
+            a_fss = audio_processor(a_fss, sampling_rate=self.sr)
+            a_fss = a_fss["input_values"][0]
+            a_fss = torch.from_numpy(a_fss)
+            with torch.no_grad():
+                prob = audio_model(a_fss.to(device))
+            prob = prob.cpu().numpy()
+            frames = [
+                str(i).zfill(6) + ".jpg"
+                for i in range(
+                    round(start_a / self.sr * fps), round(end_a / self.sr * fps + 1)
+                )
+            ]
+            probs.extend([prob] * len(frames))
+            framess.extend(frames)
+
+        if len(probs[0]) == 7:
+            emo_ABAW = NAME_EMO_AUDIO[:-1]
+        else:
+            emo_ABAW = NAME_EMO_AUDIO
+
+        df = pd.DataFrame(np.array(probs), columns=emo_ABAW)
+        df["frames"] = framess
+
+        return df, audio_plt
+
+def preprocess_audio_and_predict(
+    path_video="",
+    save_path="src/pred_results/C-EXPR-DB",
+    frame_indices=[],
+    fps=25,
+    step=0.5,
+    padding="mean",
+    window=4,
+    sr=16000,
+):
+    audio_ER = EmotionRecognition(
+        step=step,
+        window=window,
+        sr=sr,
+        save_path=save_path,
+        padding=padding,
+    )
+    df_pred, audio_plt = audio_ER.predict_emotion(path_video, frame_indices, fps)
+
+    return df_pred, audio_plt
+
+def preprocess_video_and_predict(video):
+
+    if video:
+
+        if video.split('.')[-1] == 'webm':
+            video = convert_webm_to_mp4(video)
+
+        cap = cv2.VideoCapture(video)
+        w = int(cap.get(cv2.CAP_PROP_FRAME_WIDTH))
+        h = int(cap.get(cv2.CAP_PROP_FRAME_HEIGHT))
+        fps = np.round(cap.get(cv2.CAP_PROP_FPS))
+        total_frames = int(cap.get(cv2.CAP_PROP_FRAME_COUNT))
+
+        frame_indices = get_evenly_spaced_frame_indices(total_frames, 9)
+        df_probs_audio, audio_plt = preprocess_audio_and_predict(
+            path_video=video,
+            frame_indices=frame_indices,
+            fps=fps,
+            step=config_data.AUDIO_STEP,
+            padding="mean",
+            save_path="",
+            window=4,
+            sr=16000,
+        )
+
+        lstm_features = []
+        count_frame = 1
+        count_face = 0
+        probs_dynamic = []
+        probs_static = []
+        frames = []
+        last_output = None
+        cur_face = None
+        faces = []
+
+        zeros = np.zeros((1, 7))
+
+        with torch.no_grad():
+            with mp_face_mesh.FaceMesh(
+            max_num_faces=1,
+            refine_landmarks=False,
+            min_detection_confidence=0.5,
+            min_tracking_confidence=0.5) as face_mesh:
+
+                while cap.isOpened():
+                    _, frame = cap.read()
+                    if frame is None: break
+
+                    frame_copy = frame.copy()
+                    frame_copy.flags.writeable = False
+                    frame_copy = cv2.cvtColor(frame_copy, cv2.COLOR_BGR2RGB)
+                    results = face_mesh.process(frame_copy)
+                    frame_copy.flags.writeable = True
+
+                    if results.multi_face_landmarks:
+                        for fl in results.multi_face_landmarks:
+                            startX, startY, endX, endY  = get_box(fl, w, h)
+                            cur_face = frame_copy[startY:endY, startX: endX]
+
+                            if count_face%config_data.FRAME_DOWNSAMPLING == 0:
+                                cur_face_copy = pth_processing(Image.fromarray(cur_face))
+
+                                prediction = torch.nn.functional.softmax(pth_model_static(cur_face_copy.to(device)), dim=1)
+
+                                features = torch.nn.functional.relu(activations['features']).detach().cpu().numpy()
+
+                                output_s = prediction.clone()
+                                output_s = output_s.detach().cpu().numpy()
+
+                                if len(lstm_features) == 0:
+                                    lstm_features = [features]*10
+                                else:
+                                    lstm_features = lstm_features[1:] + [features]
+
+                                lstm_f = torch.from_numpy(np.vstack(lstm_features))
+                                lstm_f = torch.unsqueeze(lstm_f, 0)
+
+                                output_d = pth_model_dynamic(lstm_f.to(device)).detach().cpu().numpy()
+
+                                last_output = output_d
+
+                                if count_face == 0:
+                                    count_face += 1
+
+                            else:
+                                if last_output is not None:
+                                    output_d = last_output
+
+                                elif last_output is None:
+                                    output_d = zeros
+                                    
+                            probs_static.append(output_s[0])
+                            probs_dynamic.append(output_d[0])
+                            frames.append(count_frame)
+                    else:
+                        lstm_features = []
+                        if last_output is not None:
+                            probs_static.append(probs_static[-1])
+                            probs_dynamic.append(probs_dynamic[-1])
+                            frames.append(count_frame)
+
+                        elif last_output is None:
+                            probs_static.append(zeros[0])
+                            probs_dynamic.append(zeros[0])
+                            frames.append(count_frame)                       
+
+                    if cur_face is not None:
+
+                        if count_frame-1 in frame_indices:
+
+                            cur_face = cv2.resize(cur_face, (224,224), interpolation = cv2.INTER_AREA)
+                            cur_face = display_frame_info(cur_face, 'Frame: {}'.format(count_frame), box_scale=.3)
+                            faces.append(cur_face)
+
+                    count_frame += 1
+                    if count_face != 0:
+                        count_face += 1
+
+            img_plt = plot_images(faces)
+            
+            df_dynamic = pd.DataFrame(
+                np.array(probs_dynamic), columns=list(DICT_EMO_VIDEO.values())
+            )
+            df_static = pd.DataFrame(
+                np.array(probs_static), columns=list(DICT_EMO_VIDEO.values())
+            )
+
+            df, pred_plt = get_c_expr_db_pred(
+                stat_df=df_static,
+                dyn_df=df_dynamic,
+                audio_df=df_probs_audio,
+                name_video='',
+                weights_1=AV_WEIGHTS,
+                frame_indices=frame_indices,
+            )
+
+            av_pred = df['Audio-visual fusion'].tolist()
+
+            states = ['negative', 'neutral', 'positive']
+
+            dict_av_pred = Counter(av_pred)
+            count_states = np.zeros(3)
+            for k, v in dict_av_pred.items():
+                if k in [0]:
+                    count_states[1] += v
+                elif k in [4, 6, 8, 18]:
+                    count_states[2] += v
+                else:
+                    count_states[0] += v
+
+            state_percent = count_states/np.sum(count_states)
+
+            # if np.argmax(state_percent) in [0,2]:
+            #     text1 = "The audio-visual model predicts that a person mostly experiences {} ({:.2f}%) emotions. ".format(states[np.argmax(state_percent)], np.max(state_percent)*100)
+            # else:
+            text1 = "The audio-visual model predicts that a person is mostly in {} ({:.2f}%) state. ".format(states[np.argmax(state_percent)], np.max(state_percent)*100)
+
+            top_three = dict_av_pred.most_common(3)
+
+            top_three_text = "Predictions of the three most probable emotions: "
+            for index, count in top_three:
+                percentage = (count / np.sum(count_states)) * 100
+                top_three_text += f"{DICT_PRED[index]} ({percentage:.2f}%), "
+
+            top_three_text = top_three_text.rstrip(", ") + "."
+
+            df.to_csv(video.split('.')[0] + '.csv', index=False)
+                
+            return img_plt, audio_plt, pred_plt, text1+top_three_text, video, video.split('.')[0] + '.csv'
+            
+    else:
+        return None, None, None, None, None, None

app/authors.py

@@ -0,0 +1,31 @@
+"""
+File: authors.py
+Author: Elena Ryumina and Dmitry Ryumin
+Description: About the authors.
+License: MIT License
+"""
+
+
+AUTHORS = """
+    Authors: [Elena Ryumina](https://github.com/ElenaRyumina), [Maxim Markitantov](https://github.com/markitantov), [Dmitry Ryumin](https://github.com/DmitryRyumin), [Heysem Kaya](https://www.uu.nl/staff/HKaya) and [Alexey Karpov](https://hci.nw.ru/en/employees/1)
+
+    Authorship contribution:
+
+    App developers: ``Elena Ryumina`` and ``Dmitry Ryumin``
+
+    Methodology developers: ``Elena Ryumina``, ``Maxim Markitantov``, ``Dmitry Ryumin``, ``Heysem Kaya`` and ``Alexey Karpov``
+
+    Model developers: ``Elena Ryumina`` and ``Maxim Markitantov``
+
+    Citation
+
+    If you are using AVCER in your research, please site:
+
+    <div class="highlight highlight-text-bibtex notranslate position-relative overflow-auto" dir="auto"><pre><span class="pl-k">@inproceedings</span>{<span class="pl-en">RYUMINA2024CWPRV</span>,
+    <span class="pl-s">title</span>        = <span class="pl-s"><span class="pl-pds">{</span>Zero-Shot Audio-Visual Compound Expression Recognition Method based on Emotion Probability Fusion<span class="pl-pds">}</span></span>,
+    <span class="pl-s">author</span>       = <span class="pl-s"><span class="pl-pds">{</span>Elena Ryumina and Maxim Markitantov and Dmitry Ryumin and Heysem Kaya and Alexey Karpov<span class="pl-pds">}</span></span>,
+    <span class="pl-s">booktitle</span>    = <span class="pl-s"><span class="pl-pds">{</span>IEEE Computer Society Conference on Computer Vision and Pattern Recognition Workshops<span class="pl-pds">}</span></span>,
+    <span class="pl-s">year</span>         = <span class="pl-s"><span class="pl-pds">{</span>2024<span class="pl-pds">}</span></span>,
+    <span class="pl-s">pages</span>        = <span class="pl-s"><span class="pl-pds">{</span>1--9<span class="pl-pds">}</span></span>,
+    }</div>
+"""

app/config.py

@@ -0,0 +1,127 @@
+"""
+File: config.py
+Author: Elena Ryumina and Dmitry Ryumin
+Description: Configuration file.
+License: MIT License
+"""
+
+import toml
+from typing import Dict
+from types import SimpleNamespace
+
+
+def flatten_dict(prefix: str, d: Dict) -> Dict:
+    result = {}
+
+    for k, v in d.items():
+        if isinstance(v, dict):
+            result.update(flatten_dict(f"{prefix}{k}_", v))
+        else:
+            result[f"{prefix}{k}"] = v
+
+    return result
+
+
+config = toml.load("config.toml")
+
+config_data = flatten_dict("", config)
+
+config_data = SimpleNamespace(**config_data)
+
+DICT_EMO_VIDEO = {
+    0: "Neutral",
+    1: "Happiness",
+    2: "Sadness",
+    3: "Surprise",
+    4: "Fear",
+    5: "Disgust",
+    6: "Anger",
+}
+
+NAME_EMO_AUDIO = [
+        "Neutral",
+        "Anger",
+        "Disgust",
+        "Fear",
+        "Happiness",
+        "Sadness",
+        "Surprise",
+        "Other",
+    ]
+
+DICT_CE = {
+        "Fearfully Surprised": [3, 6],
+        "Happily Surprised": [4, 6],
+        "Sadly Surprised": [5, 6],
+        "Disgustedly Surprised": [2, 6],
+        "Angrily Surprised": [1, 6],
+        "Sadly Fearful": [3, 5],
+        "Sadly Angry": [1, 5],
+        "Sadly Disgusted": [2, 5],
+        "Fearfully Angry": [1, 3],
+        "Fearfully Disgusted": [2, 3],
+        "Angrily Disgusted": [1, 2],
+        "Happily Disgusted": [2, 4],
+    }
+
+DICT_PRED = {
+            0: 'Neutral',
+            1: 'Anger',
+            2: 'Disgust',
+            3: 'Fear',
+            4: 'Happiness',
+            5: 'Sadness',
+            6: 'Surprise',
+            7: 'Fearfully Surprised',
+            8: 'Happily Surprised',
+            9: 'Sadly Surprised',
+            10: 'Disgustedly Surprised',
+            11: 'Angrily Surprised',
+            12: 'Sadly Fearful',
+            13: 'Sadly Angry',
+            14: 'Sadly Disgusted',
+            15: 'Fearfully Angry',
+            16: 'Fearfully Disgusted',
+            17: 'Angrily Disgusted',
+            18: 'Happily Disgusted',
+    }
+
+AV_WEIGHTS = [
+        [
+            0.89900098,
+            0.10362151,
+            0.08577635,
+            0.04428126,
+            0.89679865,
+            0.02656456,
+            0.63040305,
+        ],
+        [
+            0.01223291,
+            0.21364307,
+            0.66688002,
+            0.93791526,
+            0.0398964,
+            0.48670648,
+            0.22089692,
+        ],
+        [
+            0.08876611,
+            0.68273542,
+            0.24734363,
+            0.01780348,
+            0.06330495,
+            0.48672896,
+            0.14870002,
+        ],
+    ]
+
+COLORS = {
+    0: 'blue', 
+    1: 'orange', 
+    2: 'green', 
+    3: 'red', 
+    4: 'purple', 
+    5: 'brown', 
+    6: 'pink'
+}

app/description.py

@@ -0,0 +1,20 @@
+"""
+File: description.py
+Author: Elena Ryumina and Dmitry Ryumin
+Description: Project description for the Gradio app.
+License: MIT License
+"""
+
+# Importing necessary components for the Gradio app
+from app.config import config_data
+
+DESCRIPTION_DYNAMIC = f"""\
+# Zero-Shot Audio-Visual Compound Expression Recognition (AVCER)
+
+AVCER predicts six basic emotions (Anger, Disgust, Fear, Happiness, Sadness, Surprise), neutral state (Neutral), and twelve compound emotions 
+(Fearfully Surprised, Happily Surprised, Sadly Surprised, Disgustedly Surprised, Angrily Surprised, Sadly Fearful, Sadly Angry, Sadly Disgusted, 
+Fearfully Angry, Fearfully Disgusted, Angrily Disgusted, Happily Disgusted).
+
+<div class="app-flex-container">
+    <img src="https://img.shields.io/badge/version-v{config_data.APP_VERSION}-rc0" alt="Version">
+"""

app/model.py

@@ -0,0 +1,72 @@
+"""
+File: model.py
+Author: Elena Ryumina and Dmitry Ryumin
+Description: This module provides functions for loading and processing a pre-trained deep learning model
+             for facial expression recognition.
+License: MIT License
+"""
+
+import torch
+import requests
+
+# Importing necessary components for the Gradio app
+from app.config import config_data
+from app.model_architectures import ResNet50, LSTMPyTorch, ExprModelV3
+from transformers import AutoFeatureExtractor
+
+device = "cuda" if torch.cuda.is_available() else "cpu"
+
+def load_model(model_url, model_path):
+    try:
+        with requests.get(model_url, stream=True) as response:
+            with open(model_path, "wb") as file:
+                for chunk in response.iter_content(chunk_size=8192):
+                    file.write(chunk)
+        return model_path
+    except Exception as e:
+        print(f"Error loading model: {e}")
+        return None
+    
+gradients = {}
+def get_gradients(name):
+    def hook(model, input, output):
+        gradients[name] = output
+    return hook
+
+activations = {}
+def get_activations(name):
+    def hook(model, input, output):
+        activations[name] = output.detach()
+    return hook
+
+test_static = torch.rand(1, 3, 224, 224)
+test_dynamic = torch.rand(1, 10, 512)
+test_audio = torch.rand(1, 64000)
+
+path_static = load_model(config_data.model_static_url, config_data.model_static_path)   
+pth_model_static = ResNet50(7, channels=3)
+pth_model_static.load_state_dict(torch.load(path_static))
+pth_model_static.to(device)
+pth_model_static.eval()
+pth_model_static(test_static.to(device))
+
+pth_model_static.layer4.register_full_backward_hook(get_gradients('layer4'))
+pth_model_static.layer4.register_forward_hook(get_activations('layer4'))
+pth_model_static.fc1.register_forward_hook(get_activations('features'))
+
+path_dynamic = load_model(config_data.model_dynamic_url, config_data.model_dynamic_path) 
+pth_model_dynamic = LSTMPyTorch()
+pth_model_dynamic.load_state_dict(torch.load(path_dynamic))
+pth_model_dynamic.to(device)
+pth_model_dynamic.eval()
+pth_model_dynamic(test_dynamic.to(device))
+
+path_audio_model_1 = "audeering/wav2vec2-large-robust-12-ft-emotion-msp-dim"
+path_audio_model_2 = load_model(config_data.model_audio_url, config_data.model_audio_path)  
+audio_processor = AutoFeatureExtractor.from_pretrained(path_audio_model_1)
+
+audio_model = ExprModelV3.from_pretrained(path_audio_model_1)
+audio_model.load_state_dict(torch.load(path_audio_model_2)["model_state_dict"])
+audio_model.to(device)
+audio_model.eval()
+audio_model(test_audio.to(device))

app/model_architectures.py

@@ -0,0 +1,483 @@
+"""
+File: model.py
+Author: Elena Ryumina and Dmitry Ryumin
+Description: This module provides model architectures.
+License: MIT License
+"""
+
+import torch
+import torch.nn as  nn
+import torch.nn.functional as F
+import math
+import numpy as np
+
+from transformers.models.wav2vec2.modeling_wav2vec2 import (
+    Wav2Vec2Model,
+    Wav2Vec2PreTrainedModel,
+)
+from typing import Optional
+
+
+class Bottleneck(nn.Module):
+    expansion = 4
+    def __init__(self, in_channels, out_channels, i_downsample=None, stride=1):
+        super(Bottleneck, self).__init__()
+        
+        self.conv1 = nn.Conv2d(in_channels, out_channels, kernel_size=1, stride=stride, padding=0, bias=False)
+        self.batch_norm1 = nn.BatchNorm2d(out_channels, eps=0.001, momentum=0.99)
+        
+        self.conv2 = nn.Conv2d(out_channels, out_channels, kernel_size=3, padding='same', bias=False)
+        self.batch_norm2 = nn.BatchNorm2d(out_channels, eps=0.001, momentum=0.99)
+        
+        self.conv3 = nn.Conv2d(out_channels, out_channels*self.expansion, kernel_size=1, stride=1, padding=0, bias=False)
+        self.batch_norm3 = nn.BatchNorm2d(out_channels*self.expansion, eps=0.001, momentum=0.99)
+        
+        self.i_downsample = i_downsample
+        self.stride = stride
+        self.relu = nn.ReLU()
+        
+    def forward(self, x):
+        identity = x.clone()
+        x = self.relu(self.batch_norm1(self.conv1(x)))
+        
+        x = self.relu(self.batch_norm2(self.conv2(x)))
+        
+        x = self.conv3(x)
+        x = self.batch_norm3(x)
+        
+        #downsample if needed
+        if self.i_downsample is not None:
+            identity = self.i_downsample(identity)
+        #add identity
+        x+=identity
+        x=self.relu(x)
+        
+        return x
+
+
+class Conv2dSame(torch.nn.Conv2d):
+
+    def calc_same_pad(self, i: int, k: int, s: int, d: int) -> int:
+        return max((math.ceil(i / s) - 1) * s + (k - 1) * d + 1 - i, 0)
+
+    def forward(self, x: torch.Tensor) -> torch.Tensor:
+        ih, iw = x.size()[-2:]
+
+        pad_h = self.calc_same_pad(i=ih, k=self.kernel_size[0], s=self.stride[0], d=self.dilation[0])
+        pad_w = self.calc_same_pad(i=iw, k=self.kernel_size[1], s=self.stride[1], d=self.dilation[1])
+
+        if pad_h > 0 or pad_w > 0:
+            x = F.pad(
+                x, [pad_w // 2, pad_w - pad_w // 2, pad_h // 2, pad_h - pad_h // 2]
+            )
+        return F.conv2d(
+            x,
+            self.weight,
+            self.bias,
+            self.stride,
+            self.padding,
+            self.dilation,
+            self.groups,
+        )
+
+
+class ResNet(nn.Module):
+    def __init__(self, ResBlock, layer_list, num_classes, num_channels=3):
+        super(ResNet, self).__init__()
+        self.in_channels = 64
+
+        self.conv_layer_s2_same = Conv2dSame(num_channels, 64, 7, stride=2, groups=1, bias=False)
+        self.batch_norm1 = nn.BatchNorm2d(64, eps=0.001, momentum=0.99)
+        self.relu = nn.ReLU()
+        self.max_pool = nn.MaxPool2d(kernel_size = 3, stride=2)
+        
+        self.layer1 = self._make_layer(ResBlock, layer_list[0], planes=64, stride=1)
+        self.layer2 = self._make_layer(ResBlock, layer_list[1], planes=128, stride=2)
+        self.layer3 = self._make_layer(ResBlock, layer_list[2], planes=256, stride=2)
+        self.layer4 = self._make_layer(ResBlock, layer_list[3], planes=512, stride=2)
+        
+        self.avgpool = nn.AdaptiveAvgPool2d((1,1))
+        self.fc1 = nn.Linear(512*ResBlock.expansion, 512)
+        self.relu1 = nn.ReLU()
+        self.fc2 = nn.Linear(512, num_classes)
+
+    def extract_features(self, x):
+        x = self.relu(self.batch_norm1(self.conv_layer_s2_same(x)))
+        x = self.max_pool(x)
+        # print(x.shape)
+        x = self.layer1(x)
+        x = self.layer2(x)
+        x = self.layer3(x)
+        x = self.layer4(x)
+        
+        x = self.avgpool(x)
+        x = x.reshape(x.shape[0], -1)
+        x = self.fc1(x)
+        return x
+        
+    def forward(self, x):
+        x = self.extract_features(x)
+        x = self.relu1(x)
+        x = self.fc2(x)
+        return x
+        
+    def _make_layer(self, ResBlock, blocks, planes, stride=1):
+        ii_downsample = None
+        layers = []
+        
+        if stride != 1 or self.in_channels != planes*ResBlock.expansion:
+            ii_downsample = nn.Sequential(
+                nn.Conv2d(self.in_channels, planes*ResBlock.expansion, kernel_size=1, stride=stride, bias=False, padding=0),
+                nn.BatchNorm2d(planes*ResBlock.expansion, eps=0.001, momentum=0.99)
+            )
+            
+        layers.append(ResBlock(self.in_channels, planes, i_downsample=ii_downsample, stride=stride))
+        self.in_channels = planes*ResBlock.expansion
+        
+        for i in range(blocks-1):
+            layers.append(ResBlock(self.in_channels, planes))
+            
+        return nn.Sequential(*layers)
+
+
+def ResNet50(num_classes, channels=3):
+    return ResNet(Bottleneck, [3,4,6,3], num_classes, channels)
+
+
+class LSTMPyTorch(nn.Module):
+    def __init__(self):
+        super(LSTMPyTorch, self).__init__()
+        
+        self.lstm1 = nn.LSTM(input_size=512, hidden_size=512, batch_first=True, bidirectional=False)
+        self.lstm2 = nn.LSTM(input_size=512, hidden_size=256, batch_first=True, bidirectional=False)
+        self.fc = nn.Linear(256, 7)
+        # self.softmax = nn.Softmax(dim=1)
+
+    def forward(self, x):
+        x, _ = self.lstm1(x)
+        x, _ = self.lstm2(x)        
+        x = self.fc(x[:, -1, :])
+        # x = self.softmax(x)
+        return x
+
+
+class ExprModelV3(Wav2Vec2PreTrainedModel):
+    def __init__(self, config) -> None:
+        super().__init__(config)
+        self.config = config
+        self.wav2vec2 = Wav2Vec2Model(config)
+
+        self.tl1 = TransformerLayer(
+            input_dim=1024, num_heads=32, dropout=0.1, positional_encoding=True
+        )
+        self.tl2 = TransformerLayer(
+            input_dim=1024, num_heads=16, dropout=0.1, positional_encoding=True
+        )
+
+        self.f_size = 1024
+
+        self.time_downsample = torch.nn.Sequential(
+            torch.nn.Conv1d(
+                self.f_size, self.f_size, kernel_size=5, stride=3, dilation=2
+            ),
+            torch.nn.BatchNorm1d(self.f_size),
+            torch.nn.MaxPool1d(5),
+            torch.nn.ReLU(),
+            torch.nn.Conv1d(self.f_size, self.f_size, kernel_size=3),
+            torch.nn.BatchNorm1d(self.f_size),
+            torch.nn.AdaptiveAvgPool1d(1),
+            torch.nn.ReLU(),
+        )
+
+        self.feature_downsample = nn.Linear(self.f_size, 8)
+
+        self.init_weights()
+        self.unfreeze_last_n_blocks(4)
+
+    def freeze_conv_only(self):
+        # freeze conv
+        for param in self.wav2vec2.feature_extractor.conv_layers.parameters():
+            param.requires_grad = False
+
+    def unfreeze_last_n_blocks(self, num_blocks: int) -> None:
+        # freeze all wav2vec
+        for param in self.wav2vec2.parameters():
+            param.requires_grad = False
+
+        # unfreeze last n transformer blocks
+        for i in range(0, num_blocks):
+            for param in self.wav2vec2.encoder.layers[-1 * (i + 1)].parameters():
+                param.requires_grad = True
+
+    def forward(self, x):
+        x = self.wav2vec2(x)[0]
+
+        x = self.tl1(query=x, key=x, value=x)
+        x = self.tl2(query=x, key=x, value=x)
+
+        x = x.permute(0, 2, 1)
+        x = self.time_downsample(x)
+
+        x = x.squeeze()
+        x = self.feature_downsample(x)
+        return x
+
+
+class ScaledDotProductAttention_MultiHead(nn.Module):
+
+    def __init__(self):
+        super(ScaledDotProductAttention_MultiHead, self).__init__()
+        self.softmax = nn.Softmax(dim=-1)
+
+    def forward(self, query, key, value, mask=None):
+        if mask is not None:
+            raise ValueError("Mask is not supported yet")
+
+        # key, query, value shapes: [batch_size, num_heads, seq_len, dim]
+        emb_dim = key.shape[-1]
+
+        # Calculate attention weights
+        attention_weights = torch.matmul(query, key.transpose(-2, -1)) / math.sqrt(
+            emb_dim
+        )
+
+        # masking
+        if mask is not None:
+            raise ValueError("Mask is not supported yet")
+
+        # Softmax
+        attention_weights = self.softmax(attention_weights)
+
+        # modify value
+        value = torch.matmul(attention_weights, value)
+
+        return value, attention_weights
+
+
+class PositionWiseFeedForward(nn.Module):
+
+    def __init__(self, input_dim, hidden_dim, dropout: float = 0.1):
+        super().__init__()
+        self.layer_1 = nn.Linear(input_dim, hidden_dim)
+        self.layer_2 = nn.Linear(hidden_dim, input_dim)
+        self.layer_norm = nn.LayerNorm(input_dim)
+        self.dropout = nn.Dropout(dropout)
+
+    def forward(self, x):
+        # feed-forward network
+        x = self.layer_1(x)
+        x = self.dropout(x)
+        x = F.relu(x)
+        x = self.layer_2(x)
+
+        return x
+
+
+class Add_and_Norm(nn.Module):
+
+    def __init__(self, input_dim, dropout: Optional[float] = 0.1):
+        super().__init__()
+        self.layer_norm = nn.LayerNorm(input_dim)
+        if dropout is not None:
+            self.dropout = nn.Dropout(dropout)
+
+    def forward(self, x1, residual):
+        x = x1
+        # apply dropout of needed
+        if hasattr(self, "dropout"):
+            x = self.dropout(x)
+        # add and then norm
+        x = x + residual
+        x = self.layer_norm(x)
+
+        return x
+
+
+class MultiHeadAttention(nn.Module):
+
+    def __init__(self, input_dim, num_heads, dropout: Optional[float] = 0.1):
+        super().__init__()
+        self.input_dim = input_dim
+        self.num_heads = num_heads
+        if input_dim % num_heads != 0:
+            raise ValueError("input_dim must be divisible by num_heads")
+        self.head_dim = input_dim // num_heads
+        self.dropout = dropout
+
+        # initialize weights
+        self.query_w = nn.Linear(input_dim, self.num_heads * self.head_dim, bias=False)
+        self.keys_w = nn.Linear(input_dim, self.num_heads * self.head_dim, bias=False)
+        self.values_w = nn.Linear(input_dim, self.num_heads * self.head_dim, bias=False)
+        self.ff_layer_after_concat = nn.Linear(
+            self.num_heads * self.head_dim, input_dim, bias=False
+        )
+
+        self.attention = ScaledDotProductAttention_MultiHead()
+
+        if self.dropout is not None:
+            self.dropout = nn.Dropout(dropout)
+
+    def forward(self, queries, keys, values, mask=None):
+        # query, keys, values shapes: [batch_size, seq_len, input_dim]
+        batch_size, len_query, len_keys, len_values = (
+            queries.size(0),
+            queries.size(1),
+            keys.size(1),
+            values.size(1),
+        )
+
+        # linear transformation before attention
+        queries = (
+            self.query_w(queries)
+            .view(batch_size, len_query, self.num_heads, self.head_dim)
+            .transpose(1, 2)
+        )  # [batch_size, num_heads, seq_len, dim]
+        keys = (
+            self.keys_w(keys)
+            .view(batch_size, len_keys, self.num_heads, self.head_dim)
+            .transpose(1, 2)
+        )  # [batch_size, num_heads, seq_len, dim]
+        values = (
+            self.values_w(values)
+            .view(batch_size, len_values, self.num_heads, self.head_dim)
+            .transpose(1, 2)
+        )  # [batch_size, num_heads, seq_len, dim]
+
+        # attention itself
+        values, attention_weights = self.attention(
+            queries, keys, values, mask=mask
+        )  # values shape:[batch_size, num_heads, seq_len, dim]
+
+        # concatenation
+        out = (
+            values.transpose(1, 2)
+            .contiguous()
+            .view(batch_size, len_values, self.num_heads * self.head_dim)
+        )  # [batch_size, seq_len, num_heads * dim = input_dim]
+        # go through last linear layer
+        out = self.ff_layer_after_concat(out)
+
+        return out
+
+
+class EncoderLayer(nn.Module):
+
+    def __init__(
+        self,
+        input_dim,
+        num_heads,
+        dropout: Optional[float] = 0.1,
+        positional_encoding: bool = True,
+    ):
+        super(EncoderLayer, self).__init__()
+        self.positional_encoding = positional_encoding
+        self.input_dim = input_dim
+        self.num_heads = num_heads
+        self.head_dim = input_dim // num_heads
+        self.dropout = dropout
+
+        # initialize layers
+        self.self_attention = MultiHeadAttention(input_dim, num_heads, dropout=dropout)
+        self.feed_forward = PositionWiseFeedForward(
+            input_dim, input_dim, dropout=dropout
+        )
+        self.add_norm_after_attention = Add_and_Norm(input_dim, dropout=dropout)
+        self.add_norm_after_ff = Add_and_Norm(input_dim, dropout=dropout)
+
+        # calculate positional encoding
+        if self.positional_encoding:
+            self.positional_encoding = PositionalEncoding(input_dim)
+
+    def forward(self, x):
+        # x shape: [batch_size, seq_len, input_dim]
+        # positional encoding
+        if self.positional_encoding:
+            x = self.positional_encoding(x)
+
+        # multi-head attention
+        residual = x
+        x = self.self_attention(x, x, x)
+        x = self.add_norm_after_attention(x, residual)
+
+        # feed forward
+        residual = x
+        x = self.feed_forward(x)
+        x = self.add_norm_after_ff(x, residual)
+
+        return x
+
+
+class PositionalEncoding(nn.Module):
+
+    def __init__(self, d_model: int, dropout: float = 0.1, max_len: int = 5000):
+        super().__init__()
+        self.dropout = nn.Dropout(p=dropout)
+
+        position = torch.arange(max_len).unsqueeze(1)
+        div_term = torch.exp(
+            torch.arange(0, d_model, 2) * (-math.log(10000.0) / d_model)
+        )
+        pe = torch.zeros(max_len, 1, d_model)
+        pe[:, 0, 0::2] = torch.sin(position * div_term)
+        pe[:, 0, 1::2] = torch.cos(position * div_term)
+        pe = pe.permute(
+            1, 0, 2
+        )  # [seq_len, batch_size, embedding_dim] -> [batch_size, seq_len, embedding_dim]
+        self.register_buffer("pe", pe)
+
+    def forward(self, x: torch.Tensor) -> torch.Tensor:
+        """
+        Args:
+            x: Tensor, shape [batch_size, seq_len, embedding_dim]
+        """
+        x = x + self.pe[:, : x.size(1)]
+        return self.dropout(x)
+
+
+class TransformerLayer(nn.Module):
+
+    def __init__(
+        self,
+        input_dim,
+        num_heads,
+        dropout: Optional[float] = 0.1,
+        positional_encoding: bool = True,
+    ):
+        super(TransformerLayer, self).__init__()
+        self.positional_encoding = positional_encoding
+        self.input_dim = input_dim
+        self.num_heads = num_heads
+        self.head_dim = input_dim // num_heads
+        self.dropout = dropout
+
+        # initialize layers
+        self.self_attention = MultiHeadAttention(input_dim, num_heads, dropout=dropout)
+        self.feed_forward = PositionWiseFeedForward(
+            input_dim, input_dim, dropout=dropout
+        )
+        self.add_norm_after_attention = Add_and_Norm(input_dim, dropout=dropout)
+        self.add_norm_after_ff = Add_and_Norm(input_dim, dropout=dropout)
+
+        # calculate positional encoding
+        if self.positional_encoding:
+            self.positional_encoding = PositionalEncoding(input_dim)
+
+    def forward(self, key, value, query, mask=None):
+        # key, value, and query shapes: [batch_size, seq_len, input_dim]
+        # positional encoding
+        if self.positional_encoding:
+            key = self.positional_encoding(key)
+            value = self.positional_encoding(value)
+            query = self.positional_encoding(query)
+
+        # multi-head attention
+        residual = query
+        x = self.self_attention(queries=query, keys=key, values=value, mask=mask)
+        x = self.add_norm_after_attention(x, residual)
+
+        # feed forward
+        residual = x
+        x = self.feed_forward(x)
+        x = self.add_norm_after_ff(x, residual)
+
+        return x

app/plot.py

@@ -0,0 +1,177 @@
+"""
+File: config.py
+Author: Elena Ryumina and Dmitry Ryumin
+Description: Plotting statistical information.
+License: MIT License
+"""
+import matplotlib.pyplot as plt
+import numpy as np
+import cv2
+import torch
+
+# Importing necessary components for the Gradio app
+from app.config import DICT_PRED
+
+def show_cam_on_image(
+    img: np.ndarray,
+    mask: np.ndarray,
+    use_rgb: bool = False,
+    colormap: int = cv2.COLORMAP_JET,
+    image_weight: float = 0.5,
+) -> np.ndarray:
+    """This function overlays the cam mask on the image as an heatmap.
+    By default the heatmap is in BGR format.
+
+    :param img: The base image in RGB or BGR format.
+    :param mask: The cam mask.
+    :param use_rgb: Whether to use an RGB or BGR heatmap, this should be set to True if 'img' is in RGB format.
+    :param colormap: The OpenCV colormap to be used.
+    :param image_weight: The final result is image_weight * img + (1-image_weight) * mask.
+    :returns: The default image with the cam overlay.
+
+    Implemented by https://github.com/jacobgil/pytorch-grad-cam/blob/master/pytorch_grad_cam/utils/image.py
+    """
+    heatmap = cv2.applyColorMap(np.uint8(255 * mask), colormap)
+    if use_rgb:
+        heatmap = cv2.cvtColor(heatmap, cv2.COLOR_BGR2RGB)
+    heatmap = np.float32(heatmap) / 255
+
+    if np.max(img) > 1:
+        raise Exception("The input image should np.float32 in the range [0, 1]")
+
+    if image_weight < 0 or image_weight > 1:
+        raise Exception(
+            f"image_weight should be in the range [0, 1].\
+                Got: {image_weight}"
+        )
+
+    cam = (1 - image_weight) * heatmap + image_weight * img
+    cam = cam / np.max(cam)
+    return np.uint8(255 * cam)
+
+
+def get_heatmaps(
+    gradients, activations, name_layer, face_image, use_rgb=True, image_weight=0.6
+):
+    gradient = gradients[name_layer]
+    activation = activations[name_layer]
+    pooled_gradients = torch.mean(gradient[0], dim=[0, 2, 3])
+    for i in range(activation.size()[1]):
+        activation[:, i, :, :] *= pooled_gradients[i]
+    heatmap = torch.mean(activation, dim=1).squeeze().cpu()
+    heatmap = np.maximum(heatmap, 0)
+    heatmap /= torch.max(heatmap)
+    heatmap = torch.unsqueeze(heatmap, -1)
+    heatmap = cv2.resize(heatmap.detach().numpy(), (224, 224))
+    cur_face_hm = cv2.resize(face_image, (224, 224))
+    cur_face_hm = np.float32(cur_face_hm) / 255
+
+    heatmap = show_cam_on_image(
+        cur_face_hm, heatmap, use_rgb=use_rgb, image_weight=image_weight
+    )
+
+    return heatmap
+
+def plot_compound_expression_prediction(
+    dict_preds: dict[str, list[float]],
+    save_path: str = None,
+    frame_indices: list[int] = None,
+    colors: list[str] = ["green", "orange", "red", "purple", "blue"],
+    figsize: tuple = (12, 6),
+    title: str = "Confusion Matrix",
+) -> plt.Figure:
+    fig, ax = plt.subplots(figsize=figsize)
+
+    for idx, (k, v) in enumerate(dict_preds.items()):
+        if idx == 2:
+            offset = (idx+1 - len(dict_preds) // 2) * 0.1
+        elif idx == 3:
+            offset = (idx-1 - len(dict_preds) // 2) * 0.1
+        else:
+            offset = (idx - len(dict_preds) // 2) * 0.1
+        shifted_v = [val + offset + 1 for val in v]
+        ax.plot(range(1, len(shifted_v) + 1), shifted_v, color=colors[idx], linestyle='dotted', label=k)
+
+    ax.legend()
+    ax.grid(True)
+    ax.set_xlabel("Number of frames")
+    ax.set_ylabel("Basic emotion / compound expression")
+    ax.set_title(title)
+
+    ax.set_xticks([i+1 for i in frame_indices])
+    ax.set_yticks(
+        range(0, 21)
+    )
+    ax.set_yticklabels([''] + list(DICT_PRED.values()) + [''])
+
+    fig.tight_layout()
+
+    if save_path:
+        fig.savefig(
+            save_path,
+            format=save_path.rsplit(".", 1)[1],
+            bbox_inches="tight",
+            pad_inches=0,
+        )
+
+    return fig
+
+def display_frame_info(img, text, margin=1.0, box_scale=1.0):
+    img_copy = img.copy()
+    img_h, img_w, _ = img_copy.shape
+    line_width = int(min(img_h, img_w) * 0.001)
+    thickness = max(int(line_width / 3), 1)
+
+    font_face = cv2.FONT_HERSHEY_SIMPLEX
+    font_color = (0, 0, 0)
+    font_scale = thickness / 1.5
+
+    t_w, t_h = cv2.getTextSize(text, font_face, font_scale, None)[0]
+
+    margin_n = int(t_h * margin)
+    sub_img = img_copy[0 + margin_n: 0 + margin_n + t_h + int(2 * t_h * box_scale),
+              img_w - t_w - margin_n - int(2 * t_h * box_scale): img_w - margin_n]
+
+    white_rect = np.ones(sub_img.shape, dtype=np.uint8) * 255
+
+    img_copy[0 + margin_n: 0 + margin_n + t_h + int(2 * t_h * box_scale),
+    img_w - t_w - margin_n - int(2 * t_h * box_scale):img_w - margin_n] = cv2.addWeighted(sub_img, 0.5, white_rect, .5, 1.0)
+
+    cv2.putText(img=img_copy,
+                text=text,
+                org=(img_w - t_w - margin_n - int(2 * t_h * box_scale) // 2,
+                     0 + margin_n + t_h + int(2 * t_h * box_scale) // 2),
+                fontFace=font_face,
+                fontScale=font_scale,
+                color=font_color,
+                thickness=thickness,
+                lineType=cv2.LINE_AA,
+                bottomLeftOrigin=False)
+
+    return img_copy
+
+def plot_audio(time_axis, waveform, frame_indices, fps, figsize=(10, 4)) -> plt.Figure:
+    frame_times = np.array(frame_indices) / fps
+
+    fig, ax = plt.subplots(figsize=figsize)
+    ax.plot(time_axis, waveform[0])
+    ax.set_xlabel('Time (frames)')
+    ax.set_ylabel('Amplitude')
+    ax.grid(True)
+
+    ax.set_xticks(frame_times)
+    ax.set_xticklabels([f'{int(frame_time*fps)+1}' for frame_time in frame_times])
+
+    fig.tight_layout()
+
+    return fig
+
+def plot_images(image_paths):
+    fig, axes = plt.subplots(1, len(image_paths), figsize=(12, 2))
+
+    for ax, img_path in zip(axes, image_paths):
+        ax.imshow(img_path)
+        ax.axis('off')
+
+    fig.tight_layout()
+    return fig

app/utils.py

@@ -0,0 +1,273 @@
+"""
+File: face_utils.py
+Author: Elena Ryumina and Dmitry Ryumin
+Description: This module contains utility functions related to facial landmarks and image processing.
+License: MIT License
+"""
+
+import numpy as np
+import pandas as pd
+import math
+
+import subprocess
+import torchaudio
+import torch
+import os
+
+from PIL import Image
+from torchvision import transforms
+
+# Importing necessary components for the Gradio app
+from app.config import NAME_EMO_AUDIO, DICT_CE, config_data
+from app.plot import plot_compound_expression_prediction, plot_audio
+
+
+def norm_coordinates(normalized_x, normalized_y, image_width, image_height):
+    x_px = min(math.floor(normalized_x * image_width), image_width - 1)
+    y_px = min(math.floor(normalized_y * image_height), image_height - 1)
+    return x_px, y_px
+
+
+def get_box(fl, w, h):
+    idx_to_coors = {}
+    for idx, landmark in enumerate(fl.landmark):
+        landmark_px = norm_coordinates(landmark.x, landmark.y, w, h)
+        if landmark_px:
+            idx_to_coors[idx] = landmark_px
+
+    x_min = np.min(np.asarray(list(idx_to_coors.values()))[:, 0])
+    y_min = np.min(np.asarray(list(idx_to_coors.values()))[:, 1])
+    endX = np.max(np.asarray(list(idx_to_coors.values()))[:, 0])
+    endY = np.max(np.asarray(list(idx_to_coors.values()))[:, 1])
+
+    (startX, startY) = (max(0, x_min), max(0, y_min))
+    (endX, endY) = (min(w - 1, endX), min(h - 1, endY))
+
+    return startX, startY, endX, endY
+
+
+def pth_processing(fp):
+    class PreprocessInput(torch.nn.Module):
+        def init(self):
+            super(PreprocessInput, self).init()
+
+        def forward(self, x):
+            x = x.to(torch.float32)
+            x = torch.flip(x, dims=(0,))
+            x[0, :, :] -= 91.4953
+            x[1, :, :] -= 103.8827
+            x[2, :, :] -= 131.0912
+            return x
+
+    def get_img_torch(img, target_size=(224, 224)):
+        transform = transforms.Compose([transforms.PILToTensor(), PreprocessInput()])
+        img = img.resize(target_size, Image.Resampling.NEAREST)
+        img = transform(img)
+        img = torch.unsqueeze(img, 0)
+        return img
+
+    return get_img_torch(fp)
+
+def convert_webm_to_mp4(input_file):
+
+    path_save = input_file.split('.')[0] + ".mp4"
+
+    if not os.path.exists(path_save):
+        ff_video = "ffmpeg -i {} -c:v copy -c:a aac -strict experimental {}".format(
+            input_file, path_save
+        )
+        subprocess.call(ff_video, shell=True)
+
+    return path_save
+
+def convert_mp4_to_mp3(path, frame_indices, fps, sampling_rate=16000):
+
+    path_save = path.split('.')[0] + ".wav"
+    if not os.path.exists(path_save):
+        ff_audio = "ffmpeg -i {} -vn -acodec pcm_s16le -ar 44100 -ac 2 {}".format(
+            path, path_save
+        )
+        subprocess.call(ff_audio, shell=True)
+    wav, sr = torchaudio.load(path_save)
+
+    num_frames = wav.numpy().shape[1]
+    time_axis = [i / sr for i in range(num_frames)]
+
+    plt = plot_audio(time_axis, wav, frame_indices, fps, (12, 2))
+
+    if wav.size(0) > 1:
+        wav = wav.mean(dim=0, keepdim=True)
+
+    if sr != sampling_rate:
+        transform = torchaudio.transforms.Resample(orig_freq=sr, new_freq=sampling_rate)
+        wav = transform(wav)
+        sr = sampling_rate
+
+    assert sr == sampling_rate
+    return wav.squeeze(0), plt
+
+
+def pad_wav(wav, max_length):
+    current_length = len(wav)
+    if current_length < max_length:
+        repetitions = (max_length + current_length - 1) // current_length
+        wav = torch.cat([wav] * repetitions, dim=0)[:max_length]
+    elif current_length > max_length:
+        wav = wav[:max_length]
+
+    return wav
+
+
+def pad_wav_zeros(wav, max_length, mode="constant"):
+
+    if mode == "mean":
+        wav = torch.nn.functional.pad(
+            wav,
+            (0, max(0, max_length - wav.shape[0])),
+            mode="constant",
+            value=torch.mean(wav),
+        )
+
+    else:
+        wav = torch.nn.functional.pad(
+            wav, (0, max(0, max_length - wav.shape[0])), mode=mode
+        )
+
+    return wav
+
+def softmax(matrix):
+    exp_matrix = np.exp(matrix - np.max(matrix, axis=1, keepdims=True))
+    return exp_matrix / np.sum(exp_matrix, axis=1, keepdims=True)
+
+
+def get_compound_expression(pred, com_emo):
+    pred = np.asarray(pred)
+    prob = np.zeros((len(pred), len(com_emo)))
+    for idx, (_, v) in enumerate(com_emo.items()):
+        idx_1 = v[0]
+        idx_2 = v[1]
+        prob[:, idx] = pred[:, idx_1] + pred[:, idx_2]
+    return prob
+
+
+def get_image_location(curr_video, frame):
+    frame = int(frame.split(".")[0]) + 1
+    frame = str(frame).zfill(5) + ".jpg"
+    return f"{curr_video}/{frame}"
+
+
+def save_txt(column_names, file_names, labels, save_name):
+    data_lines = [",".join(column_names)]
+    for file_name, label in zip(file_names, labels):
+        data_lines.append(f"{file_name},{label}")
+
+    with open(save_name, "w") as file:
+        for line in data_lines:
+            file.write(line + "\n")
+    
+def get_mix_pred(emo_pred, ce_prob):
+    pred = []
+    for idx, curr_pred in enumerate(emo_pred):
+        if np.max(curr_pred) > config_data.CONFIDENCE_BE:
+            pred.append(np.argmax(curr_pred))
+        else:
+            pred.append(ce_prob[idx]+6)
+    return pred
+
+def get_c_expr_db_pred(
+    stat_df: pd.DataFrame,
+    dyn_df: pd.DataFrame,
+    audio_df: pd.DataFrame,
+    name_video: str,
+    weights_1: list[float],
+    frame_indices: list[int],
+) -> tuple[np.ndarray, np.ndarray, np.ndarray, np.ndarray, list[str]]:
+    """
+    Predict compound expressions using audio-visual emotional probabilities, optimized weights, and rules.
+
+    Args:
+        stat_df (pd.DataFrame): DataFrame containing static visual probabilities.
+        dyn_df (pd.DataFrame): DataFrame containing dynamic visual probabilities.
+        audio_df (pd.DataFrame): DataFrame containing audio probabilities.
+        name_video (str): Name of the video.
+        weights_1 (List[float]): List of weights for the Dirichlet-based fusion.
+
+    Returns:
+        Tuple[np.ndarray, np.ndarray, np.ndarray, np.ndarray, List[str]]: Predictions for compound expressions,
+            and list of image locations.
+    """
+
+    stat_df["image_location"] = [
+        f"{name_video}/{str(f+1).zfill(5)}.jpg" for f in stat_df.index
+    ]
+    dyn_df["image_location"] = [
+        f"{name_video}/{str(f+1).zfill(5)}.jpg" for f in dyn_df.index
+    ]
+
+    image_location = dyn_df.image_location.tolist()
+
+    stat_df = stat_df[stat_df.image_location.isin(image_location)][NAME_EMO_AUDIO[:-1]].values
+    dyn_df = softmax(
+        dyn_df[dyn_df.image_location.isin(image_location)][NAME_EMO_AUDIO[:-1]].values
+    )
+
+    audio_df = audio_df.groupby(["frames"]).mean().reset_index()
+    audio_df = audio_df.rename(columns={"frames": "image_location"})
+    audio_df["image_location"] = [
+        get_image_location(name_video, i) for i in audio_df.image_location
+    ]
+    audio_df = softmax(
+        audio_df[audio_df.image_location.isin(image_location)][NAME_EMO_AUDIO[:-1]].values
+    )
+
+    if len(image_location) > len(audio_df):
+        last_pred_audio = audio_df[-1]
+        audio_df = np.vstack(
+            (audio_df, [last_pred_audio] * (len(image_location) - len(audio_df)))
+        )
+
+    predictions = [stat_df, dyn_df, audio_df]
+    num_predictions = len(predictions)
+
+    if weights_1:
+        final_predictions = predictions[0] * weights_1[0]
+        for i in range(1, num_predictions):
+            final_predictions += predictions[i] * weights_1[i]
+
+    else:
+        final_predictions = np.sum(predictions, axis=0) / num_predictions
+
+    av_prob = np.argmax(get_compound_expression(
+                final_predictions, DICT_CE,
+            ), axis=1)
+    
+    vs_prob = get_compound_expression(
+        predictions[0], DICT_CE)
+    vd_prob = get_compound_expression(
+        predictions[1], DICT_CE)
+    a_prob = get_compound_expression(
+        predictions[2], DICT_CE)
+
+    av_pred = get_mix_pred(final_predictions, av_prob)
+    vs_pred = get_mix_pred(predictions[0], np.argmax(vs_prob, axis=1))
+    vd_pred = get_mix_pred(predictions[1], np.argmax(vd_prob, axis=1))
+    a_pred = get_mix_pred(predictions[2], np.argmax(a_prob, axis=1))
+            
+    dict_pred_final = {'Audio-visual fusion':av_pred, 'Static visual model':vs_pred,'Dynamic visual model':vd_pred,'Audio model':a_pred}
+
+    plt = plot_compound_expression_prediction(
+            dict_preds = dict_pred_final,
+            save_path = None,
+            frame_indices = frame_indices,
+            title = "Basic emotion and compound expression predictions")
+    
+    df = pd.DataFrame(dict_pred_final)
+
+    return df, plt
+
+def get_evenly_spaced_frame_indices(total_frames, num_frames=10):
+    if total_frames <= num_frames:
+        return list(range(total_frames))
+    
+    step = total_frames / num_frames
+    return [int(np.round(i * step)) for i in range(num_frames)]

config.toml

@@ -0,0 +1,16 @@
+APP_VERSION = "0.0.0"
+FRAME_DOWNSAMPLING = 5
+AUDIO_STEP = 0.5
+CONFIDENCE_BE = 0.7
+
+[model_static]
+url = "https://huggingface.co/ElenaRyumina/face_emotion_recognition/resolve/main/FER_static_ResNet50_AffectNet.pt"
+path = "FER_static_ResNet50_AffectNet.pt"
+
+[model_dynamic]
+url = "https://huggingface.co/ElenaRyumina/face_emotion_recognition/resolve/main/FER_dinamic_LSTM_Aff-Wild2.pt"
+path = "FER_dinamic_LSTM_Aff-Wild2.pt"
+
+[model_audio]
+url = "https://drive.usercontent.google.com/download?id=11m53Kys3mdPALxbHQYc6kyww9QlQIkHA&export=download&authuser=0&confirm=t&uuid=ff23fbb0-5e4f-40b1-85bc-1cbbcdf0aeb7&at=APZUnTV5OentCsQjMpGGmIjKHBVP%3A1717752164159"
+path = "audio_ExprModelV3.pth"

requirements.txt

@@ -0,0 +1,8 @@
+gradio==4.24.0
+requests==2.31.0
+torch==2.1.2
+torchaudio==2.1.2
+torchvision==0.16.2
+mediapipe==0.10.9
+pillow==10.2.0
+toml==0.10.