From 635496708826c5a77110d83d5e7e03f6e6354c72 Mon Sep 17 00:00:00 2001
From: zhifu gao <zhifu.gzf@alibaba-inc.com>
Date: 星期六, 29 四月 2023 00:57:32 +0800
Subject: [PATCH] Merge pull request #444 from zhaomingwork/add-offline-websocket-srv
---
funasr/runtime/onnxruntime/src/audio.cpp | 4
funasr/runtime/websocket/CMakeLists.txt | 64 ++++
funasr/runtime/websocket/websocketsrv.cpp | 158 +++++++++++
funasr/runtime/onnxruntime/src/CMakeLists.txt | 3
funasr/runtime/websocket/readme.md | 102 +++++++
funasr/runtime/websocket/websocketsrv.h | 93 ++++++
funasr/runtime/onnxruntime/CMakeLists.txt | 3
funasr/runtime/websocket/websocketclient.cpp | 221 +++++++++++++++
funasr/runtime/websocket/websocketmain.cpp | 157 +++++++++++
9 files changed, 800 insertions(+), 5 deletions(-)
diff --git a/funasr/runtime/onnxruntime/CMakeLists.txt b/funasr/runtime/onnxruntime/CMakeLists.txt
index 25b816f..9f6013f 100644
--- a/funasr/runtime/onnxruntime/CMakeLists.txt
+++ b/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()
\ No newline at end of file
diff --git a/funasr/runtime/onnxruntime/src/CMakeLists.txt b/funasr/runtime/onnxruntime/src/CMakeLists.txt
index 28a67b4..d33c540 100644
--- a/funasr/runtime/onnxruntime/src/CMakeLists.txt
+++ b/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)
\ No newline at end of file
diff --git a/funasr/runtime/onnxruntime/src/audio.cpp b/funasr/runtime/onnxruntime/src/audio.cpp
index d104500..8f46a4f 100644
--- a/funasr/runtime/onnxruntime/src/audio.cpp
+++ b/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;
}
-}
+}
\ No newline at end of file
diff --git a/funasr/runtime/websocket/CMakeLists.txt b/funasr/runtime/websocket/CMakeLists.txt
new file mode 100644
index 0000000..07d96d9
--- /dev/null
+++ b/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)
diff --git a/funasr/runtime/websocket/readme.md b/funasr/runtime/websocket/readme.md
new file mode 100644
index 0000000..d5cba00
--- /dev/null
+++ b/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":"涓�浜屼笁鍥涗簲鍏竷鍏節鍗佷竴浜屼笁鍥涗簲鍏竷鍏節鍗�"}
+```
+
diff --git a/funasr/runtime/websocket/websocketclient.cpp b/funasr/runtime/websocket/websocketclient.cpp
new file mode 100644
index 0000000..9ef1d5e
--- /dev/null
+++ b/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();
+ }
+}
\ No newline at end of file
diff --git a/funasr/runtime/websocket/websocketmain.cpp b/funasr/runtime/websocket/websocketmain.cpp
new file mode 100644
index 0000000..24e4269
--- /dev/null
+++ b/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;
+}
\ No newline at end of file
diff --git a/funasr/runtime/websocket/websocketsrv.cpp b/funasr/runtime/websocket/websocketsrv.cpp
new file mode 100644
index 0000000..7e54210
--- /dev/null
+++ b/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;
+ }
+}
diff --git a/funasr/runtime/websocket/websocketsrv.h b/funasr/runtime/websocket/websocketsrv.h
new file mode 100644
index 0000000..2d0c7bd
--- /dev/null
+++ b/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_
--
Gitblit v1.9.1