From d636d4d7b37a236034bd2fccd20df1a445c48ba2 Mon Sep 17 00:00:00 2001 From: dianjixz <18637716021@163.com> Date: Tue, 29 Apr 2025 11:21:35 +0800 Subject: [PATCH] [update] add pzmq pzmq_data class --- .../StackFlow/stackflow/StackFlow.cpp | 62 +++++++++------ .../StackFlow/stackflow/StackFlow.h | 26 ++++--- .../StackFlow/stackflow/StackFlowUtil.cpp | 2 +- ext_components/StackFlow/stackflow/pzmq.hpp | 76 ++++++++++++++----- projects/llm_framework/main_asr/src/main.cpp | 12 +-- .../llm_framework/main_audio/src/main.cpp | 35 +++++---- .../llm_framework/main_camera/src/main.cpp | 7 +- .../main_depth_anything/src/main.cpp | 8 +- projects/llm_framework/main_kws/src/main.cpp | 8 +- .../main_sys/src/remote_action.cpp | 2 +- .../main_sys/src/remote_server.cpp | 22 +++--- .../llm_framework/main_sys/src/zmq_bus.cpp | 2 +- projects/llm_framework/main_vad/src/main.cpp | 12 +-- .../llm_framework/main_whisper/src/main.cpp | 12 +-- projects/llm_framework/main_yolo/src/main.cpp | 8 +- 15 files changed, 178 insertions(+), 116 deletions(-) diff --git a/ext_components/StackFlow/stackflow/StackFlow.cpp b/ext_components/StackFlow/stackflow/StackFlow.cpp index 236cc7a..e06f478 100644 --- a/ext_components/StackFlow/stackflow/StackFlow.cpp +++ b/ext_components/StackFlow/stackflow/StackFlow.cpp @@ -14,7 +14,8 @@ std::string llm_channel_obj::uart_push_url; #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) +#define RPC_PARSE_TO_PARAM(obj) RPC_PARSE_TO_FIRST(obj), RPC_PARSE_TO_SECOND(obj) +#define PTR_RPC_PARSE_TO_PARAM(obj) PTR_RPC_PARSE_TO_FIRST(obj), PTR_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) @@ -30,23 +31,24 @@ llm_channel_obj::~llm_channel_obj() } void llm_channel_obj::subscriber_event_call(const std::function &call, - pzmq *_pzmq, const std::string &raw) + pzmq *_pzmq, const std::shared_ptr &raw) { + auto _raw = raw->string(); const char *user_inference_flage_str = "\"action\""; - std::size_t pos = raw.find(user_inference_flage_str); + std::size_t pos = _raw.find(user_inference_flage_str); while (true) { if (pos == std::string::npos) { break; - } else if ((pos > 0) && (raw[pos - 1] != '\\')) { - std::string zmq_com = sample_json_str_get(raw, "zmq_com"); + } else if ((pos > 0) && (_raw[pos - 1] != '\\')) { + std::string zmq_com = sample_json_str_get(_raw, "zmq_com"); if (!zmq_com.empty()) set_push_url(zmq_com); - request_id_ = sample_json_str_get(raw, "request_id"); - work_id_ = sample_json_str_get(raw, "work_id"); + request_id_ = sample_json_str_get(_raw, "request_id"); + work_id_ = sample_json_str_get(_raw, "work_id"); break; } - pos = raw.find(user_inference_flage_str, pos + sizeof(user_inference_flage_str)); + pos = _raw.find(user_inference_flage_str, pos + sizeof(user_inference_flage_str)); } - call(sample_json_str_get(raw, "object"), sample_json_str_get(raw, "data")); + call(sample_json_str_get(_raw, "object"), sample_json_str_get(_raw, "data")); } int llm_channel_obj::subscriber_work_id(const std::string &work_id, @@ -236,9 +238,10 @@ void StackFlow::_sys_init(const std::string &zmq_url, const std::string &data) // todo:... } -std::string StackFlow::_rpc_setup(pzmq *_pzmq, const std::string &data) +std::string StackFlow::_rpc_setup(pzmq *_pzmq, const std::shared_ptr &data) { - event_queue_.enqueue(EVENT_SETUP, RPC_PARSE_TO_PARAM(data)); + auto _data = data->string(); + event_queue_.enqueue(EVENT_SETUP, RPC_PARSE_TO_PARAM(_data)); return std::string("None"); } @@ -270,9 +273,10 @@ int StackFlow::setup(const std::string &work_id, const std::string &object, cons return -1; } -std::string StackFlow::_rpc_link(pzmq *_pzmq, const std::string &data) +std::string StackFlow::_rpc_link(pzmq *_pzmq, const std::shared_ptr &data) { - event_queue_.enqueue(EVENT_LINK, RPC_PARSE_TO_PARAM(data)); + auto _data = data->string(); + event_queue_.enqueue(EVENT_LINK, RPC_PARSE_TO_PARAM(_data)); return std::string("None"); } @@ -301,9 +305,10 @@ void StackFlow::link(const std::string &work_id, const std::string &object, cons send("None", "None", error_body, work_id); } -std::string StackFlow::_rpc_unlink(pzmq *_pzmq, const std::string &data) +std::string StackFlow::_rpc_unlink(pzmq *_pzmq, const std::shared_ptr &data) { - event_queue_.enqueue(EVENT_UNLINK, RPC_PARSE_TO_PARAM(data)); + auto _data = data->string(); + event_queue_.enqueue(EVENT_UNLINK, RPC_PARSE_TO_PARAM(_data)); return std::string("None"); } @@ -332,9 +337,11 @@ void StackFlow::unlink(const std::string &work_id, const std::string &object, co send("None", "None", error_body, work_id); } -std::string StackFlow::_rpc_work(pzmq *_pzmq, const std::string &data) +std::string StackFlow::_rpc_work(pzmq *_pzmq, const std::shared_ptr &data) { - event_queue_.enqueue(EVENT_WORK, RPC_PARSE_TO_PARAM(data)); + + auto _data = data->string(); + event_queue_.enqueue(EVENT_WORK, RPC_PARSE_TO_PARAM(_data)); return std::string("None"); } @@ -363,9 +370,12 @@ void StackFlow::work(const std::string &work_id, const std::string &object, cons send("None", "None", error_body, work_id); } -std::string StackFlow::_rpc_exit(pzmq *_pzmq, const std::string &data) +std::string StackFlow::_rpc_exit(pzmq *_pzmq, const std::shared_ptr &data) { - event_queue_.enqueue(EVENT_EXIT, RPC_PARSE_TO_PARAM(data)); + + + auto _data = data->string(); + event_queue_.enqueue(EVENT_EXIT, RPC_PARSE_TO_PARAM(_data)); return std::string("None"); } @@ -397,9 +407,12 @@ int StackFlow::exit(const std::string &work_id, const std::string &object, const return 0; } -std::string StackFlow::_rpc_pause(pzmq *_pzmq, const std::string &data) +std::string StackFlow::_rpc_pause(pzmq *_pzmq, const std::shared_ptr &data) { - event_queue_.enqueue(EVENT_PAUSE, RPC_PARSE_TO_PARAM(data)); + + + auto _data = data->string(); + event_queue_.enqueue(EVENT_PAUSE, RPC_PARSE_TO_PARAM(_data)); return std::string("None"); } @@ -428,9 +441,12 @@ void StackFlow::pause(const std::string &work_id, const std::string &object, con send("None", "None", error_body, work_id); } -std::string StackFlow::_rpc_taskinfo(pzmq *_pzmq, const std::string &data) +std::string StackFlow::_rpc_taskinfo(pzmq *_pzmq, const std::shared_ptr &data) { - event_queue_.enqueue(EVENT_TASKINFO, RPC_PARSE_TO_PARAM(data)); + + + auto _data = data->string(); + event_queue_.enqueue(EVENT_TASKINFO, RPC_PARSE_TO_PARAM(_data)); return std::string("None"); } diff --git a/ext_components/StackFlow/stackflow/StackFlow.h b/ext_components/StackFlow/stackflow/StackFlow.h index 2098458..6827753 100644 --- a/ext_components/StackFlow/stackflow/StackFlow.h +++ b/ext_components/StackFlow/stackflow/StackFlow.h @@ -98,7 +98,7 @@ public: return enstream_; } void subscriber_event_call(const std::function &call, pzmq *_pzmq, - const std::string &raw); + const std::shared_ptr &raw); int subscriber_work_id(const std::string &work_id, const std::function &call); void stop_subscriber_work_id(const std::string &work_id); @@ -185,6 +185,14 @@ public: } }; +class stackflow_data { + union { + std::string *rawobj; + std::string *object; + }; + std::string *data; +}; + class StackFlow { private: std::atomic_int work_id_num_cout_; @@ -252,7 +260,7 @@ public: return llm_task_channel_.at(_work_id_num); } - std::string _rpc_setup(pzmq *_pzmq, const std::string &data); + std::string _rpc_setup(pzmq *_pzmq, const std::shared_ptr &data); void _setup(const std::string &zmq_url, const std::string &data) { // printf("void _setup run \n"); @@ -263,7 +271,7 @@ public: virtual int setup(const std::string &zmq_url, const std::string &raw); virtual int setup(const std::string &work_id, const std::string &object, const std::string &data); - std::string _rpc_link(pzmq *_pzmq, const std::string &data); + std::string _rpc_link(pzmq *_pzmq, const std::shared_ptr &data); void _link(const std::string &zmq_url, const std::string &data) { // printf("void _link run \n"); @@ -274,7 +282,7 @@ public: virtual void link(const std::string &zmq_url, const std::string &raw); virtual void link(const std::string &work_id, const std::string &object, const std::string &data); - std::string _rpc_unlink(pzmq *_pzmq, const std::string &data); + std::string _rpc_unlink(pzmq *_pzmq, const std::shared_ptr &data); void _unlink(const std::string &zmq_url, const std::string &data) { // printf("void _unlink run \n"); @@ -285,7 +293,7 @@ public: virtual void unlink(const std::string &zmq_url, const std::string &raw); virtual void unlink(const std::string &work_id, const std::string &object, const std::string &data); - std::string _rpc_exit(pzmq *_pzmq, const std::string &data); + std::string _rpc_exit(pzmq *_pzmq, const std::shared_ptr &data); void _exit(const std::string &zmq_url, const std::string &data) { request_id_ = sample_json_str_get(data, "request_id"); @@ -295,7 +303,7 @@ public: virtual int exit(const std::string &zmq_url, const std::string &raw); virtual int exit(const std::string &work_id, const std::string &object, const std::string &data); - std::string _rpc_work(pzmq *_pzmq, const std::string &data); + std::string _rpc_work(pzmq *_pzmq, const std::shared_ptr &data); void _work(const std::string &zmq_url, const std::string &data) { request_id_ = sample_json_str_get(data, "request_id"); @@ -305,7 +313,7 @@ public: virtual void work(const std::string &zmq_url, const std::string &raw); virtual void work(const std::string &work_id, const std::string &object, const std::string &data); - std::string _rpc_pause(pzmq *_pzmq, const std::string &data); + std::string _rpc_pause(pzmq *_pzmq, const std::shared_ptr &data); void _pause(const std::string &zmq_url, const std::string &data) { request_id_ = sample_json_str_get(data, "request_id"); @@ -315,7 +323,7 @@ public: virtual void pause(const std::string &zmq_url, const std::string &raw); virtual void pause(const std::string &work_id, const std::string &object, const std::string &data); - std::string _rpc_taskinfo(pzmq *_pzmq, const std::string &data); + std::string _rpc_taskinfo(pzmq *_pzmq, const std::shared_ptr &data); void _taskinfo(const std::string &zmq_url, const std::string &data) { request_id_ = sample_json_str_get(data, "request_id"); @@ -379,7 +387,7 @@ public: return false; } pzmq _call("sys"); - _call.call_rpc_action("release_unit", _work_id, [](pzmq *_pzmq, const std::string &data) {}); + _call.call_rpc_action("release_unit", _work_id, [](pzmq *_pzmq, const std::shared_ptr &data) {}); llm_task_channel_[_work_id_num].reset(); llm_task_channel_.erase(_work_id_num); // SLOGI("release work_id %s success", _work_id.c_str()); diff --git a/ext_components/StackFlow/stackflow/StackFlowUtil.cpp b/ext_components/StackFlow/stackflow/StackFlowUtil.cpp index a5cebea..8225b8b 100644 --- a/ext_components/StackFlow/stackflow/StackFlowUtil.cpp +++ b/ext_components/StackFlow/stackflow/StackFlowUtil.cpp @@ -358,7 +358,7 @@ std::string StackFlows::unit_call(const std::string &unit_name, const std::strin { std::string value; pzmq _call(unit_name); - _call.call_rpc_action(unit_action, data, [&value](pzmq *_pzmq, const std::string &raw) { value = raw; }); + _call.call_rpc_action(unit_action, data, [&value](pzmq *_pzmq, const std::shared_ptr &raw) { value = raw->string(); }); return value; } diff --git a/ext_components/StackFlow/stackflow/pzmq.hpp b/ext_components/StackFlow/stackflow/pzmq.hpp index 1a7baf9..c70a692 100644 --- a/ext_components/StackFlow/stackflow/pzmq.hpp +++ b/ext_components/StackFlow/stackflow/pzmq.hpp @@ -18,10 +18,48 @@ #define ZMQ_RPC_CALL (ZMQ_REQ | 0x80) namespace StackFlows { + +class pzmq_data { +private: + zmq_msg_t msg; + +public: + pzmq_data(/* args */) + { + zmq_msg_init(&msg); + } + std::shared_ptr get_string() + { + auto len = zmq_msg_size(&msg); + return std::make_shared((const char *)zmq_msg_data(&msg), zmq_msg_size(&msg)); + } + std::string string() + { + auto len = zmq_msg_size(&msg); + return std::string((const char *)zmq_msg_data(&msg), zmq_msg_size(&msg)); + } + void *data() + { + return zmq_msg_data(&msg); + } + size_t size() + { + return zmq_msg_size(&msg); + } + zmq_msg_t *get() + { + return &msg; + } + ~pzmq_data() + { + zmq_msg_close(&msg); + } +}; + class pzmq { public: - typedef std::function rpc_callback_fun; - typedef std::function msg_callback_fun; + typedef std::function &)> rpc_callback_fun; + typedef std::function &)> msg_callback_fun; private: const int rpc_url_head_length = 6; @@ -85,7 +123,7 @@ public: } return zmq_url_; } - std::string _rpc_list_action(pzmq *self, const std::string &_None) + std::string _rpc_list_action(pzmq *self, const std::shared_ptr &_None) { std::string action_list; action_list.reserve(128); @@ -131,8 +169,7 @@ public: int call_rpc_action(const std::string &action, const std::string &data, const msg_callback_fun &raw_call) { int ret; - zmq_msg_t msg; - zmq_msg_init(&msg); + std::shared_ptr msg_ptr = std::make_shared(); try { if (NULL == zmq_socket_) { if (rpc_server_.empty()) return -1; @@ -150,13 +187,13 @@ public: } // action { - zmq_msg_recv(&msg, zmq_socket_, 0); + zmq_msg_recv(msg_ptr->get(), zmq_socket_, 0); } - raw_call(this, std::string((const char *)zmq_msg_data(&msg), zmq_msg_size(&msg))); + raw_call(this, msg_ptr); } catch (int e) { ret = e; } - zmq_msg_close(&msg); + msg_ptr.reset(); close_zmq(); return ret; } @@ -293,8 +330,7 @@ public: items[0].revents = 0; }; while (!flage_.load()) { - zmq_msg_t msg; - zmq_msg_init(&msg); + std::shared_ptr msg_ptr = std::make_shared(); if (mode_ == ZMQ_PULL) { ret = zmq_poll(items, 1, -1); if (ret == -1) { @@ -305,30 +341,28 @@ public: continue; } } - ret = zmq_msg_recv(&msg, zmq_socket_, 0); + ret = zmq_msg_recv(msg_ptr->get(), zmq_socket_, 0); if (ret <= 0) { - zmq_msg_close(&msg); + msg_ptr.reset(); continue; } - std::string raw_data((const char *)zmq_msg_data(&msg), zmq_msg_size(&msg)); + if (mode_ == ZMQ_RPC_FUN) { - zmq_msg_t msg1; - zmq_msg_init(&msg1); - zmq_msg_recv(&msg1, zmq_socket_, 0); - std::string _raw_data((const char *)zmq_msg_data(&msg1), zmq_msg_size(&msg1)); + std::shared_ptr msg1_ptr = std::make_shared(); + zmq_msg_recv(msg1_ptr->get(), zmq_socket_, 0); std::string retval; try { std::unique_lock lock(zmq_fun_mtx_); - retval = zmq_fun_.at(raw_data)(this, _raw_data); + retval = zmq_fun_.at(msg_ptr->string())(this, msg1_ptr); } catch (...) { retval = "NotAction"; } zmq_send(zmq_socket_, retval.c_str(), retval.length(), 0); - zmq_msg_close(&msg1); + msg1_ptr.reset(); } else { - raw_call(this, raw_data); + raw_call(this, msg_ptr); } - zmq_msg_close(&msg); + msg_ptr.reset(); } } void close_zmq() diff --git a/projects/llm_framework/main_asr/src/main.cpp b/projects/llm_framework/main_asr/src/main.cpp index f09ade0..ebe503b 100644 --- a/projects/llm_framework/main_asr/src/main.cpp +++ b/projects/llm_framework/main_asr/src/main.cpp @@ -426,8 +426,8 @@ public: llm_task_obj->kws_awake(); if ((!audio_url_.empty()) && (llm_task_obj->audio_flage_ == false)) { std::weak_ptr _llm_task_obj = llm_task_obj; - llm_channel->subscriber(audio_url_, [_llm_task_obj](pzmq *_pzmq, const std::string &raw) { - _llm_task_obj.lock()->sys_pcm_on_data(raw); + llm_channel->subscriber(audio_url_, [_llm_task_obj](pzmq *_pzmq, const std::shared_ptr &raw) { + _llm_task_obj.lock()->sys_pcm_on_data(raw->string()); }); llm_task_obj->audio_flage_ = true; } @@ -515,8 +515,8 @@ public: if (input.find("sys") != std::string::npos) { audio_url_ = unit_call("audio", "cap", input); std::weak_ptr _llm_task_obj = llm_task_obj; - llm_channel->subscriber(audio_url_, [_llm_task_obj](pzmq *_pzmq, const std::string &raw) { - _llm_task_obj.lock()->sys_pcm_on_data(raw); + llm_channel->subscriber(audio_url_, [_llm_task_obj](pzmq *_pzmq, const std::shared_ptr &raw) { + _llm_task_obj.lock()->sys_pcm_on_data(raw->string()); }); llm_task_obj->audio_flage_ = true; } else if (input.find("asr") != std::string::npos) { @@ -563,8 +563,8 @@ public: if (data.find("sys") != std::string::npos) { if (audio_url_.empty()) audio_url_ = unit_call("audio", "cap", data); std::weak_ptr _llm_task_obj = llm_task_obj; - llm_channel->subscriber(audio_url_, [_llm_task_obj](pzmq *_pzmq, const std::string &raw) { - _llm_task_obj.lock()->sys_pcm_on_data(raw); + llm_channel->subscriber(audio_url_, [_llm_task_obj](pzmq *_pzmq, const std::shared_ptr &raw) { + _llm_task_obj.lock()->sys_pcm_on_data(raw->string()); }); llm_task_obj->audio_flage_ = true; llm_task_obj->inputs_.push_back(data); diff --git a/projects/llm_framework/main_audio/src/main.cpp b/projects/llm_framework/main_audio/src/main.cpp index 32407d3..f12d3af 100644 --- a/projects/llm_framework/main_audio/src/main.cpp +++ b/projects/llm_framework/main_audio/src/main.cpp @@ -389,10 +389,11 @@ public: return LLM_NONE; } - std::string play(pzmq *_pzmq, const std::string &rawdata) + std::string play(pzmq *_pzmq, const std::shared_ptr &rawdata) { - std::string zmq_url = RPC_PARSE_TO_FIRST(rawdata); - std::string audio_json = RPC_PARSE_TO_SECOND(rawdata); + auto _rawdata = rawdata->string(); + 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"); @@ -400,29 +401,31 @@ public: return ret_val; } - std::string play_raw(pzmq *_pzmq, const std::string &rawdata) + std::string play_raw(pzmq *_pzmq, const std::shared_ptr &rawdata) { - if (rawdata.empty()) return std::string("rawdata empty"); - _play(rawdata); + auto _rawdata = rawdata->string(); + if (_rawdata.empty()) return std::string("rawdata empty"); + _play(_rawdata); return LLM_NONE; } - std::string enqueue_play(pzmq *_pzmq, const std::string &rawdata) + std::string enqueue_play(pzmq *_pzmq, const std::shared_ptr &rawdata) { audio_clear_flage_ = false; - event_queue_.enqueue(EVENT_QUEUE_PLAY, rawdata, ""); + event_queue_.enqueue(EVENT_QUEUE_PLAY, rawdata->string(), ""); return LLM_NONE; } - std::string audio_status(pzmq *_pzmq, const std::string &rawdata) + std::string audio_status(pzmq *_pzmq, const std::shared_ptr &rawdata) { - if (rawdata == "play") { + std::string _rawdata = rawdata->string(); + if (_rawdata == "play") { if (ax_play_status()) { return std::string("None"); } else { return std::string("Runing"); } - } else if (rawdata == "cap") { + } else if (_rawdata == "cap") { if (ax_cap_status()) { return std::string("None"); } else { @@ -447,19 +450,19 @@ public: } } - std::string play_stop(pzmq *_pzmq, const std::string &rawdata) + std::string play_stop(pzmq *_pzmq, const std::shared_ptr &rawdata) { _play_stop(); return LLM_NONE; } - std::string queue_play_stop(pzmq *_pzmq, const std::string &rawdata) + std::string queue_play_stop(pzmq *_pzmq, const std::shared_ptr &rawdata) { audio_clear_flage_ = true; return LLM_NONE; } - std::string cap(pzmq *_pzmq, const std::string &rawdata) + std::string cap(pzmq *_pzmq, const std::shared_ptr &rawdata) { if (cap_status_ == 0) { _cap(); @@ -468,7 +471,7 @@ public: return sys_pcm_cap_channel; } - std::string cap_stop(pzmq *_pzmq, const std::string &rawdata) + std::string cap_stop(pzmq *_pzmq, const std::shared_ptr &rawdata) { if (cap_status_ > 0) { cap_status_--; @@ -479,7 +482,7 @@ public: return LLM_NONE; } - std::string cap_stop_all(pzmq *_pzmq, const std::string &rawdata) + std::string cap_stop_all(pzmq *_pzmq, const std::shared_ptr &rawdata) { cap_status_ = 0; _cap_stop(); diff --git a/projects/llm_framework/main_camera/src/main.cpp b/projects/llm_framework/main_camera/src/main.cpp index 26bbd77..c2c97f2 100644 --- a/projects/llm_framework/main_camera/src/main.cpp +++ b/projects/llm_framework/main_camera/src/main.cpp @@ -590,11 +590,12 @@ public: "list_camera", std::bind(&llm_camera::list_camera, this, std::placeholders::_1, std::placeholders::_2)); } - std::string list_camera(pzmq *_pzmq, const std::string &rawdata) + std::string list_camera(pzmq *_pzmq, const std::shared_ptr &rawdata) { + auto _rawdata = rawdata->string(); nlohmann::json req_body; - std::string zmq_url = RPC_PARSE_TO_FIRST(rawdata); - std::string param_json = RPC_PARSE_TO_SECOND(rawdata); + std::string zmq_url = RPC_PARSE_TO_FIRST(_rawdata); + std::string param_json = RPC_PARSE_TO_SECOND(_rawdata); std::vector devices; glob_t glob_result; glob("/dev/video*", GLOB_TILDE, NULL, &glob_result); diff --git a/projects/llm_framework/main_depth_anything/src/main.cpp b/projects/llm_framework/main_depth_anything/src/main.cpp index b7bb591..7685bc8 100644 --- a/projects/llm_framework/main_depth_anything/src/main.cpp +++ b/projects/llm_framework/main_depth_anything/src/main.cpp @@ -419,8 +419,8 @@ public: std::weak_ptr _llm_task_obj = llm_task_obj; std::weak_ptr _llm_channel = llm_channel; llm_channel->subscriber( - input_url, [this, _llm_task_obj, _llm_channel](pzmq *_pzmq, const std::string &raw) { - this->task_camera_data(_llm_task_obj, _llm_channel, raw); + input_url, [this, _llm_task_obj, _llm_channel](pzmq *_pzmq, const std::shared_ptr &raw) { + this->task_camera_data(_llm_task_obj, _llm_channel, raw->string()); }); } } @@ -466,8 +466,8 @@ public: std::weak_ptr _llm_task_obj = llm_task_obj; std::weak_ptr _llm_channel = llm_channel; llm_channel->subscriber(input_url, - [this, _llm_task_obj, _llm_channel](pzmq *_pzmq, const std::string &raw) { - this->task_camera_data(_llm_task_obj, _llm_channel, raw); + [this, _llm_task_obj, _llm_channel](pzmq *_pzmq, const std::shared_ptr &raw) { + this->task_camera_data(_llm_task_obj, _llm_channel, raw->string()); }); } llm_task_obj->inputs_.push_back(data); diff --git a/projects/llm_framework/main_kws/src/main.cpp b/projects/llm_framework/main_kws/src/main.cpp index fc4f31c..9c05ce8 100644 --- a/projects/llm_framework/main_kws/src/main.cpp +++ b/projects/llm_framework/main_kws/src/main.cpp @@ -336,8 +336,8 @@ public: } if ((!audio_url_.empty()) && (llm_task_obj->audio_flage_ == false)) { std::weak_ptr _llm_task_obj = llm_task_obj; - llm_channel->subscriber(audio_url_, [_llm_task_obj](pzmq *_pzmq, const std::string &raw) { - _llm_task_obj.lock()->sys_pcm_on_data(raw); + llm_channel->subscriber(audio_url_, [_llm_task_obj](pzmq *_pzmq, const std::shared_ptr &raw) { + _llm_task_obj.lock()->sys_pcm_on_data(raw->string()); }); llm_task_obj->audio_flage_ = true; } @@ -456,8 +456,8 @@ public: if (input.find("sys") != std::string::npos) { audio_url_ = unit_call("audio", "cap", "None"); std::weak_ptr _llm_task_obj = llm_task_obj; - llm_channel->subscriber(audio_url_, [_llm_task_obj](pzmq *_pzmq, const std::string &raw) { - _llm_task_obj.lock()->sys_pcm_on_data(raw); + llm_channel->subscriber(audio_url_, [_llm_task_obj](pzmq *_pzmq, const std::shared_ptr &raw) { + _llm_task_obj.lock()->sys_pcm_on_data(raw->string()); }); llm_task_obj->audio_flage_ = true; } else if (input.find("kws") != std::string::npos) { diff --git a/projects/llm_framework/main_sys/src/remote_action.cpp b/projects/llm_framework/main_sys/src/remote_action.cpp index 2593369..6121f69 100644 --- a/projects/llm_framework/main_sys/src/remote_action.cpp +++ b/projects/llm_framework/main_sys/src/remote_action.cpp @@ -25,7 +25,7 @@ int remote_call(int com_id, const std::string &json_str) 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, [](pzmq *_pzmq, const std::string &val) {}); + return clent.call_rpc_action(action, send_data, [](pzmq *_pzmq, const std::shared_ptr &val) {}); } void remote_action_work() diff --git a/projects/llm_framework/main_sys/src/remote_server.cpp b/projects/llm_framework/main_sys/src/remote_server.cpp index 5a4567f..48ea242 100644 --- a/projects/llm_framework/main_sys/src/remote_server.cpp +++ b/projects/llm_framework/main_sys/src/remote_server.cpp @@ -171,9 +171,9 @@ int c_sys_release_unit(char const *unit) return sys_release_unit(unit); } -std::string rpc_allocate_unit(pzmq *_pzmq, const std::string &raw) +std::string rpc_allocate_unit(pzmq *_pzmq, const std::shared_ptr &raw) { - unit_data *unit_info = sys_allocate_unit(raw); + unit_data *unit_info = sys_allocate_unit(raw->string()); std::string send_data; std::string send_data1; std::string str_port = std::to_string(unit_info->port_); @@ -182,29 +182,29 @@ std::string rpc_allocate_unit(pzmq *_pzmq, const std::string &raw) return send_data; } -std::string rpc_release_unit(pzmq *_pzmq, const std::string &raw) +std::string rpc_release_unit(pzmq *_pzmq, const std::shared_ptr &raw) { - sys_release_unit(raw); + sys_release_unit(raw->string()); return "Success"; } -std::string rpc_sql_select(pzmq *_pzmq, const std::string &raw) +std::string rpc_sql_select(pzmq *_pzmq, const std::shared_ptr &raw) { - return sys_sql_select(raw); + return sys_sql_select(raw->string()); } -std::string rpc_sql_set(pzmq *_pzmq, const std::string &raw) +std::string rpc_sql_set(pzmq *_pzmq, const std::shared_ptr &raw) { - std::string key = sample_json_str_get(raw, "key"); - std::string val = sample_json_str_get(raw, "val"); + std::string key = sample_json_str_get(raw->string(), "key"); + std::string val = sample_json_str_get(raw->string(), "val"); if (key.empty()) return "False"; sys_sql_set(key, val); return "Success"; } -std::string rpc_sql_unset(pzmq *_pzmq, const std::string &raw) +std::string rpc_sql_unset(pzmq *_pzmq, const std::shared_ptr &raw) { - sys_sql_unset(raw); + sys_sql_unset(raw->string()); return "Success"; } diff --git a/projects/llm_framework/main_sys/src/zmq_bus.cpp b/projects/llm_framework/main_sys/src/zmq_bus.cpp index 60693c9..b62196e 100644 --- a/projects/llm_framework/main_sys/src/zmq_bus.cpp +++ b/projects/llm_framework/main_sys/src/zmq_bus.cpp @@ -42,7 +42,7 @@ void zmq_bus_com::work(const std::string &zmq_url_format, int port) _zmq_url = std::string((char *)buff.data()); SAFE_SETTING("serial_zmq_url", _zmq_url); user_chennal_ = std::make_unique(_zmq_url, ZMQ_PULL, - [this](pzmq *_pzmq, const std::string &data) { this->send_data(data); }); + [this](pzmq *_pzmq, const std::shared_ptr &data) { this->send_data(data->string()); }); reace_data_event_thread = std::make_unique(std::bind(&zmq_bus_com::reace_data_event, this)); } diff --git a/projects/llm_framework/main_vad/src/main.cpp b/projects/llm_framework/main_vad/src/main.cpp index f0ab519..3c737bc 100644 --- a/projects/llm_framework/main_vad/src/main.cpp +++ b/projects/llm_framework/main_vad/src/main.cpp @@ -335,8 +335,8 @@ public: } if ((!audio_url_.empty()) && (llm_task_obj->audio_flage_ == false)) { std::weak_ptr _llm_task_obj = llm_task_obj; - llm_channel->subscriber(audio_url_, [_llm_task_obj](pzmq *_pzmq, const std::string &raw) { - _llm_task_obj.lock()->sys_pcm_on_data(raw); + llm_channel->subscriber(audio_url_, [_llm_task_obj](pzmq *_pzmq, const std::shared_ptr &raw) { + _llm_task_obj.lock()->sys_pcm_on_data(raw->string()); }); llm_task_obj->audio_flage_ = true; } @@ -423,8 +423,8 @@ public: if (input.find("sys") != std::string::npos) { audio_url_ = unit_call("audio", "cap", "None"); std::weak_ptr _llm_task_obj = llm_task_obj; - llm_channel->subscriber(audio_url_, [_llm_task_obj](pzmq *_pzmq, const std::string &raw) { - _llm_task_obj.lock()->sys_pcm_on_data(raw); + llm_channel->subscriber(audio_url_, [_llm_task_obj](pzmq *_pzmq, const std::shared_ptr &raw) { + _llm_task_obj.lock()->sys_pcm_on_data(raw->string()); }); llm_task_obj->audio_flage_ = true; } else if (input.find("vad") != std::string::npos) { @@ -471,8 +471,8 @@ public: if (data.find("sys") != std::string::npos) { if (audio_url_.empty()) audio_url_ = unit_call("audio", "cap", data); std::weak_ptr _llm_task_obj = llm_task_obj; - llm_channel->subscriber(audio_url_, [_llm_task_obj](pzmq *_pzmq, const std::string &raw) { - _llm_task_obj.lock()->sys_pcm_on_data(raw); + llm_channel->subscriber(audio_url_, [_llm_task_obj](pzmq *_pzmq, const std::shared_ptr &raw) { + _llm_task_obj.lock()->sys_pcm_on_data(raw->string()); }); llm_task_obj->audio_flage_ = true; llm_task_obj->inputs_.push_back(data); diff --git a/projects/llm_framework/main_whisper/src/main.cpp b/projects/llm_framework/main_whisper/src/main.cpp index 776408b..367ee87 100644 --- a/projects/llm_framework/main_whisper/src/main.cpp +++ b/projects/llm_framework/main_whisper/src/main.cpp @@ -715,8 +715,8 @@ public: llm_task_obj->kws_awake(); if ((!audio_url_.empty()) && (llm_task_obj->audio_flage_ == false)) { std::weak_ptr _llm_task_obj = llm_task_obj; - llm_channel->subscriber(audio_url_, [_llm_task_obj](pzmq *_pzmq, const std::string &raw) { - _llm_task_obj.lock()->sys_pcm_on_data(raw); + llm_channel->subscriber(audio_url_, [_llm_task_obj](pzmq *_pzmq, const std::shared_ptr &raw) { + _llm_task_obj.lock()->sys_pcm_on_data(raw->string()); }); llm_task_obj->audio_flage_ = true; } @@ -818,8 +818,8 @@ public: if (input.find("sys") != std::string::npos) { audio_url_ = unit_call("audio", "cap", input); std::weak_ptr _llm_task_obj = llm_task_obj; - llm_channel->subscriber(audio_url_, [_llm_task_obj](pzmq *_pzmq, const std::string &raw) { - _llm_task_obj.lock()->sys_pcm_on_data(raw); + llm_channel->subscriber(audio_url_, [_llm_task_obj](pzmq *_pzmq, const std::shared_ptr &raw) { + _llm_task_obj.lock()->sys_pcm_on_data(raw->string()); }); llm_task_obj->audio_flage_ = true; } else if (input.find("whisper") != std::string::npos) { @@ -874,8 +874,8 @@ public: if (data.find("sys") != std::string::npos) { if (audio_url_.empty()) audio_url_ = unit_call("audio", "cap", data); std::weak_ptr _llm_task_obj = llm_task_obj; - llm_channel->subscriber(audio_url_, [_llm_task_obj](pzmq *_pzmq, const std::string &raw) { - _llm_task_obj.lock()->sys_pcm_on_data(raw); + llm_channel->subscriber(audio_url_, [_llm_task_obj](pzmq *_pzmq, const std::shared_ptr &raw) { + _llm_task_obj.lock()->sys_pcm_on_data(raw->string()); }); llm_task_obj->audio_flage_ = true; llm_task_obj->inputs_.push_back(data); diff --git a/projects/llm_framework/main_yolo/src/main.cpp b/projects/llm_framework/main_yolo/src/main.cpp index 453ad34..208a375 100644 --- a/projects/llm_framework/main_yolo/src/main.cpp +++ b/projects/llm_framework/main_yolo/src/main.cpp @@ -504,8 +504,8 @@ public: std::weak_ptr _llm_task_obj = llm_task_obj; std::weak_ptr _llm_channel = llm_channel; llm_channel->subscriber( - input_url, [this, _llm_task_obj, _llm_channel](pzmq *_pzmq, const std::string &raw) { - this->task_camera_data(_llm_task_obj, _llm_channel, raw); + input_url, [this, _llm_task_obj, _llm_channel](pzmq *_pzmq, const std::shared_ptr &raw) { + this->task_camera_data(_llm_task_obj, _llm_channel, raw->string()); }); } } @@ -551,8 +551,8 @@ public: std::weak_ptr _llm_task_obj = llm_task_obj; std::weak_ptr _llm_channel = llm_channel; llm_channel->subscriber(input_url, - [this, _llm_task_obj, _llm_channel](pzmq *_pzmq, const std::string &raw) { - this->task_camera_data(_llm_task_obj, _llm_channel, raw); + [this, _llm_task_obj, _llm_channel](pzmq *_pzmq, const std::shared_ptr &raw) { + this->task_camera_data(_llm_task_obj, _llm_channel, raw->string()); }); } llm_task_obj->inputs_.push_back(data);