Alibaba Cloud Model Studio: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  # バッファーあたりのフレーム数。


# リアルタイム音声認識コールバックを定義します。
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 pressed, stop recognition ...')
    # 認識を停止します。
    recognition.stop()
    print('Recognition stopped.')
    print(
        '[Metric] requestId: {}, first package delay ms: {}, last package delay ms: {}'
        .format(
            recognition.get_last_request_id(),
            recognition.get_first_package_delay(),
            recognition.get_last_package_delay(),
        ))
    # プログラムを強制終了します。
    sys.exit(0)


# メイン関数を定義します。
if __name__ == '__main__':
    # シンガポールリージョンと北京リージョンの API キーは異なります。API キーの取得方法については、https://www.alibabacloud.com/help/en/model-studio/get-api-key をご参照ください。
    # 環境変数を設定していない場合は、次の行を Model Studio API キーに置き換えてください：dashscope.api_key = "sk-xxx"
    dashscope.api_key = os.environ.get('DASHSCOPE_API_KEY')

    # 以下はシンガポールリージョン用の URL です。北京リージョンのモデルを使用する場合は、URL を次のように置き換えてください：wss://dashscope.aliyuncs.com/api-ws/v1/inference
    dashscope.base_websocket_api_url='wss://dashscope-intl.aliyuncs.com/api-ws/v1/inference'

    # 認識コールバックを作成します。
    callback = Callback()

    # 非同期モードで認識サービスを呼び出します。モデル、フォーマット、サンプルレートなどの認識パラメーターをカスタマイズできます。
    recognition = Recognition(
        model='fun-asr-realtime',
        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("Press 'Ctrl+C' to stop recording and recognition...")
    # 割り込み信号（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
import dashscope
from dashscope.audio.asr import *

# シンガポールリージョンと北京リージョンの API キーは異なります。API キーの取得方法については、https://www.alibabacloud.com/help/en/model-studio/get-api-key をご参照ください。
# 環境変数を設定していない場合は、次の行を Model Studio API キーに置き換えてください：dashscope.api_key = "sk-xxx"
dashscope.api_key = os.environ.get('DASHSCOPE_API_KEY')

# 以下はシンガポールリージョン用の URL です。北京リージョンのモデルを使用する場合は、URL を次のように置き換えてください：wss://dashscope.aliyuncs.com/api-ws/v1/inference
dashscope.base_websocket_api_url='wss://dashscope-intl.aliyuncs.com/api-ws/v1/inference'

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() + ' Recognition completed')  # 認識が完了しました。

    def on_error(self, result: RecognitionResult) -> None:
        print('Recognition task_id: ', result.request_id)
        print('Recognition error: ', 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 sentence end, request_id:%s, usage:%s'
                  % (result.get_request_id(), result.get_usage(sentence)))


callback = Callback()

recognition = Recognition(model='fun-asr-realtime',
                          format='wav',
                          sample_rate=16000,
                          callback=callback)

try:
    audio_data: bytes = None
    f = open("asr_example.wav", 'rb')
    if os.path.getsize("asr_example.wav"):
        # ファイル全体を一度にバッファーに読み込みます。
        file_buffer = f.read()
        f.close()
        print("Start Recognition")
        recognition.start()

        # バッファーから 3200 バイトずつデータを送信します。
        buffer_size = len(file_buffer)
        offset = 0
        chunk_size = 3200

        while offset < buffer_size:
            # 現在送信するデータチャンクのサイズを計算します。
            remaining_bytes = buffer_size - offset
            current_chunk_size = min(chunk_size, remaining_bytes)

            # バッファーから現在のデータチャンクを抽出します。
            audio_data = file_buffer[offset:offset + current_chunk_size]

            # 音声データフレームを送信します。
            recognition.send_audio_frame(audio_data)
            # オフセットを更新します。
            offset += current_chunk_size

            # リアルタイム伝送を模倣するために遅延を追加します。
            time.sleep(0.1)

        recognition.stop()
    else:
        raise Exception(
            'The provided file was empty (zero bytes long)')
except Exception as e:
    raise e

print(
    '[Metric] requestId: {}, first package delay ms: {}, last package delay ms: {}'
    .format(
        recognition.get_last_request_id(),
        recognition.get_first_package_delay(),
        recognition.get_last_package_delay(),
    ))