Python SDK

$ pip install setuptools

$ python setup.py install

//recognizer_demo.py and transcriber_demo.py are execution files for one-sentence recognition and real-time speech recognition, respectively.

//Enter the appID that you get when you purchase a service in the platform
app_id = '#####'
  
//Enter appSecret, which you get when you purchase a service in the platform
app_secret = '#####'
  
//Enter the path of the voice file to be identified. Change it to the path of the customized audio file
audio_path = '####'
  
//Input language, format see platform documentation center-Speech recognition-Development Guide
lang_type = 'ja-JP'

$ python recognizer_demo.py 
$ python transcriber_demo.py 

# -*- coding: utf-8 -*-
import os
import time
import threading
import speech_rec
from speech_rec.callbacks import SpeechRecognizerCallback
from speech_rec.parameters import DefaultParameters

token = None
expire_time = 7  # Expiration time

class Callback(SpeechRecognizerCallback):
    """
    The parameters of the constructor are not required. You can add them as needed
    The name parameter in the example can be used as the audio file name to be recognized for distinguishing in multithreading
    """

    def __init__(self, name='SpeechRecognizer'):
        self._name = name

    def started(self, message):
        print('MyCallback.OnRecognitionStarted: %s' % message)

    def result_changed(self, message):
        print('MyCallback.OnRecognitionResultChanged: file: %s, task_id: %s, payload: %s' % (
            self._name, message['header']['task_id'], message['payload']))

    def completed(self, message):
        print('MyCallback.OnRecognitionCompleted: file: %s, task_id:%s, payload:%s' % (
            self._name, message['header']['task_id'], message['payload']))

    def task_failed(self, message):
        print(message)

    def warning_info(self, message):
        print(message)

    def channel_closed(self):
        print('MyCallback.OnRecognitionChannelClosed')

def solution(client, app_id, app_secret, audio_path, lang_type, kwargs):
    """
    Recognize speech,single thread
    :param client: SpeechClient
    :param app_id: Your app_id
    :param app_secret: Your app_secret
    :param audio_path: Audio path
    :param lang_type: Language type
    """
    assert os.path.exists(audio_path), "Audio file path error, please check your audio path."
    sample_rate = kwargs.get("sample_rate", DefaultParameters.SAMPLE_RATE_16K)
    each_audio_format = kwargs.get("audio_format", DefaultParameters.MP3)
    field_ = kwargs.get("field", DefaultParameters.FIELD)

    if judging_expire_time(app_id, app_secret, expire_time):
        callback = Callback(audio_path)
        recognizer = client.create_recognizer(callback)
        recognizer.set_app_id(app_id)
        recognizer.set_token(token)
        # fixme You can customize the configuration according to the official website documentation
        payload = {
            "lang_type": lang_type,
            "format": each_audio_format,
            "field": field_,
            "sample_rate": sample_rate,
        }
        recognizer._payload.update(**payload)
        try:
            ret = recognizer.start()
            if ret < 0:
                return ret
            print('sending audio...')
            cnt = 0
            with open(audio_path, 'rb') as f:
                audio = f.read(7680)
                while audio:
                    cnt += 0.24
                    ret = recognizer.send(audio)
                    if ret < 0:
                        break
                    time.sleep(0.24)
                    audio = f.read(7680)
            recognizer.stop()
        except Exception as ee:
            print(f"send ee:{ee}")
        finally:
            recognizer.close()
    else:
        print("token expired")


def judging_expire_time(app_id, app_secret, extime):
    global token
    new_time = time.time()
    token_file = "SpeechRecognizer_token.txt"
    if not os.path.exists(token_file):
        client.get_token(app_id, app_secret, token_file)
    with open(token_file, "r", encoding="utf-8") as fr:
        token_info = eval(fr.read())
    old_time = token_info['time']
    token = token_info['token']
    flag = True
    if new_time - old_time > 60 * 60 * 24 * (extime - 1):
        flag, _ = client.get_token(app_id, app_secret, token_file)
        if flag:
            flag = True
            pass
        else:
            for i in range(7):
                flag, _ = client.get_token(app_id, app_secret, token_file)
                if flag is not None:
                    flag = True
                    break
    return flag


def channels_split_solution(audio_path, right_path, left_path, **kwargs):
    client = kwargs.get('client')
    appid = kwargs.get('app_id')
    appsecret = kwargs.get('app_secret')
    langtype = kwargs.get('lang_type')
    remove_audio = kwargs.get('rm_audio', True)
    client.auto_split_audio(audio_path, right_path, left_path)
    thread_list = []
    thread_r = threading.Thread(target=solution, args=(client, appid, appsecret, right_path, langtype, kwargs))
    thread_list.append(thread_r)
    thread_l = threading.Thread(target=solution, args=(client, appid, appsecret, left_path, langtype, kwargs))
    thread_list.append(thread_l)
    for thread in thread_list:
        thread.start()
    for thread in thread_list:
        thread.join()
    if remove_audio:
        os.remove(right_path)
        os.remove(left_path)
    pass


if __name__ == "__main__":
    client = speech_rec.SpeechClient()
    # Set the level of output log information：DEBUG、INFO、WARNING、ERROR
    client.set_log_level('INFO')
    # Type your app_id and app_secret
    app_id = ""  # your app id
    app_secret = ""  # your app secret
    audio_path = ""  # audio path
    lang_type = ""  # lang type
    field = ""  # field
    sample_rate = 16000  # sample rate [int] 16000 or 8000
    audio_format = ""  # audio format
    assert app_id and app_secret and audio_path and lang_type and field and sample_rate and audio_format, "Please check args"
    channel = client.get_audio_info(audio_path)['channel']
    # fixme This is just a simple example, please modify it according to your needs.
    multi = False
    process_num = 4
    if channel == 1:
        kwargs = {
            "field": field,
            "sample_rate": sample_rate,
            "audio_format":audio_format
        }
        solution(client, app_id, app_secret, audio_path, lang_type, kwargs)

    elif channel == 2:
        # Dual channel 8K audio solution
        channels_split_solution(audio_path=audio_path,
                                left_path=f"left.{audio_format}",
                                right_path=f"right.{audio_format}",
                                client=client,
                                app_id=app_id,
                                app_secret=app_secret,
                                lang_type=lang_type,
                                field=field,
                                sample_rate=sample_rate,
                                audio_format=audio_format)