update

.gitignore

@@ -17,3 +17,4 @@
 
 #**/*.wav
 **/*.xlsx
+**/*.onnx

examples/online_model_test/step_1_predict.py

@@ -0,0 +1,216 @@
+#!/usr/bin/python3
+# -*- coding: utf-8 -*-
+import argparse
+import glob
+import json
+import os
+from pathlib import Path
+from tqdm import tqdm
+
+import librosa
+import numpy as np
+import onnxruntime as ort
+import pandas as pd
+import torch
+import torchaudio
+
+
+def get_args():
+    parser = argparse.ArgumentParser()
+    parser.add_argument(
+        "--audio_dir",
+        default=r"D:\Users\tianx\HuggingDatasets\international_voice\data\sea-idn\audio_lib_hkg_1\audio_lib_hkg_1\zh-TW",
+        type=str,
+    )
+    parser.add_argument("--onnx_model_file", default="zh-TW.onnx", type=str)
+    parser.add_argument("--target_duration", default=8.0, type=float)
+
+    parser.add_argument("--output_file", default="zh_tw_predict.xlsx", type=str)
+
+    args = parser.parse_args()
+    return args
+
+
+class OnlineModelConfig(object):
+    def __init__(self,
+                 sample_rate: int = 8000,
+                 n_fft: int = 1024,
+                 hop_size: int = 512,
+                 n_mels: int = 80,
+                 f_min: float = 10.0,
+                 f_max: float = 3800.0,
+                 ):
+        self.sample_rate = sample_rate
+        self.n_fft = n_fft
+        self.hop_size = hop_size
+        self.n_mels = n_mels
+        self.f_min = f_min
+        self.f_max = f_max
+
+
+class OnlineModelInference(object):
+    def __init__(self,
+                 model_path: str,
+                 ):
+        self.model_path = model_path
+
+        providers = [
+            "CUDAExecutionProvider", "CPUExecutionProvider"
+        ] if torch.cuda.is_available() else [
+            "CPUExecutionProvider"
+        ]
+        self.session = ort.InferenceSession(self.model_path, providers=providers)
+
+        self.config = OnlineModelConfig()
+
+        self.mel_transform = torchaudio.transforms.MelSpectrogram(
+            sample_rate=self.config.sample_rate,
+            n_fft=self.config.n_fft,
+            hop_length=self.config.hop_size,
+            n_mels=self.config.n_mels,
+            f_min=self.config.f_min,
+            f_max=self.config.f_max,
+            window_fn=torch.hamming_window
+        )
+
+    def predict_by_ndarray(self,
+                           sub_signal: np.ndarray,
+                           h: np.ndarray = None,
+                           c: np.ndarray = None,
+                           ):
+        # sub_signal, shape: [num_samples,]
+        sub_signal = torch.tensor(sub_signal, dtype=torch.float32)
+
+        sub_signal = sub_signal.unsqueeze(0)
+        # sub_signal, shape: [1, num_samples]
+        mel_spec = self.mel_transform.forward(sub_signal)
+        # mel_spec, shape: [1, n_mels, n_frames]
+        mel_spec = torch.transpose(mel_spec, dim0=1, dim1=2)
+        # mel_spec, shape: [1, n_frames, n_mels]
+
+        h = torch.tensor(h) if h is not None else None
+        c = torch.tensor(c) if h is not None else None
+        label, prob, h, c = self.predict_by_mel_spec(mel_spec, h=h, c=c)
+        # h, c: torch.Tensor
+        h = h.numpy()
+        c = c.numpy()
+        return label, prob, h, c
+
+    def predict_by_mel_spec(self,
+                            mel_spec: torch.Tensor,
+                            h: torch.Tensor = None,
+                            c: torch.Tensor = None,
+                            ):
+        # mel_spec, shape: [1, n_frames, n_mels]
+
+        if h is None:
+            h = np.zeros((3, 1, 64), dtype=np.float32)  # 3层LSTM，批次大小1，隐藏大小64
+        else:
+            h = h.numpy()
+        if c is None:
+            c = np.zeros((3, 1, 64), dtype=np.float32)  # 3层LSTM，批次大小1，隐藏大小64
+        else:
+            c = c.numpy()
+
+        mel_spec_np = mel_spec.numpy()
+        outputs = self.session.run(
+            input_feed={
+                "input": mel_spec_np,
+                "h": h,
+                "c": c
+            },
+            output_names=[
+                "output", "h_out", "c_out"
+            ],
+        )
+        logits, h, c = outputs
+        # logits, np.ndarray, shape: [b, num_labels]
+        # h, c: np.ndarray
+        h = torch.tensor(h)
+        c = torch.tensor(c)
+
+        probs = torch.softmax(torch.tensor(logits), dim=1)
+        max_prob, predicted_label_index = torch.max(probs, dim=1)
+
+        label = self.get_label_by_index(predicted_label_index.item())
+        prob = max_prob.item()
+        return label, prob, h, c
+
+    @staticmethod
+    def get_label_by_index(index: int):
+        label_map = {
+            0: "voice",
+            1: "voicemail",
+            2: "mute",
+            3: "noise"
+        }
+        result = label_map[index]
+        return result
+
+
+def main():
+    args = get_args()
+
+    audio_dir = Path(args.audio_dir)
+
+    model = OnlineModelInference(model_path=args.onnx_model_file)
+
+    result = list()
+    for filename in tqdm(audio_dir.glob("**/active_media_r_*.wav")):
+        splits = filename.stem.split("_")
+        call_id = splits[3]
+        language = splits[4]
+        scene_id = splits[5]
+
+        signal, sample_rate = librosa.load(filename.as_posix(), sr=8000)
+        duration = librosa.get_duration(y=signal, sr=sample_rate)
+        signal_length = len(signal)
+        if signal_length == 0:
+            continue
+
+        target_duration = args.target_duration * sample_rate
+        target_duration = int(target_duration)
+
+        predict_result = list()
+        h = None
+        c = None
+        for begin in range(0, target_duration, sample_rate*2):
+            end = begin + sample_rate*2
+            sub_signal = signal[begin: end]
+            if len(sub_signal) == 0:
+                break
+            label, prob, h, c = model.predict_by_ndarray(sub_signal, h=h, c=c)
+            predict_result.append({
+                "label": label,
+                "prob": prob,
+            })
+        label_list = [p["label"] for p in predict_result]
+        predict_result_ = json.dumps(predict_result, ensure_ascii=False, indent=4)
+        label2 = predict_result[0]["label"]
+        prob2 = predict_result[0]["prob"]
+
+        ground_truth_ = "voicemail" if any([l == "voicemail" for l in label_list]) else "else"
+        flag = 1 if label2 == "voicemail" else 0
+
+        row = {
+            "call_id": call_id,
+            "language": language,
+            "scene_id": scene_id,
+            "filename": filename.as_posix(),
+            "duration": duration,
+            "predict_result": predict_result_,
+            "label2": label2,
+            "prob2": prob2,
+            "ground_truth_": ground_truth_,
+            "flag": flag,
+        }
+        result.append(row)
+
+    result = pd.DataFrame(result)
+    result.to_excel(args.output_file, index=False)
+
+    return
+
+
+if __name__ == "__main__":
+    main()

examples/online_model_test/step_2_audio_filter.py

@@ -0,0 +1,43 @@
+#!/usr/bin/python3
+# -*- coding: utf-8 -*-
+import argparse
+from pathlib import Path
+import shutil
+
+import pandas as pd
+
+
+def get_args():
+    parser = argparse.ArgumentParser()
+
+    parser.add_argument("--predict_file", default="zh_tw_predict.xlsx", type=str)
+    parser.add_argument(
+        "--output_dir",
+        default=r"D:\Users\tianx\HuggingDatasets\international_voice\data\calling\886",
+        type=str,
+    )
+    args = parser.parse_args()
+    return args
+
+
+def main():
+    args = get_args()
+
+    output_dir = Path(args.output_dir)
+
+    df = pd.read_excel(args.predict_file)
+    for i, row in df.iterrows():
+        filename = row["filename"]
+        ground_truth_ = row["ground_truth_"]
+
+        if ground_truth_ == "voicemail":
+            shutil.copy(
+                filename,
+                output_dir.as_posix()
+            )
+
+    return
+
+
+if __name__ == "__main__":
+    main()

examples/online_model_test/step_3_make_test.py

@@ -0,0 +1,74 @@
+#!/usr/bin/python3
+# -*- coding: utf-8 -*-
+import argparse
+from pathlib import Path
+import shutil
+
+from gradio_client import Client, handle_file
+import librosa
+import pandas as pd
+from tqdm import tqdm
+
+
+def get_args():
+    parser = argparse.ArgumentParser()
+
+    parser.add_argument(
+        "--src_dir",
+        default=r"D:\Users\tianx\HuggingDatasets\international_voice\data\calling\886",
+        type=str,
+    )
+    parser.add_argument(
+        "--tgt_dir",
+        default=r"D:\Users\tianx\HuggingDatasets\international_voice\data\voice_test_examples\886\96",
+        type=str,
+    )
+    parser.add_argument(
+        "--early_media_file",
+        default=r"D:\Users\tianx\HuggingDatasets\international_voice\data\voice_test_examples\886\97\early_media_ba95fafd-8e2f-488f-8e5a-9bada95e24fb.wav",
+        type=str,
+    )
+    args = parser.parse_args()
+    return args
+
+
+def main():
+    args = get_args()
+
+    src_dir = Path(args.src_dir)
+    tgt_dir = Path(args.tgt_dir)
+
+    client = Client("http://10.75.27.247:7861/")
+
+    for filename in tqdm(src_dir.glob("*.wav")):
+        splits = filename.stem.split("_")
+        call_id = splits[3]
+
+        filename_ = filename.as_posix()
+        y, sr = librosa.load(filename_)
+        duration = librosa.get_duration(y=y, sr=sr)
+        if duration < 20:
+            filename_, _ = client.predict(
+                audio_t=handle_file(filename_),
+                pad_seconds=20,
+                pad_mode="repeat",
+                api_name="/when_click_pad_audio"
+            )
+
+        active_media_file = tgt_dir / f"active_media_{call_id}.wav"
+        early_media_file = tgt_dir / f"early_media_{call_id}.wav"
+
+        shutil.copy(
+            filename_,
+            active_media_file.as_posix(),
+        )
+        shutil.copy(
+            args.early_media_file,
+            early_media_file.as_posix(),
+        )
+
+    return
+
+
+if __name__ == "__main__":
+    main()

main.py

@@ -1,5 +1,25 @@
 #!/usr/bin/python3
 # -*- coding: utf-8 -*-
+"""
+docker build -t cc_audio_8:v20250828_1343 .
+docker stop cc_audio_8_7864 && docker rm cc_audio_8_7864
+docker run -itd \
+--name cc_audio_8_7864 \
+--restart=always \
+--network host \
+-e server_port=7865 \
+cc_audio_8:v20250828_1343 /bin/bash
+
+docker run -itd \
+--name cc_audio_8_7864 \
+--network host \
+--gpus all \
+--privileged \
+--ipc=host \
+python:3.12 /bin/bash
+
+nohup python3 main.py --server_port 7864 --hf_token hf_coRVvzwA****jLmZHwJobEX &
+"""
 import argparse
 from functools import lru_cache
 from pathlib import Path
@@ -17,11 +37,11 @@ import torch
 from project_settings import environment, project_path
 from toolbox.torch.utils.data.vocabulary import Vocabulary
 from tabs.cls_tab import get_cls_tab
-from tabs.split_tabs import get_split_tab
+from tabs.split_tab import get_split_tab
+from tabs.voicemail_tab import get_voicemail_tab
 from tabs.shell_tab import get_shell_tab
 
 
-
 def get_args():
     parser = argparse.ArgumentParser()
     parser.add_argument(
@@ -83,39 +103,6 @@ def load_model(model_file: Path):
     return d
 
 
-def click_button(audio: np.ndarray,
-                 model_name: str,
-                 ground_true: str) -> Tuple[str, float]:
-
-    sample_rate, signal = audio
-
-    model_file = "trained_models/{}.zip".format(model_name)
-    model_file = Path(model_file)
-    d = load_model(model_file)
-
-    model = d["model"]
-    vocabulary = d["vocabulary"]
-
-    inputs = signal / (1 << 15)
-    inputs = torch.tensor(inputs, dtype=torch.float32)
-    inputs = torch.unsqueeze(inputs, dim=0)
-
-    with torch.no_grad():
-        logits = model.forward(inputs)
-        probs = torch.nn.functional.softmax(logits, dim=-1)
-        label_idx = torch.argmax(probs, dim=-1)
-
-    label_idx = label_idx.cpu()
-    probs = probs.cpu()
-
-    label_idx = label_idx.numpy()[0]
-    prob = probs.numpy()[0][label_idx]
-
-    label_str = vocabulary.get_token_from_index(label_idx, namespace="labels")
-
-    return label_str, round(prob, 4)
-
-
 def main():
     args = get_args()
 
@@ -148,6 +135,10 @@ def main():
                 examples_dir=args.examples_dir,
                 trained_model_dir=args.trained_model_dir,
             )
+            _ = get_voicemail_tab(
+                examples_dir=args.examples_dir,
+                trained_model_dir=args.trained_model_dir,
+            )
             _ = get_split_tab(
                 examples_dir=args.examples_dir,
                 trained_model_dir=args.trained_model_dir,

requirements.txt

@@ -12,3 +12,4 @@ evaluate
 gradio
 python-dotenv
 numpy
+onnxruntime

tabs/voicemail_tab.py

@@ -0,0 +1,149 @@
+#!/usr/bin/python3
+# -*- coding: utf-8 -*-
+import json
+from functools import lru_cache
+from pathlib import Path
+import shutil
+import tempfile
+import zipfile
+from typing import Tuple
+
+import gradio as gr
+import torch
+
+from project_settings import project_path
+from toolbox.torch.utils.data.vocabulary import Vocabulary
+
+
+@lru_cache(maxsize=100)
+def load_model(model_file: Path):
+    with zipfile.ZipFile(model_file, "r") as f_zip:
+        out_root = Path(tempfile.gettempdir()) / "cc_audio_8"
+        if out_root.exists():
+            shutil.rmtree(out_root.as_posix())
+        out_root.mkdir(parents=True, exist_ok=True)
+        f_zip.extractall(path=out_root)
+
+    tgt_path = out_root / model_file.stem
+    jit_model_file = tgt_path / "trace_model.zip"
+    vocab_path = tgt_path / "vocabulary"
+
+    vocabulary = Vocabulary.from_files(vocab_path.as_posix())
+
+    with open(jit_model_file.as_posix(), "rb") as f:
+        model = torch.jit.load(f)
+    model.eval()
+
+    shutil.rmtree(tgt_path)
+
+    d = {
+        "model": model,
+        "vocabulary": vocabulary
+    }
+    return d
+
+
+def when_click_voicemail_button(audio_t,
+                                model_name: str,
+                                ground_true: str) -> Tuple[str, float]:
+
+    sample_rate, signal = audio_t
+
+    model_file = project_path / f"trained_models/{model_name}.zip"
+    d = load_model(model_file)
+
+    model = d["model"]
+    vocabulary = d["vocabulary"]
+
+    inputs = signal / (1 << 15)
+    inputs = torch.tensor(inputs, dtype=torch.float32)
+    inputs = torch.unsqueeze(inputs, dim=0)
+
+    num_samples = inputs.shape[-1]
+
+    outputs = list()
+    with torch.no_grad():
+        for begin in range(0, num_samples, sample_rate*2):
+            end = begin + int(sample_rate*2)
+            sub_inputs = inputs[:, begin:end]
+            if sub_inputs.shape[-1] < sample_rate:
+                raise AssertionError(f"audio duration less than: {sample_rate}")
+
+            logits = model.forward(sub_inputs)
+            probs = torch.nn.functional.softmax(logits, dim=-1)
+            label_idx = torch.argmax(probs, dim=-1)
+
+            label_idx = label_idx.cpu()
+            probs = probs.cpu()
+
+            label_idx = label_idx.numpy()[0]
+            prob = probs.numpy()[0][label_idx]
+            prob: float = round(float(prob), 4)
+
+            label_str: str = vocabulary.get_token_from_index(label_idx, namespace="labels")
+
+            outputs.append({
+                "label": label_str,
+                "prob": prob,
+            })
+    outputs = json.dumps(outputs, ensure_ascii=False, indent=4)
+    return outputs
+
+
+def get_voicemail_tab(examples_dir: str, trained_model_dir: str):
+    voicemail_examples_dir = Path(examples_dir)
+    voicemail_trained_model_dir = Path(trained_model_dir)
+
+    # models
+    voicemail_model_choices = list()
+    for filename in voicemail_trained_model_dir.glob("*.zip"):
+        model_name = filename.stem
+        if model_name == "examples":
+            continue
+        voicemail_model_choices.append(model_name)
+    model_choices = list(sorted(voicemail_model_choices))
+
+    # examples zip
+    voicemail_example_zip_file = voicemail_trained_model_dir / "examples.zip"
+    with zipfile.ZipFile(voicemail_example_zip_file.as_posix(), "r") as f_zip:
+        out_root = voicemail_examples_dir
+        if out_root.exists():
+            shutil.rmtree(out_root.as_posix())
+        out_root.mkdir(parents=True, exist_ok=True)
+        f_zip.extractall(path=out_root)
+
+    # examples
+    voicemail_examples = list()
+    for filename in voicemail_examples_dir.glob("**/*/*.wav"):
+        label = filename.parts[-2]
+        voicemail_examples.append([
+            filename.as_posix(),
+            model_choices[0],
+            label
+        ])
+
+    with gr.TabItem("voicemail"):
+        with gr.Row():
+            with gr.Column(scale=3):
+                voicemail_audio = gr.Audio(label="audio")
+                with gr.Row():
+                    with gr.Column(scale=3):
+                        voicemail_model_name = gr.Dropdown(choices=model_choices, value=model_choices[0], label="model_name")
+                    with gr.Column(scale=3):
+                        voicemail_ground_true = gr.Textbox(label="ground_true")
+
+                voicemail_button = gr.Button("run", variant="primary")
+            with gr.Column(scale=3):
+                voicemail_outputs = gr.Textbox(label="outputs")
+
+        voicemail_button.click(
+            when_click_voicemail_button,
+            inputs=[voicemail_audio, voicemail_model_name, voicemail_ground_true],
+            outputs=[voicemail_outputs],
+        )
+
+    return locals()
+
+
+if __name__ == "__main__":
+    pass