speech.py

import os

# from sys import path
# path.append(r"./pretrained_models")
import gradio as gr
import numpy as np
import soundfile as sf
import scipy.io.wavfile as wav
# import torch
# import csv
# import json
# import math
# import os
# import struct
import torchaudio

# logging.basicConfig(stream=sys.stdout, level=logging.DEBUG)
# from pydub import AudioSegment
# import pyaudio
from speechbrain.pretrained.interfaces import foreign_class

# import scipy.io.wavfile as wav
from paraformer import AudioReader, CttPunctuator, FSMNVad, ParaformerOffline

from gradio_client import Client


client = Client("Liusuthu/TextDepression")


sum1=0
sum2=0
sum3=0

# import time

os.environ["no_proxy"] = "localhost,127.0.0.1,::1"
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)
        print(string)
        if string=="[{}]":
            text="空"
            score=0
        else:
            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):
    global sum1,sum2,sum3
    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("a.wav", signal, sample_rate)
        signal, sample_rate = torchaudio.load("a.wav")
        signal1 = torchaudio.transforms.Resample(orig_freq=sample_rate, new_freq=16000)(
            signal
        )
        torchaudio.save("out.wav", signal1, 16000, encoding="PCM_S", bits_per_sample=16)
        Audio = "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])

            if text_results!="[{}]":
                text,score1=text_score(text_results)
            else:
                text="空"
                score1=0
            # # text_emo=str(get_text_score(text_results))
            # print("from func:speech_score————text:",text,",score1:",score1)
            score=(1/3)*score1+(2/3)*score2
            print("语音score:",score2,",文本score:",score1,",整体score:",score,",文本:",text)
            sum1=sum1+int(score2>0)
            sum2=sum2+int(score1>0)
            sum3=sum3+int(score>0)
            
    
            return text,score



#############################################################################################

def classify_continuous(audio):
    print(type(audio))
    print(audio)
    sample_rate, signal = audio  # 这是语音的输入
    signal = signal.astype(np.float32)
    signal /= np.max(np.abs(signal))
    sf.write("a.wav", signal, sample_rate)
    signal, sample_rate = torchaudio.load("a.wav")
    signal1 = torchaudio.transforms.Resample(orig_freq=sample_rate, new_freq=16000)(
        signal
    )
    torchaudio.save("out.wav", signal1, 16000, encoding="PCM_S", bits_per_sample=16)
    Audio = "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)
        return text_results, out_prob.squeeze(0).numpy(), text_lab[-1]



def run_results(file):
    global sum1,sum2,sum3
    sum1=0
    sum2=0
    sum3=0
    for i in range(29):
        print("第",i+1,"段音频：")
        if i<9:
            filename=file+"/0"+str(i+1)+".wav"
        else:
            filename=file+"/"+str(i+1)+".wav"
        audio = wav.read(filename)
        text,score=speech_score(audio)
        # print(score)
        with open("data.txt",'a', encoding="utf8") as f:
            f.write(str(score)+'\n')
    #处理完毕后
    print("本次批量处理结果：")
    with open("data.txt",'r', encoding="utf8") as f:
        for j in f:
            print(j,end="")
    print("语音:",sum1,",文本:",sum2,",融合：",sum3)
    return "测试完毕"



# with gr.Blocks() as demo:
#     with gr.Row():
#         with gr.Column(scale=1):
#             with gr.Row():
#                 input_audio=gr.Audio(sources='microphone')
#             with gr.Row():
#                 submit_button=gr.Button("提交音频")
#         with gr.Column(scale=2):    
#             with gr.Row():
#                 with gr.Row():
#                     with gr.Column():
#                         with gr.Row():
#                             output_log1=gr.Textbox(interactive=True)
#                         with gr.Row():
#                             output_log2=gr.Textbox(interactive=False)
#             with gr.Row():
#                 run_button=gr.Button("开始测试")

#     submit_button.click(
#         fn=speech_score,
#         inputs=[input_audio],
#         outputs=[output_log1,output_log2]
#     )
#     run_button.click(
#         fn=run_results,
#         inputs=[output_log1],
#         outputs=[output_log2]
#     )

demo = gr.Interface(
    classify_continuous,
    gr.Audio("02020027-HC/07.wav"),
    [
        gr.Text(label="语音识别结果"),
        gr.Text(label="音频情感识别1"),
        gr.Text(label="音频情感识别2"),
    ],
)

demo.launch()