ai-experiment/stt_runner.py

import sys
from pathlib import Path
import sherpa_onnx, soundfile as sf
from core import argsroom as ap
from config.model import CONV_FRONTEND, ENCODER, DECODER, TOKENIZER
from config.asr import LANGUAGE, HOTWORDS, NUM_THREADS, SAMPLE_RATE, FEATURE_DIM, PROVIDER, MAX_TOTAL_LEN, MAX_NEW_TOKENS

def stt_run(args):

    print("Recognize...")
    recognizer = sherpa_onnx.OfflineRecognizer.from_qwen3_asr( # qwen3 asr
        conv_frontend   = str(CONV_FRONTEND),
        encoder         = str(ENCODER),
        decoder         = str(DECODER),
        tokenizer       = str(TOKENIZER),
        hotwords        = HOTWORDS,
        num_threads     = NUM_THREADS,
        sample_rate     = SAMPLE_RATE,
        feature_dim     = FEATURE_DIM,
        provider        = PROVIDER,
        max_total_len   = MAX_TOTAL_LEN,
        max_new_tokens  = MAX_NEW_TOKENS,
    )
    print("Recognizer ready!")

    for f in args.sounds: # Multi-file
        if not Path(f).is_file():
            print(f"Skip. file not found: {f}", file=sys.stderr)
            continue

        audio, sr  = sf.read(f, dtype="float32", always_2d=True)
        audio      = audio[:, 0]

        stream     = recognizer.create_stream()
        
        language   = args.language if args.language is not None else LANGUAGE
        if language:
            stream.set_option("language", language)
        stream.accept_waveform(sr, audio)

        recognizer.decode_stream(stream) # Inference execution for `stream.result`
        
        text       = stream.result.text

        if "<asr_text>" in text: # qwen3 asr format
            text = text.split("<asr_text>", 1)[1]

        print()
        print(f"File      : {f}")
        print(f"Duration  : {len(audio) / sr:.2f} s")
        print(f"Result    : {text}")
        print()

if __name__ == "__main__":
    stt_run( ap.parser.parse_args() )