Merge pull request #444 from zhaomingwork/add-offline-websocket-srv

zhifu gao

2023-04-29 635496708826c5a77110d83d5e7e03f6e6354c72

Merge pull request #444 from zhaomingwork/add-offline-websocket-srv

Add offline websocket srv

 funasr/runtime/onnxruntime/CMakeLists.txt

@@ -38,5 +38,4 @@
    include_directories(${PROJECT_SOURCE_DIR}/third_party/glog)
    set(BUILD_TESTING OFF)
    add_subdirectory(third_party/glog)
endif()

endif()

 funasr/runtime/onnxruntime/src/CMakeLists.txt

@@ -28,5 +28,4 @@
add_executable(funasr-onnx-offline "funasr-onnx-offline.cpp")
add_executable(funasr-onnx-offline-rtf "funasr-onnx-offline-rtf.cpp")
target_link_libraries(funasr-onnx-offline PUBLIC funasr)
target_link_libraries(funasr-onnx-offline-rtf PUBLIC funasr)

target_link_libraries(funasr-onnx-offline-rtf PUBLIC funasr)

 funasr/runtime/onnxruntime/src/audio.cpp

@@ -380,8 +380,10 @@
    FILE* fp;
    fp = fopen(filename, "rb");
    if (fp == nullptr)
    {
        LOG(ERROR) << "Failed to read " << filename;
        return false;
    }
    fseek(fp, 0, SEEK_END);
    uint32_t n_file_len = ftell(fp);
    fseek(fp, 0, SEEK_SET);
@@ -517,4 +519,4 @@
        frame_queue.push(frame);
        frame = NULL;
    }
}
}

 funasr/runtime/websocket/CMakeLists.txt

New file
@@ -0,0 +1,64 @@
cmake_minimum_required(VERSION 3.10)

project(FunASRWebscoket) 

set(CMAKE_RUNTIME_OUTPUT_DIRECTORY ${CMAKE_BINARY_DIR}/bin)


option(ENABLE_WEBSOCKET "Whether to build websocket server" ON)
 
if(ENABLE_WEBSOCKET)
  cmake_policy(SET CMP0135 NEW)

  include(FetchContent)
  FetchContent_Declare(websocketpp
  GIT_REPOSITORY https://github.com/zaphoyd/websocketpp.git
    GIT_TAG 0.8.2
    SOURCE_DIR ${PROJECT_SOURCE_DIR}/third_party/websocket
    )
  
  FetchContent_MakeAvailable(websocketpp)
  include_directories(${PROJECT_SOURCE_DIR}/third_party/websocket)
   

  FetchContent_Declare(asio
     URL   https://github.com/chriskohlhoff/asio/archive/refs/tags/asio-1-24-0.tar.gz
   SOURCE_DIR ${PROJECT_SOURCE_DIR}/third_party/asio
  )
  
  FetchContent_MakeAvailable(asio)
  include_directories(${PROJECT_SOURCE_DIR}/third_party/asio/asio/include)
 
  FetchContent_Declare(json
     URL   https://github.com/nlohmann/json/archive/refs/tags/v3.11.2.tar.gz
   SOURCE_DIR ${PROJECT_SOURCE_DIR}/third_party/json
  )
  
  FetchContent_MakeAvailable(json)
  include_directories(${PROJECT_SOURCE_DIR}/third_party/json/include)
 
 

endif()

# Include generated *.pb.h files
link_directories(${ONNXRUNTIME_DIR}/lib)

include_directories(${PROJECT_SOURCE_DIR}/../onnxruntime/include/)
include_directories(${PROJECT_SOURCE_DIR}/../onnxruntime/third_party/yaml-cpp/include/)
include_directories(${PROJECT_SOURCE_DIR}/../onnxruntime/third_party/kaldi-native-fbank)

add_subdirectory(${PROJECT_SOURCE_DIR}/../onnxruntime/third_party/yaml-cpp yaml-cpp)
add_subdirectory(${PROJECT_SOURCE_DIR}/../onnxruntime/third_party/kaldi-native-fbank/kaldi-native-fbank/csrc csrc)
add_subdirectory(${PROJECT_SOURCE_DIR}/../onnxruntime/src src)

include_directories(${PROJECT_SOURCE_DIR}/../onnxruntime/third_party/glog)
set(BUILD_TESTING OFF)
add_subdirectory(${PROJECT_SOURCE_DIR}/../onnxruntime/third_party/glog glog)
 

add_executable(websocketmain "websocketmain.cpp" "websocketsrv.cpp")
add_executable(websocketclient "websocketclient.cpp")

target_link_libraries(websocketclient PUBLIC funasr)
target_link_libraries(websocketmain PUBLIC funasr)

 funasr/runtime/websocket/readme.md

New file
@@ -0,0 +1,102 @@
# ONNXRuntime-cpp for Websocket Server



## Export the model

### Install [modelscope and funasr](https://github.com/alibaba-damo-academy/FunASR#installation)



```shell

# pip3 install torch torchaudio

pip install -U modelscope funasr

# For the users in China, you could install with the command:

# pip install -U modelscope funasr -f https://modelscope.oss-cn-beijing.aliyuncs.com/releases/repo.html -i https://mirror.sjtu.edu.cn/pypi/web/simple

```



### Export [onnx model](https://github.com/alibaba-damo-academy/FunASR/tree/main/funasr/export)



```shell

python -m funasr.export.export_model --model-name damo/speech_paraformer-large_asr_nat-zh-cn-16k-common-vocab8404-pytorch --export-dir ./export --type onnx --quantize True

```



## Building for Linux/Unix



### Download onnxruntime

```shell

# download an appropriate onnxruntime from https://github.com/microsoft/onnxruntime/releases/tag/v1.14.0

# here we get a copy of onnxruntime for linux 64

wget https://github.com/microsoft/onnxruntime/releases/download/v1.14.0/onnxruntime-linux-x64-1.14.0.tgz

tar -zxvf onnxruntime-linux-x64-1.14.0.tgz

```



### Install openblas

```shell

sudo apt-get install libopenblas-dev #ubuntu

# sudo yum -y install openblas-devel #centos

```



### Build runtime

```shell

git clone https://github.com/alibaba-damo-academy/FunASR.git && cd funasr/runtime/websocket

mkdir build && cd build

cmake  -DCMAKE_BUILD_TYPE=release .. -DONNXRUNTIME_DIR=/path/to/onnxruntime-linux-x64-1.14.0

make

```

## Run the websocket server



```shell

cd bin

websocketmain  [--model_thread_num <int>] [--decoder_thread_num

                        <int>] [--io_thread_num <int>] [--port <int>]

                        [--listen_ip <string>] [--wav-scp <string>]

                        [--wav-path <string>] [--punc-config <string>]

                        [--punc-model <string>] --am-config <string>

                        --am-cmvn <string> --am-model <string>

                        [--vad-config <string>] [--vad-cmvn <string>]

                        [--vad-model <string>] [--] [--version] [-h]

Where:

   --wav-scp <string>

     wave scp path

   --wav-path <string>

     wave file path



   --punc-config <string>

     punc config path

   --punc-model <string>

     punc model path



   --am-config <string>

     (required)  am config path

   --am-cmvn <string>

     (required)  am cmvn path

   --am-model <string>

     (required)  am model path



   --vad-config <string>

     vad config path

   --vad-cmvn <string>

     vad cmvn path

   --vad-model <string>

     vad model path

   --decoder_thread_num <int>

     number of threads for decoder

   --io_thread_num <int>

     number of threads for network io

  
   Required: --am-config <string> --am-cmvn <string> --am-model <string> 
   If use vad, please add: [--vad-config <string>] [--vad-cmvn <string>] [--vad-model <string>]

   If use punc, please add: [--punc-config <string>] [--punc-model <string>] 
example:

   websocketmain --am-config /FunASR/funasr/runtime/onnxruntime/export/damo/speech_paraformer-large_asr_nat-zh-cn-16k-common-vocab8404-pytorch/config.yaml --am-model /FunASR/funasr/runtime/onnxruntime/export/damo/speech_paraformer-large_asr_nat-zh-cn-16k-common-vocab8404-pytorch/model.onnx --am-cmvn /FunASR/funasr/runtime/onnxruntime/export/damo/speech_paraformer-large_asr_nat-zh-cn-16k-common-vocab8404-pytorch/am.mvn

```



## Run websocket client test



```shell

Usage: websocketclient server_ip port wav_path threads_num



example:



websocketclient 127.0.0.1 8889 funasr/runtime/websocket/test.pcm.wav 64



result json, example like:

{"text":"一二三四五六七八九十一二三四五六七八九十"}

```




 funasr/runtime/websocket/websocketclient.cpp

New file
@@ -0,0 +1,221 @@
/**
 * Copyright FunASR (https://github.com/alibaba-damo-academy/FunASR). All Rights
 * Reserved. MIT License  (https://opensource.org/licenses/MIT)
 */
/* 2022-2023 by zhaomingwork */

// client for websocket, support multiple threads
// Usage: websocketclient server_ip port wav_path threads_num

#define ASIO_STANDALONE 1
#include <websocketpp/client.hpp>
#include <websocketpp/common/thread.hpp>
#include <websocketpp/config/asio_no_tls_client.hpp>

#include "audio.h"

/**
 * Define a semi-cross platform helper method that waits/sleeps for a bit.
 */
void wait_a_bit() {
#ifdef WIN32
  Sleep(1000);
#else
  sleep(1);
#endif
}
typedef websocketpp::config::asio_client::message_type::ptr message_ptr;

class websocket_client {
 public:
  typedef websocketpp::client<websocketpp::config::asio_client> client;
  typedef websocketpp::lib::lock_guard<websocketpp::lib::mutex> scoped_lock;

  websocket_client() : m_open(false), m_done(false) {
    // set up access channels to only log interesting things
    m_client.clear_access_channels(websocketpp::log::alevel::all);
    m_client.set_access_channels(websocketpp::log::alevel::connect);
    m_client.set_access_channels(websocketpp::log::alevel::disconnect);
    m_client.set_access_channels(websocketpp::log::alevel::app);

    // Initialize the Asio transport policy
    m_client.init_asio();

    // Bind the handlers we are using
    using websocketpp::lib::bind;
    using websocketpp::lib::placeholders::_1;
    m_client.set_open_handler(bind(&websocket_client::on_open, this, _1));
    m_client.set_close_handler(bind(&websocket_client::on_close, this, _1));
    m_client.set_close_handler(bind(&websocket_client::on_close, this, _1));

    m_client.set_message_handler(
        [this](websocketpp::connection_hdl hdl, message_ptr msg) {
          on_message(hdl, msg);
        });

    m_client.set_fail_handler(bind(&websocket_client::on_fail, this, _1));
    m_client.clear_access_channels(websocketpp::log::alevel::all);
  }
  void on_message(websocketpp::connection_hdl hdl, message_ptr msg) {
    const std::string& payload = msg->get_payload();
    switch (msg->get_opcode()) {
      case websocketpp::frame::opcode::text:
        std::cout << "on_message=" << payload << std::endl;
    }
  }
  // This method will block until the connection is complete
  void run(const std::string& uri, const std::string& wav_path) {
    // Create a new connection to the given URI
    websocketpp::lib::error_code ec;
    client::connection_ptr con = m_client.get_connection(uri, ec);
    if (ec) {
      m_client.get_alog().write(websocketpp::log::alevel::app,
                                "Get Connection Error: " + ec.message());
      return;
    }
    this->wav_path = std::move(wav_path);
    // Grab a handle for this connection so we can talk to it in a thread
    // safe manor after the event loop starts.
    m_hdl = con->get_handle();

    // Queue the connection. No DNS queries or network connections will be
    // made until the io_service event loop is run.
    m_client.connect(con);

    // Create a thread to run the ASIO io_service event loop
    websocketpp::lib::thread asio_thread(&client::run, &m_client);

    send_wav_data();
    asio_thread.join();
  }

  // The open handler will signal that we are ready to start sending data
  void on_open(websocketpp::connection_hdl) {
    m_client.get_alog().write(websocketpp::log::alevel::app,
                              "Connection opened, starting data!");

    scoped_lock guard(m_lock);
    m_open = true;
  }

  // The close handler will signal that we should stop sending data
  void on_close(websocketpp::connection_hdl) {
    m_client.get_alog().write(websocketpp::log::alevel::app,
                              "Connection closed, stopping data!");

    scoped_lock guard(m_lock);
    m_done = true;
  }

  // The fail handler will signal that we should stop sending data
  void on_fail(websocketpp::connection_hdl) {
    m_client.get_alog().write(websocketpp::log::alevel::app,
                              "Connection failed, stopping data!");

    scoped_lock guard(m_lock);
    m_done = true;
  }
  // send wav to server
  void send_wav_data() {
    uint64_t count = 0;
    std::stringstream val;

    Audio audio(1);
    int32_t sampling_rate = 16000;

    if (!audio.LoadPcmwav(wav_path.c_str(), &sampling_rate)) {
      std::cout << "error in load wav" << std::endl;
      return;
    }

    float* buff;
    int len;
    int flag = 0;
    bool wait = false;
    while (1) {
      {
        scoped_lock guard(m_lock);
        // If the connection has been closed, stop generating data
        if (m_done) {
          break;
        }

        // If the connection hasn't been opened yet wait a bit and retry
        if (!m_open) {
          wait = true;
        } else {
          break;
        }
      }

      if (wait) {
        std::cout << "wait.." << m_open << std::endl;
        wait_a_bit();

        continue;
      }
    }
    websocketpp::lib::error_code ec;
    // fetch wav data use asr engine api
    while (audio.Fetch(buff, len, flag) > 0) {
      short iArray[len];

      // convert float -1,1 to short -32768,32767
      for (size_t i = 0; i < len; ++i) {
        iArray[i] = (short)(buff[i] * 32767);
      }
      // send data to server
      m_client.send(m_hdl, iArray, len * sizeof(short),
                    websocketpp::frame::opcode::binary, ec);
      std::cout << "sended data len=" << len * sizeof(short) << std::endl;
      // The most likely error that we will get is that the connection is
      // not in the right state. Usually this means we tried to send a
      // message to a connection that was closed or in the process of
      // closing. While many errors here can be easily recovered from,
      // in this simple example, we'll stop the data loop.
      if (ec) {
        m_client.get_alog().write(websocketpp::log::alevel::app,
                                  "Send Error: " + ec.message());
        break;
      }

      wait_a_bit();
    }

    m_client.send(m_hdl, "Done", websocketpp::frame::opcode::text, ec);
    wait_a_bit();
  }

 private:
  client m_client;
  websocketpp::connection_hdl m_hdl;
  websocketpp::lib::mutex m_lock;
  std::string wav_path;
  bool m_open;
  bool m_done;
};

int main(int argc, char* argv[]) {
  if (argc < 5) {
    printf("Usage: %s server_ip port wav_path threads_num\n", argv[0]);
    exit(-1);
  }
  std::string server_ip = argv[1];
  std::string port = argv[2];
  std::string wav_path = argv[3];
  int threads_num = atoi(argv[4]);
  std::vector<websocketpp::lib::thread> client_threads;

  std::string uri = "ws://" + server_ip + ":" + port;

  for (size_t i = 0; i < threads_num; i++) {
    client_threads.emplace_back([uri, wav_path]() {
      websocket_client c;
      c.run(uri, wav_path);
    });
  }

  for (auto& t : client_threads) {
    t.join();
  }
}

 funasr/runtime/websocket/websocketmain.cpp

New file
@@ -0,0 +1,157 @@
/**
 * Copyright FunASR (https://github.com/alibaba-damo-academy/FunASR). All Rights
 * Reserved. MIT License  (https://opensource.org/licenses/MIT)
 */
/* 2022-2023 by zhaomingwork */

// io server
// Usage:websocketmain  [--model_thread_num <int>] [--decoder_thread_num
//                        <int>] [--io_thread_num <int>] [--port <int>]
//                        [--listen_ip <string>] [--wav-scp <string>]
//                        [--wav-path <string>] [--punc-config <string>]
//                        [--punc-model <string>] --am-config <string>
//                        --am-cmvn <string> --am-model <string>
//                        [--vad-config <string>] [--vad-cmvn <string>]
//                        [--vad-model <string>] [--] [--version] [-h]
#include "websocketsrv.h"

using namespace std;
void GetValue(TCLAP::ValueArg<std::string>& value_arg, string key,
              std::map<std::string, std::string>& model_path) {
  if (value_arg.isSet()) {
    model_path.insert({key, value_arg.getValue()});
    LOG(INFO) << key << " : " << value_arg.getValue();
  }
}
int main(int argc, char* argv[]) {
  try {
    google::InitGoogleLogging(argv[0]);
    FLAGS_logtostderr = true;

    TCLAP::CmdLine cmd("websocketmain", ' ', "1.0");
    TCLAP::ValueArg<std::string> vad_model("", VAD_MODEL_PATH, "vad model path",
                                           false, "", "string");
    TCLAP::ValueArg<std::string> vad_cmvn("", VAD_CMVN_PATH, "vad cmvn path",
                                          false, "", "string");
    TCLAP::ValueArg<std::string> vad_config(
        "", VAD_CONFIG_PATH, "vad config path", false, "", "string");

    TCLAP::ValueArg<std::string> am_model("", AM_MODEL_PATH, "am model path",
                                          true, "", "string");
    TCLAP::ValueArg<std::string> am_cmvn("", AM_CMVN_PATH, "am cmvn path", true,
                                         "", "string");
    TCLAP::ValueArg<std::string> am_config("", AM_CONFIG_PATH, "am config path",
                                           true, "", "string");

    TCLAP::ValueArg<std::string> punc_model(
        "", PUNC_MODEL_PATH, "punc model path", false, "", "string");
    TCLAP::ValueArg<std::string> punc_config(
        "", PUNC_CONFIG_PATH, "punc config path", false, "", "string");

    TCLAP::ValueArg<std::string> wav_path("", WAV_PATH, "wave file path", false,
                                          "", "string");
    TCLAP::ValueArg<std::string> wav_scp("", WAV_SCP, "wave scp path", false,
                                         "", "string");

    TCLAP::ValueArg<std::string> listen_ip("", "listen_ip", "listen_ip", false,
                                           "0.0.0.0", "string");
    TCLAP::ValueArg<int> port("", "port", "port", false, 8889, "int");
    TCLAP::ValueArg<int> io_thread_num("", "io_thread_num", "io_thread_num",
                                       false, 8, "int");
    TCLAP::ValueArg<int> decoder_thread_num(
        "", "decoder_thread_num", "decoder_thread_num", false, 8, "int");
    TCLAP::ValueArg<int> model_thread_num("", "model_thread_num",
                                          "model_thread_num", false, 1, "int");

    cmd.add(vad_model);
    cmd.add(vad_cmvn);
    cmd.add(vad_config);
    cmd.add(am_model);
    cmd.add(am_cmvn);
    cmd.add(am_config);
    cmd.add(punc_model);
    cmd.add(punc_config);
    cmd.add(wav_path);
    cmd.add(wav_scp);
    cmd.add(listen_ip);
    cmd.add(port);
    cmd.add(io_thread_num);
    cmd.add(decoder_thread_num);
    cmd.add(model_thread_num);
    cmd.parse(argc, argv);

    std::map<std::string, std::string> model_path;
    GetValue(vad_model, VAD_MODEL_PATH, model_path);
    GetValue(vad_cmvn, VAD_CMVN_PATH, model_path);
    GetValue(vad_config, VAD_CONFIG_PATH, model_path);
    GetValue(am_model, AM_MODEL_PATH, model_path);
    GetValue(am_cmvn, AM_CMVN_PATH, model_path);
    GetValue(am_config, AM_CONFIG_PATH, model_path);
    GetValue(punc_model, PUNC_MODEL_PATH, model_path);
    GetValue(punc_config, PUNC_CONFIG_PATH, model_path);
    GetValue(wav_path, WAV_PATH, model_path);
    GetValue(wav_scp, WAV_SCP, model_path);


    std::string s_listen_ip = listen_ip.getValue();
    int s_port = port.getValue();
    int s_io_thread_num = io_thread_num.getValue();
    int s_decoder_thread_num = decoder_thread_num.getValue();

    int s_model_thread_num = model_thread_num.getValue();

 
    asio::io_context io_decoder;  // context for decoding

    std::vector<std::thread> decoder_threads;

    auto conn_guard = asio::make_work_guard(
        io_decoder);  // make sure threads can wait in the queue

    // create threads pool
    for (int32_t i = 0; i < s_decoder_thread_num; ++i) {
      decoder_threads.emplace_back([&io_decoder]() { io_decoder.run(); });
    }

    server server_;       // server for websocket
    server_.init_asio();  // init asio
    server_.set_reuse_addr(
        true);  // reuse address as we create multiple threads

    // list on port for accept
    server_.listen(asio::ip::address::from_string(s_listen_ip), s_port);

    WebSocketServer websocket_srv(io_decoder,
                                  &server_);  // websocket server for asr engine
    websocket_srv.initAsr(model_path, s_model_thread_num);  // init asr model
    std::cout << "asr model init finished. listen on port:" << s_port
              << std::endl;

    // Start the ASIO network io_service run loop
    if (s_io_thread_num == 1) {
      server_.run();
    } else {
      typedef websocketpp::lib::shared_ptr<websocketpp::lib::thread> thread_ptr;
      std::vector<thread_ptr> ts;
      // create threads for io network
      for (size_t i = 0; i < s_io_thread_num; i++) {
        ts.push_back(websocketpp::lib::make_shared<websocketpp::lib::thread>(
            &server::run, &server_));
      }
      // wait for theads
      for (size_t i = 0; i < s_io_thread_num; i++) {
        ts[i]->join();
      }
    }

    // wait for theads
    for (auto& t : decoder_threads) {
      t.join();
    }

  } catch (std::exception const& e) {
    std::cerr << "Error: " << e.what() << std::endl;
  }

  return 0;
}

 funasr/runtime/websocket/websocketsrv.cpp

New file
@@ -0,0 +1,158 @@
/**

 * Copyright FunASR (https://github.com/alibaba-damo-academy/FunASR). All Rights

 * Reserved. MIT License  (https://opensource.org/licenses/MIT)

 */

/* 2022-2023 by zhaomingwork */



// websocket server for asr engine

// take some ideas from https://github.com/k2-fsa/sherpa-onnx

// online-websocket-server-impl.cc, thanks. The websocket server has two threads

// pools, one for handle network data and one for asr decoder.

// now only support offline engine.



#include "websocketsrv.h"



#include <thread>

#include <utility>

#include <vector>



// feed buffer to asr engine for decoder

void WebSocketServer::do_decoder(const std::vector<char>& buffer,

                                 websocketpp::connection_hdl& hdl) {

  try {

    int num_samples = buffer.size();  // the size of the buf



    if (!buffer.empty()) {

      // fout.write(buffer.data(), buffer.size());

      // feed data to asr engine

      FUNASR_RESULT Result = FunASRRecogPCMBuffer(

          asr_hanlde, buffer.data(), buffer.size(), 16000, RASR_NONE, NULL);



      std::string asr_result =

          ((FUNASR_RECOG_RESULT*)Result)->msg;  // get decode result



      websocketpp::lib::error_code ec;

      nlohmann::json jsonresult;        // result json

      jsonresult["text"] = asr_result;  // put result in 'text'



      // send the json to client

      server_->send(hdl, jsonresult.dump(), websocketpp::frame::opcode::text,

                    ec);



      std::cout << "buffer.size=" << buffer.size()

                << ",result json=" << jsonresult.dump() << std::endl;

      if (!isonline) {

        //  close the client if it is not online asr

        server_->close(hdl, websocketpp::close::status::normal, "DONE", ec);

        // fout.close();

      }

    }



  } catch (std::exception const& e) {

    std::cerr << "Error: " << e.what() << std::endl;

  }

}



void WebSocketServer::on_open(websocketpp::connection_hdl hdl) {

  scoped_lock guard(m_lock);     // for threads safty

  check_and_clean_connection();  // remove closed connection

  sample_map.emplace(

      hdl, std::make_shared<std::vector<char>>());  // put a new data vector for

                                                    // new connection

  std::cout << "on_open, active connections: " << sample_map.size()

            << std::endl;

}



void WebSocketServer::on_close(websocketpp::connection_hdl hdl) {

  scoped_lock guard(m_lock);

  sample_map.erase(hdl);  // remove data vector when  connection is closed

  std::cout << "on_close, active connections: " << sample_map.size()

            << std::endl;

}



// remove closed connection

void WebSocketServer::check_and_clean_connection() {

  std::vector<websocketpp::connection_hdl> to_remove;  // remove list

  auto iter = sample_map.begin();

  while (iter != sample_map.end()) {  // loop to find closed connection

    websocketpp::connection_hdl hdl = iter->first;

    server::connection_ptr con = server_->get_con_from_hdl(hdl);

    if (con->get_state() != 1) {  // session::state::open ==1

      to_remove.push_back(hdl);

    }

    iter++;

  }

  for (auto hdl : to_remove) {

    sample_map.erase(hdl);

    std::cout << "remove one connection " << std::endl;

  }

}

void WebSocketServer::on_message(websocketpp::connection_hdl hdl,

                                 message_ptr msg) {

  unique_lock lock(m_lock);

  // find the sample data vector according to one connection

  std::shared_ptr<std::vector<char>> sample_data_p = nullptr;



  auto it = sample_map.find(hdl);

  if (it != sample_map.end()) {

    sample_data_p = it->second;

  }

  lock.unlock();

  if (sample_data_p == nullptr) {

    std::cout << "error when fetch sample data vector" << std::endl;

    return;

  }



  const std::string& payload = msg->get_payload();  // get msg type



  switch (msg->get_opcode()) {

    case websocketpp::frame::opcode::text:

      if (payload == "Done") {

        std::cout << "client done" << std::endl;



        if (isonline) {

          // do_close(ws);

        } else {

          // for offline, send all receive data to decoder engine

          asio::post(io_decoder_, std::bind(&WebSocketServer::do_decoder, this,

                                            std::move(*(sample_data_p.get())),

                                            std::move(hdl)));

        }

      }

      break;

    case websocketpp::frame::opcode::binary: {

      // recived binary data

      const auto* pcm_data = static_cast<const char*>(payload.data());

      int32_t num_samples = payload.size();



      if (isonline) {

        // if online TODO(zhaoming) still not done

        std::vector<char> s(pcm_data, pcm_data + num_samples);

        asio::post(io_decoder_, std::bind(&WebSocketServer::do_decoder, this,

                                          std::move(s), std::move(hdl)));

      } else {

        // for offline, we add receive data to end of the sample data vector

        sample_data_p->insert(sample_data_p->end(), pcm_data,

                              pcm_data + num_samples);

      }



      break;

    }

    default:

      break;

  }

}



// init asr model

void WebSocketServer::initAsr(std::map<std::string, std::string>& model_path,

                              int thread_num) {

  try {

    // init model with api



    asr_hanlde = FunASRInit(model_path, thread_num);

    std::cout << "model ready" << std::endl;



  } catch (const std::exception& e) {

    std::cout << e.what() << std::endl;

  }

}


 funasr/runtime/websocket/websocketsrv.h

New file
@@ -0,0 +1,93 @@
/**
 * Copyright FunASR (https://github.com/alibaba-damo-academy/FunASR). All Rights
 * Reserved. MIT License  (https://opensource.org/licenses/MIT)
 */
/* 2022-2023 by zhaomingwork */

// websocket server for asr engine
// take some ideas from https://github.com/k2-fsa/sherpa-onnx
// online-websocket-server-impl.cc, thanks. The websocket server has two threads
// pools, one for handle network data and one for asr decoder.
// now only support offline engine.

#ifndef WEBSOCKETSRV_SERVER_H_
#define WEBSOCKETSRV_SERVER_H_

#include <iostream>
#include <map>
#include <memory>
#include <string>
#include <thread>
#include <utility>
#define ASIO_STANDALONE 1  // not boost
#include <glog/logging.h>

#include <fstream>
#include <functional>
#include <websocketpp/common/thread.hpp>
#include <websocketpp/config/asio_no_tls.hpp>
#include <websocketpp/server.hpp>

#include "asio.hpp"
#include "com-define.h"
#include "libfunasrapi.h"
#include "nlohmann/json.hpp"
#include "tclap/CmdLine.h"
typedef websocketpp::server<websocketpp::config::asio> server;
typedef server::message_ptr message_ptr;
using websocketpp::lib::bind;
using websocketpp::lib::placeholders::_1;
using websocketpp::lib::placeholders::_2;
typedef websocketpp::lib::lock_guard<websocketpp::lib::mutex> scoped_lock;
typedef websocketpp::lib::unique_lock<websocketpp::lib::mutex> unique_lock;

typedef struct {
  std::string msg;
  float snippet_time;
} FUNASR_RECOG_RESULT;

class WebSocketServer {
 public:
  WebSocketServer(asio::io_context& io_decoder, server* server_)
      : io_decoder_(io_decoder), server_(server_) {
    // set message handle
    server_->set_message_handler(
        [this](websocketpp::connection_hdl hdl, message_ptr msg) {
          on_message(hdl, msg);
        });
    // set open handle
    server_->set_open_handler(
        [this](websocketpp::connection_hdl hdl) { on_open(hdl); });
    // set close handle
    server_->set_close_handler(
        [this](websocketpp::connection_hdl hdl) { on_close(hdl); });
    // begin accept
    server_->start_accept();
    // not print log
    server_->clear_access_channels(websocketpp::log::alevel::all);
  }
  void do_decoder(const std::vector<char>& buffer,
                  websocketpp::connection_hdl& hdl);

  void initAsr(std::map<std::string, std::string>& model_path, int thread_num);
  void on_message(websocketpp::connection_hdl hdl, message_ptr msg);
  void on_open(websocketpp::connection_hdl hdl);
  void on_close(websocketpp::connection_hdl hdl);

 private:
  void check_and_clean_connection();
  asio::io_context& io_decoder_;  // threads for asr decoder
  // std::ofstream fout;
  FUNASR_HANDLE asr_hanlde;  // asr engine handle
  bool isonline = false;  // online or offline engine, now only support offline
  server* server_;        // websocket server

  // use map to keep the received samples data from one connection in offline
  // engine. if for online engline, a data struct is needed(TODO)
  std::map<websocketpp::connection_hdl, std::shared_ptr<std::vector<char>>,
           std::owner_less<websocketpp::connection_hdl>>
      sample_map;
  websocketpp::lib::mutex m_lock;  // mutex for sample_map
};

#endif  // WEBSOCKETSRV_SERVER_H_