From 0ceff540daf48b1b47a5e679623ba3756e5837aa Mon Sep 17 00:00:00 2001 From: dianjixz <18637716021@163.com> Date: Fri, 22 Nov 2024 13:50:43 +0800 Subject: [PATCH] [update] change StackFlow component msg --- .../StackFlow/stackflow/StackFlow.cpp | 69 +++++++------------ .../StackFlow/stackflow/StackFlow.h | 2 +- .../StackFlow/stackflow/StackFlowUtil.cpp | 9 +++ .../StackFlow/stackflow/StackFlowUtil.h | 14 ++++ .../llm_framework/main_audio/src/main.cpp | 21 ++++-- projects/llm_framework/main_kws/src/main.cpp | 2 +- .../main_sys/src/remote_action.cpp | 12 ++-- .../main_sys/src/remote_server.cpp | 24 +++---- projects/llm_framework/setup.ini | 2 +- 9 files changed, 80 insertions(+), 75 deletions(-) diff --git a/ext_components/StackFlow/stackflow/StackFlow.cpp b/ext_components/StackFlow/stackflow/StackFlow.cpp index 862d26c..7fbc6da 100644 --- a/ext_components/StackFlow/stackflow/StackFlow.cpp +++ b/ext_components/StackFlow/stackflow/StackFlow.cpp @@ -10,6 +10,12 @@ using namespace StackFlows; std::string llm_channel_obj::uart_push_url; +// Deprecated +#define RPC_PARSE_TO_PARAM_OLD(obj) \ + sample_json_str_get(obj, "zmq_com"), sample_unescapeString(sample_json_str_get(obj, "raw_data")) + +#define RPC_PARSE_TO_PARAM(obj) RPC_PARSE_TO_FIRST(obj), RPC_PARSE_TO_SECOND(obj) + llm_channel_obj::llm_channel_obj(const std::string &_publisher_url, const std::string &inference_url, const std::string &unit_name) : unit_name_(unit_name), inference_url_(inference_url) @@ -47,9 +53,7 @@ int llm_channel_obj::subscriber_work_id(const std::string &work_id, // std::string part1 = matches[1].str(); id_num = std::stoi(matches[2].str()); std::string input_url_name = work_id + ".out_port"; - std::string input_url; - pzmq _call("sys"); - _call.call_rpc_action("sql_select", input_url_name, [&](const std::string &data) { input_url = data; }); + std::string input_url = unit_call("sys", "sql_select", input_url_name); if (input_url.empty()) { return -1; } @@ -217,9 +221,7 @@ void StackFlow::_sys_init(const std::string &zmq_url, const std::string &data) std::string StackFlow::_rpc_setup(const std::string &data) { - SLOGI("_rpc_setup:%s", data.c_str()); - event_queue_.enqueue(EVENT_SETUP, sample_json_str_get(data, "zmq_com"), - sample_unescapeString(sample_json_str_get(data, "raw_data"))); + event_queue_.enqueue(EVENT_SETUP, RPC_PARSE_TO_PARAM(data)); return std::string("None"); } @@ -253,8 +255,7 @@ int StackFlow::setup(const std::string &work_id, const std::string &object, cons std::string StackFlow::_rpc_link(const std::string &data) { - event_queue_.enqueue(EVENT_LINK, sample_json_str_get(data, "zmq_com"), - sample_unescapeString(sample_json_str_get(data, "raw_data"))); + event_queue_.enqueue(EVENT_LINK, RPC_PARSE_TO_PARAM(data)); return std::string("None"); } @@ -285,8 +286,7 @@ void StackFlow::link(const std::string &work_id, const std::string &object, cons std::string StackFlow::_rpc_unlink(const std::string &data) { - event_queue_.enqueue(EVENT_UNLINK, sample_json_str_get(data, "zmq_com"), - sample_unescapeString(sample_json_str_get(data, "raw_data"))); + event_queue_.enqueue(EVENT_UNLINK, RPC_PARSE_TO_PARAM(data)); return std::string("None"); } @@ -317,8 +317,7 @@ void StackFlow::unlink(const std::string &work_id, const std::string &object, co std::string StackFlow::_rpc_work(const std::string &data) { - event_queue_.enqueue(EVENT_WORK, sample_json_str_get(data, "zmq_com"), - sample_unescapeString(sample_json_str_get(data, "raw_data"))); + event_queue_.enqueue(EVENT_WORK, RPC_PARSE_TO_PARAM(data)); return std::string("None"); } @@ -349,8 +348,7 @@ void StackFlow::work(const std::string &work_id, const std::string &object, cons std::string StackFlow::_rpc_exit(const std::string &data) { - event_queue_.enqueue(EVENT_EXIT, sample_json_str_get(data, "zmq_com"), - sample_unescapeString(sample_json_str_get(data, "raw_data"))); + event_queue_.enqueue(EVENT_EXIT, RPC_PARSE_TO_PARAM(data)); return std::string("None"); } @@ -384,8 +382,7 @@ int StackFlow::exit(const std::string &work_id, const std::string &object, const std::string StackFlow::_rpc_pause(const std::string &data) { - event_queue_.enqueue(EVENT_PAUSE, sample_json_str_get(data, "zmq_com"), - sample_unescapeString(sample_json_str_get(data, "raw_data"))); + event_queue_.enqueue(EVENT_PAUSE, RPC_PARSE_TO_PARAM(data)); return std::string("None"); } @@ -416,8 +413,7 @@ void StackFlow::pause(const std::string &work_id, const std::string &object, con std::string StackFlow::_rpc_taskinfo(const std::string &data) { - event_queue_.enqueue(EVENT_TASKINFO, sample_json_str_get(data, "zmq_com"), - sample_unescapeString(sample_json_str_get(data, "raw_data"))); + event_queue_.enqueue(EVENT_TASKINFO, RPC_PARSE_TO_PARAM(data)); return std::string("None"); } @@ -449,15 +445,13 @@ void StackFlow::taskinfo(const std::string &work_id, const std::string &object, int StackFlow::sys_register_unit(const std::string &unit_name) { int work_id_number; - std::string out_port; - std::string inference_port; + std::string component_msg = unit_call("sys", "register_unit", unit_name); + std::string str_port = RPC_PARSE_TO_FIRST(component_msg); + work_id_number = std::stoi(str_port); + std::string tmp_buf = RPC_PARSE_TO_SECOND(component_msg); + std::string out_port = RPC_PARSE_TO_FIRST(tmp_buf); + std::string inference_port = RPC_PARSE_TO_SECOND(tmp_buf); - pzmq _call("sys"); - _call.call_rpc_action("register_unit", unit_name, [&](const std::string &unit_info) { - work_id_number = std::stoi(sample_json_str_get(unit_info, "work_id_number")); - out_port = sample_json_str_get(unit_info, "out_port"); - inference_port = sample_json_str_get(unit_info, "inference_port"); - }); SLOGI("work_id_number:%d, out_port:%s, inference_port:%s ", work_id_number, out_port.c_str(), inference_port.c_str()); llm_task_channel_[work_id_number] = std::make_shared(out_port, inference_port, unit_name_); @@ -475,8 +469,7 @@ bool StackFlow::sys_release_unit(int work_id_num, const std::string &work_id) _work_id = work_id; _work_id_num = sample_get_work_id_num(work_id); } - pzmq _call("sys"); - _call.call_rpc_action("release_unit", _work_id, [&](const std::string &unit_info) {}); + unit_call("sys", "release_unit", _work_id); llm_task_channel_[_work_id_num].reset(); llm_task_channel_.erase(_work_id_num); SLOGI("release work_id %s success", _work_id.c_str()); @@ -485,10 +478,7 @@ bool StackFlow::sys_release_unit(int work_id_num, const std::string &work_id) std::string StackFlow::sys_sql_select(const std::string &key) { - std::string val; - pzmq _call("sys"); - _call.call_rpc_action("sql_select", key, [&val](const std::string &data) { val = data; }); - return val; + return sample_unescapeString(unit_call("sys", "sql_select", key)); } void StackFlow::sys_sql_set(const std::string &key, const std::string &val) @@ -496,23 +486,12 @@ void StackFlow::sys_sql_set(const std::string &key, const std::string &val) nlohmann::json out_body; out_body["key"] = key; out_body["val"] = val; - pzmq _call("sys"); - _call.call_rpc_action("sql_set", out_body.dump(), [](const std::string &data) {}); + unit_call("sys", "sql_set", out_body.dump()); } void StackFlow::sys_sql_unset(const std::string &key) { - std::string val; - pzmq _call("sys"); - _call.call_rpc_action("sql_unset", key, [](const std::string &data) {}); -} - -std::string StackFlow::unit_call(const std::string &unit_name, const std::string &unit_action, const std::string &data) -{ - std::string value; - pzmq _call(unit_name); - _call.call_rpc_action(unit_action, data, [&value](const std::string &raw) { value = raw; }); - return value; + unit_call("sys", "sql_unset", key); } void StackFlow::_repeat_loop(const std::string &action, const std::string &ms) diff --git a/ext_components/StackFlow/stackflow/StackFlow.h b/ext_components/StackFlow/stackflow/StackFlow.h index 6cff463..6f7d397 100644 --- a/ext_components/StackFlow/stackflow/StackFlow.h +++ b/ext_components/StackFlow/stackflow/StackFlow.h @@ -185,6 +185,7 @@ public: class StackFlow { private: std::atomic_int work_id_num_cout_; + protected: std::string unit_name_; typedef enum { @@ -352,7 +353,6 @@ public: { } - std::string unit_call(const std::string &unit_name, const std::string &unit_action, const std::string &data); std::string sys_sql_select(const std::string &key); void sys_sql_set(const std::string &key, const std::string &val); void sys_sql_unset(const std::string &key); diff --git a/ext_components/StackFlow/stackflow/StackFlowUtil.cpp b/ext_components/StackFlow/stackflow/StackFlowUtil.cpp index 956809c..f50b922 100644 --- a/ext_components/StackFlow/stackflow/StackFlowUtil.cpp +++ b/ext_components/StackFlow/stackflow/StackFlowUtil.cpp @@ -5,6 +5,7 @@ */ #include "StackFlowUtil.h" #include +#include "pzmq.hpp" std::string StackFlows::sample_json_str_get(const std::string &json_str, const std::string &json_key) { @@ -399,4 +400,12 @@ int StackFlows::encode_base64(const std::string &in, std::string &out) { out.resize(BASE64_ENCODE_OUT_SIZE(in.length())); return base64_encode((const unsigned char *)in.c_str(), in.length(), (char *)out.data()); +} + +std::string StackFlows::unit_call(const std::string &unit_name, const std::string &unit_action, const std::string &data) +{ + std::string value; + pzmq _call(unit_name); + _call.call_rpc_action(unit_action, data, [&value](const std::string &raw) { value = raw; }); + return value; } \ No newline at end of file diff --git a/ext_components/StackFlow/stackflow/StackFlowUtil.h b/ext_components/StackFlow/stackflow/StackFlowUtil.h index 1b5df5f..ffbe00f 100644 --- a/ext_components/StackFlow/stackflow/StackFlowUtil.h +++ b/ext_components/StackFlow/stackflow/StackFlowUtil.h @@ -5,8 +5,21 @@ */ #pragma once #include +#include #include #define WORK_ID_NONE -100 + +#define RPC_PUSH_PARAM(_obj, _data1, _data2) \ + do { \ + _obj.resize(1 + _data1.size() + _data2.size()); \ + _obj[0] = (unsigned char)_data1.size(); \ + memcpy((void *)(_obj.data() + 1), (void *)_data1.data(), _data1.size()); \ + memcpy((void *)(_obj.data() + _data1.size() + 1), (void *)_data2.data(), _data2.size()); \ + } while (0) + +#define RPC_PARSE_TO_FIRST(_obj) _obj.substr(1, static_cast(_obj[0])) +#define RPC_PARSE_TO_SECOND(_obj) _obj.substr(static_cast(_obj[0]) + 1) + namespace StackFlows { std::string sample_json_str_get(const std::string &json_str, const std::string &json_key); int sample_get_work_id_num(const std::string &work_id); @@ -17,4 +30,5 @@ std::string sample_unescapeString(const std::string &input); bool decode_stream(const std::string &in, std::string &out, std::unordered_map &stream_buff); int decode_base64(const std::string &in, std::string &out); int encode_base64(const std::string &in, std::string &out); +std::string unit_call(const std::string &unit_name, const std::string &unit_action, const std::string &data); }; // namespace StackFlows diff --git a/projects/llm_framework/main_audio/src/main.cpp b/projects/llm_framework/main_audio/src/main.cpp index cd8913d..141c531 100644 --- a/projects/llm_framework/main_audio/src/main.cpp +++ b/projects/llm_framework/main_audio/src/main.cpp @@ -116,6 +116,7 @@ public: self = this; cap_status_ = 0; rpc_ctx_->register_rpc_action("play", std::bind(&llm_audio::play, this, std::placeholders::_1)); + rpc_ctx_->register_rpc_action("play_raw", std::bind(&llm_audio::play_raw, this, std::placeholders::_1)); rpc_ctx_->register_rpc_action("queue_play", std::bind(&llm_audio::enqueue_play, this, std::placeholders::_1)); rpc_ctx_->register_rpc_action("play_stop", std::bind(&llm_audio::play_stop, this, std::placeholders::_1)); rpc_ctx_->register_rpc_action("queue_play_stop", @@ -369,13 +370,19 @@ public: std::string play(const std::string &rawdata) { - if (rawdata.size() < 3) return LLM_NONE; - if ((rawdata[0] == '{') && (rawdata[rawdata.size() - 1] == '}')) { - std::string src_data = sample_unescapeString(sample_json_str_get(rawdata, "raw_data")); - return parse_data(sample_json_str_get(src_data, "object"), sample_json_str_get(src_data, "data")); - } else { - _play(rawdata); - } + std::string zmq_url = RPC_PARSE_TO_FIRST(rawdata); + std::string audio_json = RPC_PARSE_TO_SECOND(rawdata); + std::string ret_val = parse_data(sample_json_str_get(audio_json, "object"), sample_json_str_get(audio_json, "data")); + request_id_ = sample_json_str_get(audio_json, "request_id"); + send(LLM_NONE, LLM_NONE, LLM_NO_ERROR, sample_json_str_get(audio_json, "work_id"), zmq_url); + return ret_val; + } + + std::string play_raw(const std::string &rawdata) + { + if(rawdata.empty()) + return std::string("rawdata empty"); + _play(rawdata); return LLM_NONE; } diff --git a/projects/llm_framework/main_kws/src/main.cpp b/projects/llm_framework/main_kws/src/main.cpp index 9eaedbf..e2ac716 100644 --- a/projects/llm_framework/main_kws/src/main.cpp +++ b/projects/llm_framework/main_kws/src/main.cpp @@ -310,7 +310,7 @@ public: } } if (post != 0) { - unit_call("audio", "play", std::string((char *)(wav_data.data() + post), size - post)); + unit_call("audio", "play_raw", std::string((char *)(wav_data.data() + post), size - post)); } } diff --git a/projects/llm_framework/main_sys/src/remote_action.cpp b/projects/llm_framework/main_sys/src/remote_action.cpp index a6e99c8..03ff87e 100644 --- a/projects/llm_framework/main_sys/src/remote_action.cpp +++ b/projects/llm_framework/main_sys/src/remote_action.cpp @@ -19,13 +19,11 @@ int remote_call(int com_id, const std::string &json_str) std::string work_id = sample_json_str_get(json_str, "work_id"); std::string work_unit = work_id.substr(0, work_id.find(".")); std::string action = sample_json_str_get(json_str, "action"); - char com_url[128]; - sprintf(com_url, zmq_c_format.c_str(), com_id); - std::string send_data = "{\"zmq_com\":\""; - send_data += std::string(com_url); - send_data += "\",\"raw_data\":\""; - send_data += sample_escapeString(json_str); - send_data += "\"}"; + char com_url[256]; + int length = snprintf(com_url, 255, zmq_c_format.c_str(), com_id); + std::string send_data; + std::string com_urls(com_url); + RPC_PUSH_PARAM(send_data, com_urls, json_str); pzmq clent(work_unit); return clent.call_rpc_action(action, send_data, [](const std::string &val) {}); } diff --git a/projects/llm_framework/main_sys/src/remote_server.cpp b/projects/llm_framework/main_sys/src/remote_server.cpp index 18ff306..83335df 100644 --- a/projects/llm_framework/main_sys/src/remote_server.cpp +++ b/projects/llm_framework/main_sys/src/remote_server.cpp @@ -171,22 +171,15 @@ int c_sys_release_unit(char const *unit) return sys_release_unit(unit); } -std::string rpc_sql_select(const std::string &raw) -{ - return sys_sql_select(raw); -} - std::string rpc_allocate_unit(const std::string &raw) { unit_data *unit_info = sys_allocate_unit(raw); - std::string retval = "{\"work_id_number\":"; - retval += std::to_string(unit_info->port_); - retval += ",\"out_port\":\""; - retval += unit_info->output_url; - retval += "\",\"inference_port\":\""; - retval += unit_info->inference_url; - retval += "\"}"; - return retval; + std::string send_data; + std::string send_data1; + std::string str_port = std::to_string(unit_info->port_); + RPC_PUSH_PARAM(send_data1, unit_info->output_url, unit_info->inference_url); + RPC_PUSH_PARAM(send_data, str_port, send_data1); + return send_data; } std::string rpc_release_unit(const std::string &raw) @@ -195,6 +188,11 @@ std::string rpc_release_unit(const std::string &raw) return "Success"; } +std::string rpc_sql_select(const std::string &raw) +{ + return sys_sql_select(raw); +} + std::string rpc_sql_set(const std::string &raw) { std::string key = sample_json_str_get(raw, "key"); diff --git a/projects/llm_framework/setup.ini b/projects/llm_framework/setup.ini index d01d0fb..e1363e0 100644 --- a/projects/llm_framework/setup.ini +++ b/projects/llm_framework/setup.ini @@ -1,7 +1,7 @@ [ssh] local_file_path = dist remote_file_path = /root/dist -remote_host = 192.168.28.62 +remote_host = m5stack-LLM.local remote_port = 22 username = root password = 123456 \ No newline at end of file