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

Add offline websocket srv

funasr/runtime/onnxruntime/CMakeLists.txt

@@ -38,5 +38,4 @@ if(ENABLE_GLOG)
     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 @@ target_link_libraries(funasr PUBLIC onnxruntime ${EXTRA_LIBS})
 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 @@ bool Audio::LoadPcmwav(const char* filename, int32_t* sampling_rate)
     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 @@ void Audio::Split(Model* recog_obj)
         frame_queue.push(frame);
         frame = NULL;
     }
-}
+}

funasr/runtime/websocket/CMakeLists.txt

@@ -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

@@ -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

@@ -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

@@ -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

@@ -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

@@ -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_