app_utils.py

import gradio as gr
import torch
import time
import numpy as np
import mediapipe as mp
from PIL import Image
import cv2
# from pytorch_grad_cam.utils.image import show_cam_on_image
import scipy.io.wavfile as wav
# Importing necessary components for the Gradio app
from model import pth_model_static, pth_model_dynamic, cam, pth_processing
from face_utils import get_box, display_info
from config import DICT_EMO, config_data
from plot import statistics_plot
from moviepy.editor import AudioFileClip

import soundfile as sf
import torchaudio
from speechbrain.pretrained.interfaces import foreign_class
from paraformer import AudioReader, CttPunctuator, FSMNVad, ParaformerOffline
from gradio_client import Client
##############################################################################################
client = Client("Liusuthu/TextDepression")

mp_face_mesh = mp.solutions.face_mesh


classifier = foreign_class(
    source="pretrained_models/local-speechbrain/emotion-recognition-wav2vec2-IEMOCAP",  # ".\\emotion-recognition-wav2vec2-IEMOCAP"
    pymodule_file="custom_interface.py",
    classname="CustomEncoderWav2vec2Classifier",
    savedir="pretrained_models/local-speechbrain/emotion-recognition-wav2vec2-IEMOCAP",
)
ASR_model = ParaformerOffline()
vad = FSMNVad()
punc = CttPunctuator()


#########################################################################################
def text_api(text:str):
    result = client.predict(
        text,  # str  in '输入文字' Textbox component
        api_name="/predict",
    )
    return result

#######################################################################
#规范函数，只管值输入输出：
def text_score(text):
    if text==None:
        gr.Warning("提交内容为空！")
    else:
        string=text_api(text)
        part1 = str.partition(string, r"text")
        want1 = part1[2]
        label = want1[4:6]
        part2 = str.partition(string, r"probability")
        want2 = part2[2]
        prob = float(want2[3:-4])
        if label=="正向":
            score=-np.log10(prob*10)
        else:
            score=np.log10(prob*10)
        # print("from func:text_score————,text:",text,",score:",score)
        return text,score

def speech_score(audio):
    if audio==None:
        gr.Warning("提交内容为空！请等待音频加载完毕后再尝试提交！")
    else:
        print(type(audio))
        print(audio)
        sample_rate, signal = audio  # 这是语音的输入
        signal = signal.astype(np.float32)
        signal /= np.max(np.abs(signal))
        sf.write("data/a.wav", signal, sample_rate)
        signal, sample_rate = torchaudio.load("data/a.wav")
        signal1 = torchaudio.transforms.Resample(orig_freq=sample_rate, new_freq=16000)(
            signal
        )
        torchaudio.save("data/out.wav", signal1, 16000, encoding="PCM_S", bits_per_sample=16)
        Audio = "data/out.wav"
        speech, sample_rate = AudioReader.read_wav_file(Audio)
        if signal == "none":
            return "none", "none", "haha"
        else:
            segments = vad.segments_offline(speech)
            text_results = ""
            for part in segments:
                _result = ASR_model.infer_offline(
                    speech[part[0] * 16 : part[1] * 16], hot_words="任意热词 空格分开"
                )
                text_results += punc.punctuate(_result)[0]
    
            out_prob, score, index, text_lab = classifier.classify_batch(signal1)
            print("from func:speech_score————type and value of prob:")
            print(type(out_prob.squeeze(0).numpy()))
            print(out_prob.squeeze(0).numpy())
            print("from func:speech_score————type and value of resul_label:")
            print(type(text_lab[-1]))
            print(text_lab[-1])
            #return text_results, out_prob.squeeze(0).numpy(), text_lab[-1], Audio
            prob=out_prob.squeeze(0).numpy()
            #print(prob)
            score2=10*prob[0]-10*prob[1]
            if score2>=0:
                score2=np.log10(score2)
            else:
                score2=-np.log10(-score2)
            # print("from func:speech_score————score2:",score2)
            # print("from func:speech_score————",text_lab[-1])
            
            text,score1=text_score(text_results)
            # # text_emo=str(get_text_score(text_results))
            # print("from func:speech_score————text:",text,",score1:",score1)
            score=(2/3)*score1+(1/3)*score2
            
    
            return text,score


def video_score(video):
    if video==None:
        gr.Warning("提交内容为空！请等待视频加载完毕后再尝试提交！")
    else:
        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))
    
        path_save_video_face = 'result_face.mp4'
        vid_writer_face = cv2.VideoWriter(path_save_video_face, cv2.VideoWriter_fourcc(*'mp4v'), fps, (224, 224))
    
        # path_save_video_hm = 'result_hm.mp4'
        # vid_writer_hm = cv2.VideoWriter(path_save_video_hm, cv2.VideoWriter_fourcc(*'mp4v'), fps, (224, 224))
    
        lstm_features = []
        count_frame = 1
        count_face = 0
        probs = []
        frames = []
        last_output = None
        last_heatmap = None 
        cur_face = None
    
        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))
                            with torch.no_grad():
                                features = torch.nn.functional.relu(pth_model_static.extract_features(cur_face_copy)).detach().numpy()
    
                            # grayscale_cam = cam(input_tensor=cur_face_copy)
                            # grayscale_cam = grayscale_cam[0, :]
                            # cur_face_hm = cv2.resize(cur_face,(224,224), interpolation = cv2.INTER_AREA)
                            # cur_face_hm = np.float32(cur_face_hm) / 255
                            # heatmap = show_cam_on_image(cur_face_hm, grayscale_cam, use_rgb=False)
                            # last_heatmap = heatmap
            
                            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)
                            with torch.no_grad():
                                output = pth_model_dynamic(lstm_f).detach().numpy()
                            last_output = output
    
                            if count_face == 0:
                                count_face += 1
    
                        else:
                            if last_output is not None:
                                output = last_output
                                # heatmap = last_heatmap
    
                            elif last_output is None:
                                output = np.empty((1, 7))
                                output[:] = np.nan
                                
                        probs.append(output[0])
                        frames.append(count_frame)
                else:
                    if last_output is not None:
                        lstm_features = []
                        empty = np.empty((7))
                        empty[:] = np.nan
                        probs.append(empty)
                        frames.append(count_frame)                        
    
                if cur_face is not None:
                    # heatmap_f = display_info(heatmap, 'Frame: {}'.format(count_frame), box_scale=.3)
    
                    cur_face = cv2.cvtColor(cur_face, cv2.COLOR_RGB2BGR)
                    cur_face = cv2.resize(cur_face, (224,224), interpolation = cv2.INTER_AREA)
                    cur_face = display_info(cur_face, 'Frame: {}'.format(count_frame), box_scale=.3)
                    vid_writer_face.write(cur_face)
                    # vid_writer_hm.write(heatmap_f)
    
                count_frame += 1
                if count_face != 0:
                    count_face += 1
    
            vid_writer_face.release()
            # vid_writer_hm.release()
    
            stat = statistics_plot(frames, probs)
    
            if not stat:
                return None, None
    
        #for debug
        print("from func:video_score————")
        print(type(frames))
        print(frames)
        print(type(probs))
        print(probs)        
        # to calculate scores
        nan=float('nan')
        s1 = 0
        s2 = 0
        s3 = 0
        # s4 = 0
        # s5 = 0
        # s6 = 0
        # s7 = 0
        frames_len=len(frames)
        for i in range(frames_len):
            if np.isnan(probs[i][0]):
                frames_len=frames_len-1
            else: 
                s1=s1+probs[i][0]
                s2=s2+probs[i][1]
                s3=s3+probs[i][2]
                # s4=s4+probs[i][3]
                # s5=s5+probs[i][4]
                # s6=s6+probs[i][5]
                # s7=s7+probs[i][6]
        s1=s1/frames_len
        s2=s2/frames_len
        s3=s3/frames_len
        # s4=s4/frames_len
        # s5=s5/frames_len
        # s6=s6/frames_len
        # s7=s7/frames_len
        # scores=[s1,s2,s3,s4,s5,s6,s7]
        # scores_str=str(scores)
        # score1=0*scores[0]-8*scores[1]+4*scores[2]+0*scores[3]+2*scores[4]+2*scores[5]+4*scores[6]
        #print("from func:video_score————score1=",score1)
        #print("from func:video_score————logs:")
        # with open("local_data/data.txt",'a', encoding="utf8") as f:
        #     f.write(scores_str+'\n')
    
        # with open("local_data/data.txt",'r', encoding="utf8") as f:
        #     for i in f:
        #         print(i)
    
    
        print(str([s1,s2,s3]))
        if s1>=0.4:
            score1=0
        else:
            if s2>=s3:
                score1=-1
            else:
                score1=+1
        #trans the audio file
        my_audio_clip = AudioFileClip(video)
        my_audio_clip.write_audiofile("data/audio.wav",ffmpeg_params=["-ac","1"])
    
        audio = wav.read('data/audio.wav')
    
        text,score2=speech_score(audio)
    
        #print("from func:video_score————text:",text)
    
        score=(score1+6*score2)/7
        #print("from func:video_score————score:",score)
        return text,score
#######################################################################