使用ffmpeg从实时流解码（BBC Radio 4）：

python3 asr.py http://as-hls-ww-live.akamaized.net/pool_904/live/ww/bbc_radio_fourfm/bbc_radio_fourfm.isml/bbc_radio_fourfm-audio%3d96000.norewind.m3u8 --model-source=speechbrain/asr-streaming-conformer-librispeech --device=cpu -v

从文件解码：

python3 asr.py some-english-speech.wav --model-source=speechbrain/asr-streaming-conformer-librispeech --device=cpu -v

from argparse import ArgumentParser
import logging

parser = ArgumentParser()
parser.add_argument("audio_path")
parser.add_argument("--model-source", required=True)
parser.add_argument("--device", default="cpu")
parser.add_argument("--ip", default="127.0.0.1")
parser.add_argument("--port", default=9431)
parser.add_argument("--chunk-size", default=24, type=int)
parser.add_argument("--left-context-chunks", default=4, type=int)
parser.add_argument("--num-threads", default=None, type=int)
parser.add_argument("--verbose", "-v", default=False, action="store_true")
args = parser.parse_args()

if args.verbose:
    logging.getLogger().setLevel(logging.INFO)

logging.info("加载库")

from speechbrain.inference.ASR import StreamingASR
from speechbrain.utils.dynamic_chunk_training import DynChunkTrainConfig
import torch

device = args.device

if args.num_threads is not None:
    torch.set_num_threads(args.num_threads)

logging.info(f"将模型从\"{args.model_source}\"加载到设备{device}")

asr = StreamingASR.from_hparams(args.model_source, run_opts={"device": device})
config = DynChunkTrainConfig(args.chunk_size, args.left_context_chunks)

logging.info(f"从URI \"{args.audio_path}\"启动流")

for text_chunk in asr.transcribe_file_streaming(args.audio_path, config):
    print(text_chunk, flush=True, end="")

在创建展示CPU上实时流式的HuggingFace空间之前，我们希望优化模型的一些方面。

在此期间，这是一个简单的临时演示，使用Gradio的实时麦克风流功能在浏览器中实现实时ASR。

如果您运行此代码，请注意：

• 现代浏览器拒绝通过不受信任的连接（普通HTTP）流式传输麦克风输入，除非是本地主机。如果您在远程服务器上运行此代码，可以使用SSH端口转发将远程端口暴露在您的机器上。
• 在Firefox上使用Gradio流式传输似乎会导致一些问题。基于Chromium的浏览器表现更好。

运行命令：

python3 gradio-asr.py --model-source speechbrain/asr-streaming-conformer-librispeech --ip=localhost --device=cpu

from argparse import ArgumentParser
from dataclasses import dataclass
import logging

parser = ArgumentParser()
parser.add_argument("--model-source", required=True)
parser.add_argument("--device", default="cpu")
parser.add_argument("--ip", default="127.0.0.1")
parser.add_argument("--port", default=9431)
parser.add_argument("--chunk-size", default=24, type=int)
parser.add_argument("--left-context-chunks", default=4, type=int)
parser.add_argument("--num-threads", default=None, type=int)
parser.add_argument("--verbose", "-v", default=False, action="store_true")
args = parser.parse_args()

if args.verbose:
    logging.getLogger().setLevel(logging.INFO)

logging.info("加载库")

from speechbrain.inference.ASR import StreamingASR, ASRStreamingContext
from speechbrain.utils.dynamic_chunk_training import DynChunkTrainConfig
import torch
import gradio as gr
import torchaudio
import numpy as np

device = args.device

if args.num_threads is not None:
    torch.set_num_threads(args.num_threads)

logging.info(f"将模型从\"{args.model_source}\"加载到设备{device}")

asr = StreamingASR.from_hparams(args.model_source, run_opts={"device": device})
config = DynChunkTrainConfig(args.chunk_size, args.left_context_chunks)

@dataclass
class GradioStreamingContext:
    context: ASRStreamingContext
    chunk_size: int
    waveform_buffer: torch.Tensor
    decoded_text: str

def transcribe(stream, new_chunk):
    sr, y = new_chunk

    y = y.astype(np.float32)
    y = torch.tensor(y, dtype=torch.float32, device=device)
    y /= max(1, torch.max(torch.abs(y)).item())  # 按块内最大绝对值归一化并避免NaN
    if len(y.shape) > 1:
        y = torch.mean(y, dim=1)  # 降混为单声道

    # 注意：我们在块边界上对重采样器的使用不佳，可能会降低准确性。
    # 注意：我们也应绝对避免每次创建重采样器
    resampler = torchaudio.transforms.Resample(orig_freq=sr, new_freq=asr.audio_normalizer.sample_rate).to(device)
    y = resampler(y)  # 临时重采样（可能到16kHz）


    if stream is None:
        stream = GradioStreamingContext(
            context=asr.make_streaming_context(config),
            chunk_size=asr.get_chunk_size_frames(config),
            waveform_buffer=y,
            decoded_text="",
        )
    else:
        stream.waveform_buffer = torch.concat((stream.waveform_buffer, y))

    while stream.waveform_buffer.size(0) > stream.chunk_size:
        chunk = stream.waveform_buffer[:stream.chunk_size]
        stream.waveform_buffer = stream.waveform_buffer[stream.chunk_size:]

        # 假批量维度
        chunk = chunk.unsqueeze(0)

        # 转录字符串列表，大小为1因为批量大小为1
        with torch.no_grad():
            transcribed = asr.transcribe_chunk(stream.context, chunk)
        stream.decoded_text += transcribed[0]

    return stream, stream.decoded_text

# 注意：延迟似乎相对较高，可能是由于以下原因：
# https://github.com/gradio-app/gradio/issues/6526

demo = gr.Interface(
    transcribe,
    ["state", gr.Audio(sources=["microphone"], streaming=True)],
    ["state", "text"],
    live=True,
)

demo.launch(server_name=args.ip, server_port=args.port)