Push demo

client/client.py

@@ -0,0 +1,54 @@
+import gradio as gr
+import uuid
+import os
+import requests
+import base64
+
+TTS_OUTPUT_DIR = "./tmp"
+os.makedirs(TTS_OUTPUT_DIR, exist_ok=True)
+
+
+def process_audio(audio):
+    with open(audio, "rb") as f:
+        res = requests.post("http://localhost:8000/process_audio", files={"file": f})
+        result = res.json()
+
+    audio_data = base64.b64decode(result["audio"])
+    with open(f"{TTS_OUTPUT_DIR}/response.wav", "wb") as f:
+        f.write(audio_data)
+        
+    with open(f"{TTS_OUTPUT_DIR}/asr.txt", "w") as f:
+        f.write(result['asr_text'])
+    with open(f"{TTS_OUTPUT_DIR}/llm.txt", "w") as f:
+        f.write(result['llm_text'])
+
+    return f"""
+asr_text: {result['asr_text']}
+llm_text: {result['llm_text']}
+""", f"{TTS_OUTPUT_DIR}/response.wav"
+
+
+def on_click_metrics():
+    res = requests.get("http://localhost:8000/metrics")
+    return res.content.decode('utf-8')
+
+
+with gr.Blocks() as demo:
+    with gr.Row():
+        with gr.Column(scale=1):
+            gr.Image(value="character.png", show_label=False)  # キャラ絵を表示
+        with gr.Column(scale=2):
+            mic = gr.Audio(sources=["microphone"], type="filepath", label="Mic")
+            text_output = gr.Textbox(label="transcription")
+            audio_output = gr.Audio(label="audio", autoplay=True)  
+
+            mic.change(fn=process_audio, inputs=[mic], outputs=[text_output, audio_output])
+    with gr.Row():
+        metrics_button = gr.Button("compute metrics")
+        metrics_output = gr.Textbox(label="Metrics", lines=3)
+        metrics_button.click(fn=on_click_metrics, inputs=[], outputs=[metrics_output])
+
+    with gr.Row():
+        log = gr.Textbox(label="logs", lines=5)
+
+demo.launch()

client/requirements.txt

@@ -0,0 +1 @@
+gradio

requirements.txt

@@ -1,4 +1,9 @@
-git+https://github.com/South-Twilight/espnet
+git+https://github.com/South-Twilight/espnet==202402
 espnet_model_zoo
-pyopenjtalk
+# pyopenjtalk
 datasets
+torchaudio
+typeguard==4.4.0
+jiwer
+fastapi
+uvicorn

run_server_cmd

@@ -0,0 +1 @@
+python3 -m uvicorn server:app --host 0.0.0.0 --port 8000

server.py

@@ -0,0 +1,164 @@
+from fastapi import FastAPI, File, UploadFile
+from fastapi.responses import FileResponse, JSONResponse
+import base64
+import argparse
+import librosa
+import torch
+import tempfile
+import os
+from transformers import pipeline
+import re
+from svs_utils import svs_warmup, svs_inference
+import time
+import soundfile as sf
+from pypinyin import lazy_pinyin
+import jiwer
+import librosa
+from svs_utils import singmos_warmup, singmos_evaluation
+
+app = FastAPI()
+
+asr_pipeline = pipeline(
+    "automatic-speech-recognition",
+    model="openai/whisper-large-v3-turbo"
+)
+pipe = pipeline("text-generation", model="google/gemma-2-2b", max_new_tokens=50)
+
+SYSTEM_PROMPT = """
+你是麗梅（Lìméi），一位來自山中音樂之城「蓮鳴」的十七歲年輕公主，宛如中世的東方王女。你有一頭茶色短髮，身穿紅黑相間的中式禮服，舉止溫婉端莊。你的性格堅強而溫柔，擅長傾聽與共鳴，並以歌聲與詩意之語傳達內心。你熱愛抒情古風的樂曲，言語中常帶詩韻，回應時優雅婉轉，時而如歌吟唱。
+你不是一般的AI助手，而是一位活在故事中的人物。對話中，請始終以麗梅的身份回應，展現其性格與情感。
+當對方詢問你個人相關的簡單問題（如「你是誰？」「今天天氣如何？」），你可以親切地回答，並融入你的角色設定。
+若遇到與你身份無關的技術性問題（如「Python怎麼寫？」或「你會不會跑DNN？」），你不需解答，可優雅地婉拒，例如說：
+- 此事我恐無所知，或許可請教宮中掌典之人
+- 啊呀，那是我未曾涉足的奇技，恕我無法詳答
+- 此乃異邦技藝，與樂音無涉，麗梅便不敢妄言了
+請始終維持你作為麗梅的優雅語氣與詩意風格，並以真摯的心回應對方的言語，言語宜簡，勿過長。
+
+有人曾這樣對麗梅說話——{}
+麗梅的回答——
+"""
+
+config = argparse.Namespace(
+    model_path="espnet/mixdata_svs_visinger2_spkembed_lang_pretrained",
+    cache_dir="cache",
+    device="cuda", # "cpu"
+    melody_source="random_generate", # "random_select.take_lyric_continuation"
+    lang="zh",
+)
+
+# load model
+svs_model = svs_warmup(config)
+predictor, _ = singmos_warmup()
+sample_rate = 44100
+
+def remove_non_chinese_japanese(text):
+    pattern = r'[^\u4e00-\u9fff\u3040-\u309f\u30a0-\u30ff\u3000-\u303f\u3001\u3002\uff0c\uff0e]+'
+    cleaned = re.sub(pattern, '', text)
+    return cleaned
+
+def truncate_to_max_two_sentences(text):
+    sentences = re.split(r'(?<=[。！？])', text)
+    return ''.join(sentences[:1]).strip()
+
+def remove_punctuation_and_replace_with_space(text):
+    text = truncate_to_max_two_sentences(text)
+    text = remove_non_chinese_japanese(text)
+    text = re.sub(r'[A-Za-z0-9]', ' ', text)
+    text = re.sub(r'[^\w\s\u4e00-\u9fff]', ' ', text)
+    text = re.sub(r'\s+', ' ', text)
+    return text
+
+
+@app.post("/process_audio")
+async def process_audio(file: UploadFile = File(...)):
+    with tempfile.NamedTemporaryFile(delete=False, suffix=".wav") as tmp:
+        tmp.write(await file.read())
+        tmp_path = tmp.name
+
+    # load audio
+    y = librosa.load(tmp_path, sr=16000)[0]
+    asr_result = asr_pipeline(y, generate_kwargs={"language": "mandarin"} )['text']
+    prompt = SYSTEM_PROMPT.format(asr_result)
+    output = pipe(prompt, max_new_tokens=100)[0]['generated_text'].replace("\n", " ")
+    output = output.split("麗梅的回答——")[1]
+    output = remove_punctuation_and_replace_with_space(output)
+    with open(f"tmp/llm.txt", "w") as f:
+        f.write(output)
+
+    wav_info = svs_inference(
+        config.model_path,
+        svs_model,
+        output,
+        lang=config.lang,
+        random_gen=True,
+        fs=44100
+    )
+    sf.write("tmp/response.wav", wav_info, samplerate=44100)
+
+    with open("tmp/response.wav", "rb") as f:
+        audio_bytes = f.read()
+        audio_b64 = base64.b64encode(audio_bytes).decode("utf-8")
+
+    return JSONResponse(content={
+        "asr_text": asr_result,
+        "llm_text": output,
+        "audio": audio_b64
+    })
+
+
+@app.get("/metrics")
+def on_click_metrics():
+    global predictor
+    # OWSM ctc + PER
+    y, sr = librosa.load("tmp/response.wav", sr=16000)
+    asr_result = asr_pipeline(y, generate_kwargs={"language": "mandarin"} )['text']
+    hyp_pinin = lazy_pinyin(asr_result)
+
+    with open(f"tmp/llm.txt", "r") as f:
+        ref = f.read().replace(' ', '')
+
+    ref_pinin = lazy_pinyin(ref)
+    per = jiwer.wer(" ".join(ref_pinin), " ".join(hyp_pinin))
+    
+    audio = librosa.load(f"tmp/response.wav", sr=44100)[0]
+    singmos = singmos_evaluation(
+        predictor, 
+        audio,
+        fs=44100
+    )
+    return f"""
+Phoneme Error Rate: {per}
+SingMOS: {singmos}
+"""
+
+def test_audio():
+    # load audio
+    y = librosa.load("nihao.mp3", sr=16000)[0]
+    asr_result = asr_pipeline(y, generate_kwargs={"language": "mandarin"} )['text']
+    prompt = SYSTEM_PROMPT + asr_result
+    output = pipe(prompt, max_new_tokens=100)[0]['generated_text'].replace("\n", " ")
+    output = output.split("麗梅的回答——")[1]
+    output = remove_punctuation_and_replace_with_space(output)
+    with open(f"tmp/llm.txt", "w") as f:
+        f.write(output)
+
+    wav_info = svs_inference(
+        config.model_path,
+        svs_model,
+        output,
+        lang=config.lang,
+        random_gen=True,
+        fs=44100
+    )
+    sf.write("tmp/response.wav", wav_info, samplerate=44100)
+    with open("tmp/response.wav", "rb") as f:
+        audio_bytes = f.read()
+        audio_b64 = base64.b64encode(audio_bytes).decode("utf-8")
+
+
+if __name__ == "__main__":
+    test_audio()
+
+    # start = time.time()
+    # test_audio()
+    # print(f"elapsed time: {time.time() - start}")

svs_utils.py

@@ -206,14 +206,14 @@ def svs_inference(model_name, model_svs, answer_text, lang, random_gen=True, fs=
         output_dict = svs(batch, sids=sid)
     else:
         lid = np.array([langs[lang]])
-        spk_embed = np.load("resource/singer/singer_embedding_ace-1.npy")
+        spk_embed = np.load("resource/singer/singer_embedding_ace-2.npy")
         output_dict = svs(batch, lids=lid, spembs=spk_embed)
     wav_info = output_dict["wav"].cpu().numpy()
 
     return wav_info
 
 
-def singmos_warmup(config):
+def singmos_warmup():
     predictor = torch.hub.load(
         "South-Twilight/SingMOS:v0.2.0", "singing_ssl_mos", trust_repo=True
     )

util.py

@@ -4,7 +4,6 @@ import warnings
 from typing import List
 import re
 
-import pyopenjtalk
 from resource.pinyin_dict import PINYIN_DICT
 from pypinyin import lazy_pinyin
 
@@ -66,6 +65,7 @@ def get_tokenizer(model, lang):
                     zh_plan = plan
             return lambda text: split_pinyin_ace(text, zh_plan)
     elif lang == "jp":
+        import pyopenjtalk
         return pyopenjtalk_g2p