Blame for app.py

"""Offlisten 在线演示：把一段音频转成带标点的文字，并报出 RTF。

这不是产品。产品是本地运行的桌面端（WASAPI 采集、不联网）；这里跑的是同一个
SenseVoice int8 模型在 Linux CPU 上的离线转写，用来证明模型与引擎的可用性。

设计约束：
- 免费 CPU Basic 档是 2 vCPU / 16 GB / 50 GB 非持久磁盘，容器会休眠后重建，
  所以模型不写进镜像，改为运行时从 Hub 拉取并缓存在容器内。
- 只接受 16 kHz 单声道交给模型；其它采样率在本地重采样，不依赖 ffmpeg。
- 所有失败都给出能照做的下一步，不返回 500。
"""

import os
import time

import gradio as gr
import numpy as np
import sherpa_onnx

MODEL_REPO = os.environ.get("OFFLISTEN_MODEL_REPO", "ToothpasteLake/offlisten-models")
MODEL_SUBDIR = "sherpa-onnx-sense-voice-zh-en-ja-ko-yue-2024-07-17"
TARGET_SR = 16000

_state = {"recognizer": None, "error": None, "lang": None}


def _find_model_files():
    """返回 (model_path, tokens_path)。优先用镜像内已有的目录，其次从 Hub 拉。"""
    local_dir = os.environ.get("OFFLISTEN_MODEL_DIR")
    if local_dir:
        m = os.path.join(local_dir, "model.int8.onnx")
        t = os.path.join(local_dir, "tokens.txt")
        if os.path.isfile(m) and os.path.isfile(t):
            return m, t
        raise RuntimeError("OFFLISTEN_MODEL_DIR 指向 %s，但里面没有 model.int8.onnx / tokens.txt" % local_dir)

    from huggingface_hub import hf_hub_download

    m = hf_hub_download(MODEL_REPO, "%s/model.int8.onnx" % MODEL_SUBDIR)
    t = hf_hub_download(MODEL_REPO, "%s/tokens.txt" % MODEL_SUBDIR)
    return m, t


def get_recognizer(language):
    """按语言惰性建识别器；同语言复用，避免每次请求都加载 229 MB。"""
    if _state["recognizer"] is not None and _state["lang"] == language:
        return _state["recognizer"]
    if _state["error"]:
        raise RuntimeError(_state["error"])
    try:
        model, tokens = _find_model_files()
        _state["recognizer"] = sherpa_onnx.OfflineRecognizer.from_sense_voice(
            model=model,
            tokens=tokens,
            language=language,
            use_itn=True,
            num_threads=int(os.environ.get("OFFLISTEN_THREADS", "2")),
        )
        _state["lang"] = language
        return _state["recognizer"]
    except Exception as e:
        _state["error"] = "%s: %s" % (type(e).__name__, e)
        raise


def _load_audio(path):
    """读成 float32 单声道 + 16 kHz。不依赖 ffmpeg。"""
    import soundfile as sf

    samples, sr = sf.read(path, dtype="float32", always_2d=True)
    mono = samples.mean(axis=1)
    if sr != TARGET_SR and len(mono) > 1:
        n = int(round(len(mono) * TARGET_SR / float(sr)))
        mono = np.interp(
            np.linspace(0.0, len(mono) - 1, n), np.arange(len(mono)), mono
        ).astype(np.float32)
    return mono, TARGET_SR


def transcribe(audio_path, language):
    if not audio_path:
        return "", "先上传一段音频，或直接用麦克风录一句。"

    try:
        samples, sr = _load_audio(audio_path)
    except Exception as e:
        return "", (
            "读不了这个文件（%s）。请换成 wav / flac / ogg / mp3；"
            "如果是从会议软件导出的 m4a，先用任意工具转成 wav。" % e
        )

    duration = len(samples) / float(sr)
    if duration < 0.2:
        return "", "音频只有 %.2f 秒，太短了。" % duration

    try:
        recognizer = get_recognizer(language)
    except Exception as e:
        return "", (
            "模型没加载起来：%s\n\n"
            "常见原因：模型仓库还没推上去（`%s`），或该仓库是私有的而 Space 没有读权限。"
            % (e, MODEL_REPO)
        )

    started = time.perf_counter()
    stream = recognizer.create_stream()
    stream.accept_waveform(sr, samples)
    recognizer.decode_stream(stream)
    elapsed = time.perf_counter() - started

    text = (stream.result.text or "").strip()
    if not text:
        return "", "这段音频里没识别出语音。确认它不是纯音乐或空白录音。"

    rtf = elapsed / duration if duration > 0 else float("nan")
    note = "音频 %.1f 秒，转写耗时 %.2f 秒，**RTF %.3f**（CPU Basic，2 vCPU，%s 线程）" % (
        duration,
        elapsed,
        rtf,
        os.environ.get("OFFLISTEN_THREADS", "2"),
    )
    return text, note


INTRO = """\
## 这是模型能力演示，不是产品

产品 **Offlisten** 在本地跑：抓系统声音、写本地文件、全程不联网。
这里把同一份 SenseVoice int8 模型放在服务器 CPU 上，只有一个目的——让你先听到效果，再决定要不要装。

- 模型：SenseVoice int8（中 / 英 / 日 / 韩 / 粤），229 MiB，自带标点与逆文本正则
- 语言选 `auto` 由模型自判；已知语种时指定会更稳
- 上传的音频只用于本次转写，不留存
"""


def build():
    with gr.Blocks(title="Offlisten ASR demo") as demo:
        gr.Markdown(INTRO)
        with gr.Row():
            with gr.Column():
                audio = gr.Audio(
                    sources=["upload", "microphone"],
                    type="filepath",
                    label="音频（wav / flac / ogg / mp3，任意采样率）",
                )
                language = gr.Dropdown(
                    ["auto", "zh", "en", "ja", "ko", "yue"],
                    value="auto",
                    label="语言",
                )
                run = gr.Button("转写", variant="primary")
            with gr.Column():
                text = gr.Textbox(label="转写结果", lines=10, show_copy_button=True)
                note = gr.Markdown()
        run.click(transcribe, inputs=[audio, language], outputs=[text, note])
        audio.change(transcribe, inputs=[audio, language], outputs=[text, note])
    return demo


if __name__ == "__main__":
    build().launch()