Alibaba Cloud Model Studio:Paraformer リアルタイム音声認識 Python SDK

マイク

import os
import signal  # キーボードイベント（「Ctrl+C」で録音を終了）を処理します。
import sys

import dashscope
import pyaudio
from dashscope.audio.asr import *

mic = None
stream = None

# 録音パラメーターの設定
sample_rate = 16000  # サンプリングレート（Hz）
channels = 1  # モノラルチャンネル
dtype = 'int16'  # データの型
format_pcm = 'pcm'  # 音声データ形式。
block_size = 3200  # バッファーあたりのフレーム数


def init_dashscope_api_key():
    """
        DashScope API キーを設定します。詳細については、以下を参照してください：
        https://github.com/aliyun/alibabacloud-bailian-speech-demo/blob/master/PREREQUISITES.md
    """

    if 'DASHSCOPE_API_KEY' in os.environ:
        dashscope.api_key = os.environ[
            'DASHSCOPE_API_KEY']  # DASHSCOPE_API_KEY 環境変数から API キーを読み込みます。
    else:
        dashscope.api_key = '<your-dashscope-api-key>'  # API キーを手動で設定します。


# リアルタイム音声認識コールバック
class Callback(RecognitionCallback):
    def on_open(self) -> None:
        global mic
        global stream
        print('RecognitionCallback open.')
        mic = pyaudio.PyAudio()
        stream = mic.open(format=pyaudio.paInt16,
                          channels=1,
                          rate=16000,
                          input=True)

    def on_close(self) -> None:
        global mic
        global stream
        print('RecognitionCallback close.')
        stream.stop_stream()
        stream.close()
        mic.terminate()
        stream = None
        mic = None

    def on_complete(self) -> None:
        print('RecognitionCallback completed.')  # 認識完了

    def on_error(self, message) -> None:
        print('RecognitionCallback task_id: ', message.request_id)
        print('RecognitionCallback error: ', message.message)
        # 実行中の音声ストリームを停止・閉じる
        if 'stream' in globals() and stream.active:
            stream.stop()
            stream.close()
        # プログラムを強制終了
        sys.exit(1)

    def on_event(self, result: RecognitionResult) -> None:
        sentence = result.get_sentence()
        if 'text' in sentence:
            print('RecognitionCallback text: ', sentence['text'])
            if RecognitionResult.is_sentence_end(sentence):
                print(
                    'RecognitionCallback sentence end, request_id:%s, usage:%s'
                    % (result.get_request_id(), result.get_usage(sentence)))


def signal_handler(sig, frame):
    print('Ctrl+C が押されました。認識を停止中...')
    # 認識の停止
    recognition.stop()
    print('認識が停止しました。')
    print(
        '[メトリック] requestId: {}, 最初のパッケージ遅延 ms: {}, 最後のパッケージ遅延 ms: {}'
        .format(
            recognition.get_last_request_id(),
            recognition.get_first_package_delay(),
            recognition.get_last_package_delay(),
        ))
    # プログラムを強制終了
    sys.exit(0)


# メイン関数
if __name__ == '__main__':
    init_dashscope_api_key()
    print('初期化中...')

    # 認識コールバックの作成
    callback = Callback()

    # 非同期で認識サービスを呼び出します。
    # モデル、フォーマット、サンプルレートなどの認識パラメーターをカスタマイズできます。
    recognition = Recognition(
        model='paraformer-realtime-v2',
        format=format_pcm,
        # 対応フォーマット：'pcm'、'wav'、'opus'、'speex'、'aac'、'amr'。詳細についてはドキュメントを参照してください。
        sample_rate=sample_rate,
        # 対応サンプルレート：8000、16000。
        semantic_punctuation_enabled=False,
        callback=callback)

    # 認識の開始
    recognition.start()

    signal.signal(signal.SIGINT, signal_handler)
    print("録音および認識を停止するには、「Ctrl+C」を押してください...")
    # マイクから音声を読み取り、認識のために送信するループ。

    while True:
        if stream:
            data = stream.read(3200, exception_on_overflow=False)
            recognition.send_audio_frame(data)
        else:
            break

    recognition.stop()

ローカルファイル

例で使用している音声ファイルは：asr_example.wav です。

import os
import time
from dashscope.audio.asr import *

# API キーを環境変数として設定していない場合は、
# 次の行のコメントを解除し、「apiKey」を実際のキーに置き換えてください。
# import dashscope
# dashscope.api_key = "apiKey"

from datetime import datetime

def get_timestamp():
    now = datetime.now()
    formatted_timestamp = now.strftime("[%Y-%m-%d %H:%M:%S.%f]")
    return formatted_timestamp

class Callback(RecognitionCallback):
    def on_complete(self) -> None:
        print(get_timestamp() + ' 認識完了')  # 認識完了

    def on_error(self, result: RecognitionResult) -> None:
        print('認識タスク ID：', result.request_id)
        print('認識エラー：', result.message)
        exit(0)

    def on_event(self, result: RecognitionResult) -> None:
        sentence = result.get_sentence()
        if 'text' in sentence:
            print(get_timestamp() + ' RecognitionCallback text: ', sentence['text'])
            if RecognitionResult.is_sentence_end(sentence):
                print(get_timestamp() + 
                    'RecognitionCallback 文末、request_id:%s, usage:%s'
                    % (result.get_request_id(), result.get_usage(sentence)))


callback = Callback()

recognition = Recognition(model='paraformer-realtime-v2',
                          format='wav',
                          sample_rate=16000,
                          # 「language_hints」パラメーターは paraformer-realtime-v2 モデルでのみサポートされています。
                          language_hints=['zh', 'en'],
                          callback=callback)

recognition.start()

try:
    audio_data: bytes = None
    f = open("asr_example.wav", 'rb')
    if os.path.getsize("asr_example.wav"):
        while True:
            audio_data = f.read(3200)
            if not audio_data:
                break
            else:
                recognition.send_audio_frame(audio_data)
            time.sleep(0.1)
    else:
        raise Exception(
            '指定されたファイルは空です（0 バイト）')
    f.close()
except Exception as e:
    raise e

recognition.stop()

print(
    '[メトリック] requestId: {}, 最初のパッケージ遅延 ms: {}, 最後のパッケージ遅延 ms: {}'
    .format(
        recognition.get_last_request_id(),
        recognition.get_first_package_delay(),
        recognition.get_last_package_delay(),
    ))