Interaction flow and implementation

package com.alibaba.tingwu.client.demo.realtimemeeting;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.alibaba.fastjson.serializer.SerializerFeature;
import com.alibaba.nls.client.protocol.NlsClient;
import com.alibaba.nls.client.protocol.asr.SpeechTranscriber;
import com.alibaba.nls.client.protocol.asr.SpeechTranscriberListener;
import com.alibaba.nls.client.protocol.asr.SpeechTranscriberResponse;
import org.junit.BeforeClass;
import org.junit.Test;

import java.io.FileInputStream;
import java.util.Arrays;

/**
 * @author tingwu2023
 * @desc This demo shows how to call the real-time speech recognition service using the MeetingJoinUrl returned after creating a meeting in a real-time meeting scenario.
 */
public class RealtimeTransTest {

    private static NlsClient NLS_CLIENT;

    /**
     * Initialize the speech recognition SDK. The SDK can be reused and used globally.
     */
    @BeforeClass
    public static void before() {
        NLS_CLIENT = new NlsClient("default");
    }

    /**
     * Single-channel real-time transcription. In most cases, you can refer to this implementation.
     */
    @Test
    public void testRealtimeTrans() throws Exception {
        // The URL is the ingest URL returned when you create a meeting using OpenAPI. The URL is typically in the format: "wss://tingwu-realtime-cn-beijing.aliyuncs.com/api/ws/v1?mc={xxxxxxzzzzyyy}"
        String meetingJoinUrl = "Enter the MeetingJoinUrl returned when you created the real-time meeting";

        SpeechTranscriber speechTranscriber = new SpeechTranscriber(NLS_CLIENT, "default" ,createListener(), meetingJoinUrl);
        speechTranscriber.start();

        // Use a local file to simulate real-time audio stream collection in a real-world scenario.
        String localAudioFile = "nls-sample-16k.wav";
        byte[] buffer = new byte[3200];
        FileInputStream fis = new FileInputStream(localAudioFile);
        int len;
        while ((len = fis.read(buffer)) > 0) {
            // TODO: Simulate sending real-time audio data frames.
            speechTranscriber.send(Arrays.copyOf(buffer, len));
            // TODO: Simulate the audio collection interval.
            Thread.sleep(100L);
        }

        // After the audio stream ends, send a stop signal to end the real-time transcription process.
        // TODO: Note that the meeting has not ended. To end the meeting, see the StopRealtimeMeetingTaskTest process.
        speechTranscriber.stop();
        speechTranscriber.close();
    }

    public SpeechTranscriberListener createListener() {
        return new SpeechTranscriberListener() {
            @Override
            public void onMessage(String message) {
                System.out.println("onMessage " + message);
                if (message == null || message.trim().length() == 0) {
                    return;
                }
                SpeechTranscriberResponse response = JSON.parseObject(message, SpeechTranscriberResponse.class);
                if("ResultTranslated".equals(response.getName())) {
                    // Translation event output. You can handle the event here.
                    System.out.println("--- ResultTranslated ---" + JSON.toJSONString(response, SerializerFeature.PrettyFormat));
                } else {
                    // Original speech recognition event output. The parent class handles the callback.
                    super.onMessage(message);
                }
            }

            @Override
            public void onTranscriberStart(SpeechTranscriberResponse response) {
                // The task_id is very important. Note that this task_id identifies the real-time audio stream ingest and recognition process, not the meeting-level TaskId.
                System.out.println("task_id: " + response.getTaskId() + ", name: " + response.getName() + ", status: " + response.getStatus());
            }

            @Override
            public void onSentenceBegin(SpeechTranscriberResponse response) {
                System.out.println("received onSentenceBegin: " + JSON.toJSONString(response));
            }

            @Override
            public void onSentenceEnd(SpeechTranscriberResponse response) {
                // A sentence is recognized. The server intelligently segments sentences and returns this message when the end of a sentence is detected.
                System.out.println("received onSentenceEnd: " + JSON.toJSONString(response));
                System.out.println("task_id: " + response.getTaskId() +
                        ", name: " + response.getName() +
                        // A status code of "20000000" indicates that the recognition is successful.
                        ", status: " + response.getStatus() +
                        // The sentence number, which increments from 1.
                        ", index: " + response.getTransSentenceIndex() +
                        // The current recognition result.
                        ", result: " + response.getTransSentenceText() +
                        // The current word-level recognition result.
                        ", words: " + response.getWords() +
                        // The start time.
                        ", begin_time: " + response.getSentenceBeginTime() +
                        // The duration of the processed audio in milliseconds.
                        ", time: " + response.getTransSentenceTime());
                // The current recognition result (a fixed result that will not change).
                String text = response.getTransSentenceText();
                // The current recognition result (unlike response.getTransSentenceText(), this result may change).
                SpeechTranscriberResponse.StashResult stashResult = response.getStashResult();
                // Concatenate the two recognition results above.
                String stashText = stashResult == null ? "" : stashResult.getText();
                System.out.println("[onSentenceEnd] text = " + text + " | stashText = " + stashText);
            }

            @Override
            public void onTranscriptionResultChange(SpeechTranscriberResponse response) {
                // An intermediate recognition result is generated. This message is returned only when OutputLevel is set to 2.
                System.out.println("received onTranscriptionResultChange: " + JSON.toJSONString(response));
                System.out.println("task_id: " + response.getTaskId() +
                        ", name: " + response.getName() +
                        // A status code of "20000000" indicates that the recognition is successful.
                        ", status: " + response.getStatus() +
                        // The sentence number, which increments from 1.
                        ", index: " + response.getTransSentenceIndex() +
                        // The current recognition result.
                        ", result: " + response.getTransSentenceText() +
                        // The current word-level recognition result.
                        ", words: " + response.getWords() +
                        // The duration of the processed audio in milliseconds.
                        ", time: " + response.getTransSentenceTime());
            }

            @Override
            public void onTranscriptionComplete(SpeechTranscriberResponse response) {
                // The recognition ends. This event is received after you call speechTranscriber.stop().
                System.out.println("received onTranscriptionComplete: " + JSON.toJSONString(response));
            }

            @Override
            public void onFail(SpeechTranscriberResponse response) {
                // A real-time recognition error occurred. Check the error code and record this task_id for troubleshooting.
                System.out.println("received onFail: " + JSON.toJSONString(response));
            }
        };
    }

}

package main

import (
	"errors"
	"log"
	"os"
	"time"

	nls "github.com/aliyun/alibabacloud-nls-go-sdk"
)

const (
	TOKEN  = "default"
	APPKEY = "default"
)

func onTaskFailed(text string, param interface{}) {
	logger, ok := param.(*nls.NlsLogger)
	if !ok {
		return
	}

	logger.Println("TaskFailed:", text)
}

func onStarted(text string, param interface{}) {
	logger, ok := param.(*nls.NlsLogger)
	if !ok {
		return
	}

	logger.Println("onStarted:", text)
}

func onSentenceBegin(text string, param interface{}) {
	logger, ok := param.(*nls.NlsLogger)
	if !ok {
		return
	}

	logger.Println("onSentenceBegin:", text)
}

func onSentenceEnd(text string, param interface{}) {
	logger, ok := param.(*nls.NlsLogger)
	if !ok {
		return
	}

	logger.Println("onSentenceEnd:", text)
}

func onResultChanged(text string, param interface{}) {
	logger, ok := param.(*nls.NlsLogger)
	if !ok {
		return
	}

	logger.Println("onResultChanged:", text)
}

func onCompleted(text string, param interface{}) {
	logger, ok := param.(*nls.NlsLogger)
	if !ok {
		return
	}

	logger.Println("onCompleted:", text)
}

func onClose(param interface{}) {
	logger, ok := param.(*nls.NlsLogger)
	if !ok {
		return
	}

	logger.Println("onClosed:")
}

func waitReady(ch chan bool, logger *nls.NlsLogger) error {
	select {
	case done := <-ch:
		{
			if !done {
				logger.Println("Wait failed")
				return errors.New("wait failed")
			}
			logger.Println("Wait done")
		}
	case <-time.After(20 * time.Second):
		{
			logger.Println("Wait timeout")
			return errors.New("wait timeout")
		}
	}
	return nil
}

func run_push_audio_stream(url string) {
	// Use a local file in PCM or Opus format to simulate real-time audio stream collection in a real-world scenario.
	pcm, err := os.Open("tingwu-sample-16k.pcm")
	if err != nil {
		log.Default().Fatalln(err)
	}

	buffers := nls.LoadPcmInChunk(pcm, 320)
	param := nls.DefaultSpeechTranscriptionParam()
	config := nls.NewConnectionConfigWithToken(url, APPKEY, TOKEN)
	logger := nls.NewNlsLogger(os.Stderr, "1", log.LstdFlags|log.Lmicroseconds)
	logger.SetLogSil(false)
	logger.SetDebug(true)
	st, err := nls.NewSpeechTranscription(config, logger,
		onTaskFailed, onStarted,
		onSentenceBegin, onSentenceEnd, onResultChanged,
		onCompleted, onClose, logger)
	if err != nil {
		logger.Fatalln(err)
		return
	}

	logger.Println("Start pushing audio stream")
	ready, err := st.Start(param, nil)
	if err != nil {
		logger.Fatalln(err)
		return
	}

	err = waitReady(ready, logger)
	if err != nil {
		logger.Fatalln(err)
		return
	}

	for _, data := range buffers.Data {
		if data != nil {
			st.SendAudioData(data.Data)
			time.Sleep(10 * time.Millisecond)
		}
	}

	ready, err = st.Stop()
	if err != nil {
		logger.Fatalln(err)
		return
	}

	err = waitReady(ready, logger)
	if err != nil {
		logger.Fatalln(err)
		return
	}

	st.Shutdown()
	logger.Println("Push audio stream done")
}

func main() {
	// The URL is the ingest URL returned when you create a recording using OpenAPI.
	run_push_audio_stream("wss://tingwu-realtime-cn-hangzhou-pre.aliyuncs.com/api/ws/v1?mc=*********h-moSNWGZO5mq-uZzu1EQbVBABVn9y8VGzWmVcAEiLNE1idoml7JU_wr17G4dDdxwQ6jiMg8OCQCrptlCnSk4hJ9K_fVfP8ngWaYk2If*********")
}

#include <fstream>
#include <string>
#include <mutex>
#include <condition_variable>
#include <thread>
#include <chrono>
#include "nlsClient.h"
#include "nlsEvent.h"
#include "speechTranscriberRequest.h"

// Implement a simple semaphore for event synchronization. In your actual project, you can replace this with your project's semaphore implementation.
class Semaphore {
    public:
        Semaphore(int count = 0) : count_(count) {
        }

        void notify() {
            std::unique_lock<std::mutex> lock(mtx_);
            ++count_;
            cv_.notify_one();
        }

        void wait() {
            std::unique_lock<std::mutex> lock(mtx_);
            cv_.wait(lock, [this](){ return count_ > 0; });
            --count_;
        }

        // Wait with timeout
        bool wait_for(const std::chrono::milliseconds& timeout) {
            std::unique_lock<std::mutex> lock(mtx_);
            if (!cv_.wait_for(lock, timeout, [this](){ return count_ > 0; })) {
                return false; // Timeout occurred
            }
            --count_;
            return true;
        }

    private:
        std::mutex mtx_;
        std::condition_variable cv_;
        int count_;
};

Semaphore kSemaphore(0);

// All information returned by the server is sent through this callback.
// cbEvent: The callback event structure. For more information, see nlsEvent.h.
// cbParam: A custom callback parameter. The default value is NULL. You can customize this parameter as needed.
void onMessage(AlibabaNls::NlsEvent *cbEvent, void *cbParam) {
    printf("onMessage: [%s]\n", cbEvent->getAllResponse());
    int result = cbEvent->parseJsonMsg(true);
    if (result) {
        printf("onMessage: parseJsonMsg failed: [%d]", result);
        return;
    }
    switch (cbEvent->getMsgType()) {
        case AlibabaNls::NlsEvent::TaskFailed:
            // If an exception occurs during the recognition process (including start(), sendAudio(), and stop()), the SDK's internal thread reports a TaskFailed event.
            // After a TaskFailed event is reported, the SDK closes the recognition connection channel. Calling sendAudio will then return a negative value. Stop sending audio.
            printf("onTaskFailed: status code=%d, task id=%s, error message=%s\n", 
                cbEvent->getStatusCode(), 
                cbEvent->getTaskId(), 
                cbEvent->getErrorMessage());
            break;
        case AlibabaNls::NlsEvent::TranscriptionStarted:
            // After you call start(), a connection is successfully established with the cloud, and the SDK's internal thread reports a started event.
            // Notify the sending thread that start() was successful and that data can be sent.
            kSemaphore.notify();
            break;
        case AlibabaNls::NlsEvent::SentenceBegin:
            printf("##### %d, %d\n", cbEvent->getMsgType(), AlibabaNls::NlsEvent::SentenceBegin),
            // The server detects the beginning of a sentence, and the SDK's internal thread reports a SentenceBegin event.
            printf("onSentenceBegin: status code=%d, task id=%s, index=%d, time=%d\n",
                    cbEvent->getStatusCode(), cbEvent->getTaskId(),
                    cbEvent->getSentenceIndex(), // The sentence number, which increments from 1.
                    cbEvent->getSentenceTime()); // The duration of the processed audio in milliseconds.
            break;
        case AlibabaNls::NlsEvent::TranscriptionResultChanged:
            // The recognition result has changed. When the SDK receives the latest result from the cloud,
            printf("onTranscriptionResultChanged: status code=%d, task id=%s, index=%d, time=%d, result=%s\n",
                cbEvent->getStatusCode(),
                cbEvent->getTaskId(),
                cbEvent->getSentenceIndex(), // The sentence number, which increments from 1.
                cbEvent->getSentenceTime(), // The duration of the processed audio in milliseconds.
                cbEvent->getResult());    // The complete recognition result for the current sentence.
            break;
        case AlibabaNls::NlsEvent::SentenceEnd:
            // The server detects the end of a sentence, and the SDK's internal thread reports a SentenceEnd event.
            printf("onSentenceEnd: status code=%d, task id=%s, index=%d, time=%d, begin_time=%d, result=%s | %s\n",
                cbEvent->getStatusCode(),
                cbEvent->getTaskId(),
                cbEvent->getSentenceIndex(), // The sentence number, which increments from 1.
                cbEvent->getSentenceTime(), // The duration of the processed audio in milliseconds.
                cbEvent->getSentenceBeginTime(), // The time of the corresponding SentenceBegin event.
                cbEvent->getResult(), cbEvent->getStashResultText());    // The complete recognition result for the current sentence. Note that you also need to concatenate the stashResultText.
            break;
        case AlibabaNls::NlsEvent::TranscriptionCompleted:
            // When the server stops real-time audio stream recognition, the SDK's internal thread reports a Completed event.
            kSemaphore.notify();
            break;
        case AlibabaNls::NlsEvent::Close:
            // When recognition ends or an exception occurs, the connection channel is closed, and the SDK's internal thread reports a ChannelClosed event.
            // Notify the sending thread that the final recognition result has been returned and that stop() can be called.
            break;
        default:
            // Other possible events. For example, if you enable the translation feature, the translation result message is returned here for you to handle.
            printf("---otherEvent---\n");
            break;
    }
}

int main( int argc, char** argv ) {
    // The URL is the ingest URL returned when you create a meeting using OpenAPI. The URL is typically in the format: "wss://tingwu-realtime-cn-beijing.aliyuncs.com/api/ws/v1?mc={xxxxxxzzzzyyy}"
    const char* websocketUrl = "Enter the MeetingJoinUrl returned when you created the real-time meeting";
    const char* localAudioPath = "nls-sample-16k.wav";  // The path to the local audio file for testing. This audio is short, so you may not be able to test the related summary results.

    // Set the SDK output log as needed. This is optional.
     AlibabaNls::NlsClient::getInstance()->setLogConfig("log-transcriber", AlibabaNls::LogDebug, 400, 50);
    // Start the worker threads. You must call this function before creating and starting a request. This can be considered the initialization of NlsClient. If the input parameter is negative, the number of available cores in the current system is used.
     AlibabaNls::NlsClient::getInstance()->startWorkThread(1);

    // Create a SpeechTranscriberRequest object for real-time audio stream recognition.
    AlibabaNls::SpeechTranscriberRequest* request =  AlibabaNls::NlsClient::getInstance()->createTranscriberRequest();
    request->setUrl(websocketUrl);
    // Set the callback function for all information returned by the server.
    request->setOnMessage(onMessage, NULL);
    request->setEnableOnMessage(true);

    int ret = request->start();
    if(ret < 0) {
        printf("start fail, error: [%d]", ret);
         AlibabaNls::NlsClient::getInstance()->releaseTranscriberRequest(request);
        return 0;
    } else {
        // Wait for start() to return synchronously before sending more audio data.
        if (kSemaphore.wait_for(std::chrono::milliseconds(10000))) {
            printf("start success\n");
        } else {
            printf("start timeout\n");
            return -1;
        }
    }

    std::ifstream fs;
    fs.open(localAudioPath, std::ios::binary | std::ios::in);
    while (!fs.eof()) {
        const int FRAME_SIZE = 3200;
        uint8_t data[FRAME_SIZE] = {0};
        fs.read((char *) data, sizeof(uint8_t) * FRAME_SIZE); 
        size_t nlen = fs.gcount();

        // Send audio data: sendAudio is an asynchronous operation. A negative return value indicates a failure, and you must stop sending. A value greater than 0 indicates success. 
        ret = request->sendAudio(data, nlen); 
        if (ret < 0) {
            // Send failed. Exit the data sending loop.
            printf("send data fail, ret: %d.\n", ret); 
            break;
        } 

        // For audio with a 16 kHz sample rate, 3200 bytes is about 100 ms of data. For 8 kHz audio, 1600 bytes is about 100 ms of data.
        // In actual use, audio data is real-time. You do not need to use sleep to control the rate. You can send it directly. Here, we use audio data from a file for simulation, so we need to control the rate to simulate a real recording scenario.
        std::this_thread::sleep_for(std::chrono::milliseconds(200));
    } // while
    fs.close();

    request->stop();
    // Wait for the server to return the final data and end the recognition.
    kSemaphore.wait_for(std::chrono::milliseconds(10000));
    AlibabaNls::NlsClient::getInstance()->releaseTranscriberRequest(request);

    // After all tasks are complete, release nlsClient before the process exits.
    AlibabaNls::NlsClient::releaseInstance();
}

import time
import threading
import sys
import nls

class TestRealtimeMeeting:
    def __init__(self, tid, test_file, url):
        self.__th = threading.Thread(target=self.__test_run)
        self.__id = tid
        self.__test_file = test_file
        self.__url = url

    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 message=>{} args=>{}".format(message, 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_result_translated(self, message, *args):
        print("test_on_translated:{}".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))
        rm = nls.NlsRealtimeMeeting(
                    url=self.__url,
                    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_result_translated=self.test_on_result_translated,
                    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 = rm.start()

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

        time.sleep(1)

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

def multiruntest(num=1):
    for i in range(0, num):
        name = "thread" + str(i)
        t = TestRealtimeMeeting(name, "nls-sample-16k.wav",
                                "Enter the MeetingJoinUrl returned when you created the real-time meeting")
        t.start()

nls.enableTrace(False)
multiruntest(1)