update

examples/evaluation/step_1_run_evaluation.py

@@ -0,0 +1,166 @@
+#!/usr/bin/python3
+# -*- coding: utf-8 -*-
+import argparse
+import json
+import os
+from pathlib import Path
+import sys
+
+pwd = os.path.abspath(os.path.dirname(__file__))
+sys.path.append(os.path.join(pwd, "../../"))
+
+import librosa
+from gradio_client import Client
+import numpy as np
+from sklearn.metrics import precision_score, recall_score, accuracy_score, f1_score
+from tqdm import tqdm
+
+
+def get_args():
+    parser = argparse.ArgumentParser()
+
+    parser.add_argument(
+        "--test_set",
+        default=r"D:\Users\tianx\HuggingDatasets\nx_noise\data\speech\en-SG\vad",
+        type=str
+    )
+    parser.add_argument(
+        "--output_file",
+        default=r"fsmn-vad.jsonl",
+        type=str
+    )
+    parser.add_argument("--expected_sample_rate", default=8000, type=int)
+
+    args = parser.parse_args()
+    return args
+
+
+def get_metrics(ground_truth, predictions, total_duration, step=0.01):
+    """
+    基于时间点离散化的评估方法
+    :param ground_truth: 真实区间列表，格式 [[start1, end1], [start2, end2], ...]
+    :param predictions: 预测区间列表，格式同上
+    :param total_duration: 音频总时长（秒）
+    :param step: 时间离散化步长（默认10ms）
+    :return: 评估指标字典
+    """
+    # 生成时间点数组
+    time_points = np.arange(0, total_duration, step)
+
+    # 生成标签数组
+    y_true = np.zeros_like(time_points, dtype=int)
+    y_pred = np.zeros_like(time_points, dtype=int)
+
+    # 标记真实语音区间
+    for start, end in ground_truth:
+        mask = (time_points >= start) & (time_points <= end)
+        y_true[mask] = 1
+
+    # 标记预测语音区间
+    for start, end in predictions:
+        mask = (time_points >= start) & (time_points <= end)
+        y_pred[mask] = 1
+
+    # 计算指标
+    result = {
+        "accuracy": accuracy_score(y_true, y_pred),
+        "precision": precision_score(y_true, y_pred, zero_division=0),
+        "recall": recall_score(y_true, y_pred, zero_division=0),
+        "f1": f1_score(y_true, y_pred, zero_division=0)
+    }
+    return result
+
+
+def main():
+    args = get_args()
+
+    client = Client("http://127.0.0.1:7866/")
+
+    test_set = Path(args.test_set)
+    output_file = Path(args.output_file)
+
+    annotation_file = test_set / "vad.json"
+
+    with open(annotation_file.as_posix(), "r", encoding="utf-8") as f:
+        annotation = json.load(f)
+
+    total = 0
+    total_accuracy = 0
+    total_precision = 0
+    total_recall = 0
+    total_f1 = 0
+    total_duration = 0
+    progress_bar = tqdm(desc="evaluation")
+    with open(output_file.as_posix(), "w", encoding="utf-8") as f:
+        for row in annotation:
+            filename = row["filename"]
+            ground_truth_vad_segments = row["vad_segments"]
+
+            filename = test_set / filename
+
+            _, _, _, message = client.predict(
+                audio_file_t={
+                    "path": filename.as_posix(),
+                    "meta": {"_type": "gradio.FileData"}
+                },
+                audio_microphone_t=None,
+                start_ring_rate=0.5,
+                end_ring_rate=0.5,
+                ring_max_length=1,
+                min_silence_length=6,
+                max_speech_length=100000,
+                min_speech_length=15,
+                engine="fsmn-vad-by-webrtcvad-nx2-dns3",
+                api_name="/when_click_vad_button"
+            )
+            js = json.loads(message)
+            prediction_vad_segments = js["vad_segments"]
+            duration = js["duration"]
+
+            metrics = get_metrics(ground_truth_vad_segments, prediction_vad_segments, duration)
+            accuracy = metrics["accuracy"]
+            precision = metrics["precision"]
+            recall = metrics["recall"]
+            f1 = metrics["f1"]
+
+            row_ = {
+                "filename": filename.as_posix(),
+                "duration": duration,
+                "ground_truth": ground_truth_vad_segments,
+                "prediction": prediction_vad_segments,
+
+                "accuracy": accuracy,
+                "precision": precision,
+                "recall": recall,
+                "f1": f1,
+            }
+            row_ = json.dumps(row_, ensure_ascii=False)
+            f.write(f"{row_}\n")
+
+            total += 1
+            total_accuracy += accuracy
+            total_precision += precision
+            total_recall += recall
+            total_f1 += f1
+            total_duration += duration
+
+            average_accuracy = total_accuracy / total
+            average_precision = total_precision / total
+            average_recall = total_recall / total
+            average_f1 = total_f1 / total
+
+            progress_bar.update(1)
+            progress_bar.set_postfix({
+                "total": total,
+                "accuracy": average_accuracy,
+                "precision": average_precision,
+                "recall": average_recall,
+                "f1": average_f1,
+                "total_duration": f"{round(total_duration / 60, 4)}min",
+            })
+
+    return
+
+
+if __name__ == "__main__":
+    main()

main.py

@@ -101,6 +101,7 @@ def generate_image(signal: np.ndarray, speech_probs: np.ndarray, sample_rate: in
 
 def when_click_vad_button(audio_file_t = None, audio_microphone_t = None,
                           start_ring_rate: float = 0.5, end_ring_rate: float = 0.3,
+                          ring_max_length: int = 10,
                           min_silence_length: int = 2,
                           max_speech_length: int = 10000, min_speech_length: int = 10,
                           engine: str = None,
@@ -112,7 +113,7 @@ def when_click_vad_button(audio_file_t = None, audio_microphone_t = None,
     audio_t: Tuple = audio_file_t or audio_microphone_t
 
     sample_rate, signal = audio_t
-    audio_duration = signal.shape[-1] // 8000
+    audio_duration = signal.shape[-1] // sample_rate
     audio = np.array(signal / (1 << 15), dtype=np.float32)
 
     infer_engine_param = vad_engines.get(engine)
@@ -128,38 +129,55 @@ def when_click_vad_button(audio_file_t = None, audio_microphone_t = None,
         vad_info = infer_engine.infer(audio)
         time_cost = time.time() - begin
 
-        fpr = time_cost / audio_duration
-        info = {
-            "time_cost": round(time_cost, 4),
-            "audio_duration": round(audio_duration, 4),
-            "fpr": round(fpr, 4)
-        }
-        message = json.dumps(info, ensure_ascii=False, indent=4)
-
         probs = vad_info["probs"]
         lsnr = vad_info["lsnr"]
         # lsnr = lsnr / np.max(np.abs(lsnr))
         lsnr = lsnr / 30
 
         frame_step = infer_engine.config.hop_size
-        probs_ = process_speech_probs(audio, probs, frame_step)
-        probs_image = generate_image(audio, probs_)
-
-        lsnr_ = process_speech_probs(audio, lsnr, frame_step)
-        lsnr_image = generate_image(audio, lsnr_)
 
         # post process
         vad_post_process = PostProcess(
             start_ring_rate=start_ring_rate,
             end_ring_rate=end_ring_rate,
+            ring_max_length=ring_max_length,
             min_silence_length=min_silence_length,
             max_speech_length=max_speech_length,
             min_speech_length=min_speech_length
         )
-        vad = vad_post_process.post_process(probs)
-        vad_ = process_speech_probs(audio, vad, frame_step)
+        vad_segments = vad_post_process.get_vad_segments(probs)
+        vad_flags = vad_post_process.get_vad_flags(probs, vad_segments)
+
+        # vad_image
+        vad_ = process_speech_probs(audio, vad_flags, frame_step)
         vad_image = generate_image(audio, vad_)
 
+        # probs_image
+        probs_ = process_speech_probs(audio, probs, frame_step)
+        probs_image = generate_image(audio, probs_)
+
+        # lsnr_image
+        lsnr_ = process_speech_probs(audio, lsnr, frame_step)
+        lsnr_image = generate_image(audio, lsnr_)
+
+        # vad segment
+        vad_segments = [
+            [
+                v[0] * frame_step / sample_rate,
+                v[1] * frame_step / sample_rate
+            ] for v in vad_segments
+        ]
+
+        # message
+        rtf = time_cost / audio_duration
+        info = {
+            "vad_segments": vad_segments,
+            "time_cost": round(time_cost, 4),
+            "duration": round(audio_duration, 4),
+            "rtf": round(rtf, 4)
+        }
+        message = json.dumps(info, ensure_ascii=False, indent=4)
+
     except Exception as e:
         raise gr.Error(f"vad failed, error type: {type(e)}, error text: {str(e)}.")
 
@@ -240,10 +258,12 @@ def main():
                         with gr.Row():
                             vad_start_ring_rate = gr.Slider(minimum=0, maximum=1, value=0.5, step=0.1, label="start_ring_rate")
                             vad_end_ring_rate = gr.Slider(minimum=0, maximum=1, value=0.3, step=0.1, label="end_ring_rate")
-                            vad_min_silence_length = gr.Number(value=30, label="min_silence_length")
                         with gr.Row():
-                            vad_max_speech_length = gr.Number(value=100000, label="max_speech_length")
-                            vad_min_speech_length = gr.Number(value=15, label="min_speech_length")
+                            vad_ring_max_length = gr.Number(value=10, label="ring_max_length (*10ms)")
+                            vad_min_silence_length = gr.Number(value=6, label="min_silence_length (*10ms)")
+                        with gr.Row():
+                            vad_max_speech_length = gr.Number(value=100000, label="max_speech_length (*10ms)")
+                            vad_min_speech_length = gr.Number(value=15, label="min_speech_length (*10ms)")
                             vad_engine = gr.Dropdown(choices=vad_engine_choices, value=vad_engine_choices[0], label="engine")
                         vad_button = gr.Button(variant="primary")
                     with gr.Column(variant="panel", scale=5):
@@ -257,6 +277,7 @@ def main():
                     inputs=[
                         vad_audio_file, vad_audio_microphone,
                         vad_start_ring_rate, vad_end_ring_rate,
+                        vad_ring_max_length,
                         vad_min_silence_length,
                         vad_max_speech_length, vad_min_speech_length,
                         vad_engine,
@@ -288,7 +309,8 @@ def main():
         # share=True,
         share=False if platform.system() == "Windows" else False,
         server_name="127.0.0.1" if platform.system() == "Windows" else "0.0.0.0",
-        server_port=args.server_port
+        server_port=args.server_port,
+        show_error=True
     )
     return
 

toolbox/vad/utils.py

@@ -9,18 +9,20 @@ class PostProcess(object):
     def __init__(self,
                  start_ring_rate: float = 0.5,
                  end_ring_rate: float = 0.5,
-                 min_silence_length: int = 1,
-                 max_speech_length: float = 10,
-                 min_speech_length: float = 2,
+                 ring_max_length: int = 10,
+                 min_silence_length: int = 6,
+                 max_speech_length: float = 100000,
+                 min_speech_length: float = 15,
                  ):
         self.start_ring_rate = start_ring_rate
         self.end_ring_rate = end_ring_rate
+        self.ring_max_length = ring_max_length
         self.max_speech_length = max_speech_length
         self.min_speech_length = min_speech_length
         self.min_silence_length = min_silence_length
 
         # segments
-        self.ring_buffer = collections.deque(maxlen=10)
+        self.ring_buffer = collections.deque(maxlen=self.ring_max_length)
         self.triggered = False
 
         # vad segments
@@ -117,19 +119,27 @@ class PostProcess(object):
             vad_segments = vad_segments + [[self.start_idx, self.end_idx]]
         return vad_segments
 
-    def post_process(self, probs: List[float]):
+    def get_vad_segments(self, probs: List[float]):
         vad_segments = list()
         segments = self.vad(probs)
         vad_segments += segments
         segments = self.last_vad_segments()
         vad_segments += segments
 
+        return vad_segments
+
+    def get_vad_flags(self, probs: List[float], vad_segments: List[Tuple[int, int]]):
         result = [0] * len(probs)
         for begin, end in vad_segments:
             result[begin: end] = [1] * (end - begin)
 
         return result
 
+    def post_process(self, probs: List[float]):
+        vad_segments = self.get_vad_segments(probs)
+        vad_flags = self.get_vad_flags(probs, vad_segments)
+        return vad_flags
+
 
 if __name__ == "__main__":
     pass