Python SDK

Prerequisites

Before you use the SDK, familiarize yourself with the API reference. For more information, see the API reference.

Download and install

1. Download the Python SDK.

   You can download the Python SDK from Github or download the streamInputTts-github-python package directly.

2. Install SDK dependencies.

   Navigate to the SDK root directory and run the following command to install the dependencies:

   python -m pip install -r requirements.txt

3. Install the SDK.

   After the dependencies are installed, run the following command to install the SDK:

   python -m pip install .

4. After installation, import the SDK using the following code:

   # -*- coding: utf-8 -*-
   import nls

Multi-threading and concurrency

In CPython, the Global Interpreter Lock prevents multiple threads from executing Python code simultaneously. However, some performance-oriented libraries may remove this limitation. To make better use of the computing resources on a multi-core computer, you can use multiprocessing or concurrent.futures.ProcessPoolExecutor. If you want to run multiple I/O-intensive tasks concurrently, multi-threading is still a suitable model.

If a single interpreter has too many threads, switching between them increases overhead and may cause SDK errors. We recommend that you do not use more than 200 threads. You can use the multiprocessing technique or manually create multiple interpreters using a script.

Key interfaces

The nls.NlsSpeechTranscriber class is used for real-time speech recognition. Its core methods are as follows:

• Return value: A Boolean value. True indicates success. False indicates failure.

Code example

import time
import threading
import sys

import nls


URL="wss://nls-gateway-cn-shanghai.aliyuncs.com/ws/v1"
TOKEN="yourToken"  # For information about how to obtain a token, see https://help.aliyun.com/document_detail/450255.html
APPKEY="yourAppkey"    # To obtain an Appkey, go to the console: https://nls-portal.console.aliyun.com/applist

# The following code repeatedly performs real-time speech recognition (file transcription) on the content of an audio file.
class TestSt:
    def __init__(self, tid, test_file):
        self.__th = threading.Thread(target=self.__test_run)
        self.__id = tid
        self.__test_file = test_file
   
    def loadfile(self, filename):
        with open(filename, "rb") as f:
            self.__data = f.read()
    
    def start(self):
        self.loadfile(self.__test_file)
        self.__th.start()

    def test_on_sentence_begin(self, message, *args):
        print("test_on_sentence_begin:{}".format(message))

    def test_on_sentence_end(self, message, *args):
        print("test_on_sentence_end:{}".format(message))

    def test_on_start(self, message, *args):
        print("test_on_start:{}".format(message))

    def test_on_error(self, message, *args):
        print("on_error args=>{}".format(args))

    def test_on_close(self, *args):
        print("on_close: args=>{}".format(args))

    def test_on_result_chg(self, message, *args):
        print("test_on_chg:{}".format(message))

    def test_on_completed(self, message, *args):
        print("on_completed:args=>{} message=>{}".format(args, message))


    def __test_run(self):
        print("thread:{} start..".format(self.__id))
        sr = nls.NlsSpeechTranscriber(
                    url=URL,
                    token=TOKEN,
                    appkey=APPKEY,
                    on_sentence_begin=self.test_on_sentence_begin,
                    on_sentence_end=self.test_on_sentence_end,
                    on_start=self.test_on_start,
                    on_result_changed=self.test_on_result_chg,
                    on_completed=self.test_on_completed,
                    on_error=self.test_on_error,
                    on_close=self.test_on_close,
                    callback_args=[self.__id]
                )

        print("{}: session start".format(self.__id))
        r = sr.start(aformat="pcm",
        enable_intermediate_result=True,
        enable_punctuation_prediction=True,
        enable_inverse_text_normalization=True)

        self.__slices = zip(*(iter(self.__data),) * 640)
        for i in self.__slices:
            sr.send_audio(bytes(i))
            time.sleep(0.01)

        sr.ctrl(ex={"test":"tttt"})
        time.sleep(1)

        r = sr.stop()
        print("{}: sr stopped:{}".format(self.__id, r))
        time.sleep(1)

def multiruntest(num=500):
    for i in range(0, num):
        name = "thread" + str(i)
        t = TestSt(name, "tests/test1.pcm")
        t.start()

nls.enableTrace(False)
multiruntest(1)