m5/StackFlow
0
0
Fork 0
mirror of https://github.com/m5stack/StackFlow.git synced 2026-05-20 11:32:11 -07:00
Code Issues Packages Projects Releases Wiki Activity

[update] StackFlow add stackflow_data && pzmq add get_param set_param

Browse Source
This commit is contained in:
dianjixz
2025-05-06 15:25:54 +08:00
parent 4e301f1cbf
commit 2a8a744777
13 changed files with 220 additions and 128 deletions
Download Patch File Download Diff File Expand all files Collapse all files
doc/projects_llm_framework_doc/llm_camera_zh.md
+28
View File
@@ -155,4 +155,32 @@ error::code 为 0 表示执行成功。
}
```
获取本机的摄像头列表。
发送 json：
```json
{
"request_id": "2",
"work_id": "camera",
"action": "list_camera"
}
```
响应 json：
```json
{
"created":1746515639,
"data":[],
"error":{
"code":0,
"message":""
},
"object":"camera.devices",
"request_id":"2",
"work_id":"camera"
}
```
> **注意：work_id 是按照单元的初始化注册顺序增加的，并不是固定的索引值。**
ext_components/StackFlow/stackflow/StackFlow.cpp
+43 -59
View File
@@ -33,7 +33,7 @@ llm_channel_obj::~llm_channel_obj()
void llm_channel_obj::subscriber_event_call(const std::function<void(const std::string &, const std::string &)> &call,
pzmq *_pzmq, const std::shared_ptr<pzmq_data> &raw)
{
auto _raw = raw->string();
auto _raw = raw->string();
const char *user_inference_flage_str = "\"action\"";
std::size_t pos = _raw.find(user_inference_flage_str);
while (true) {
@@ -153,26 +153,16 @@ int llm_channel_obj::output_to_uart(const std::string &data)
StackFlow::StackFlow::StackFlow(const std::string &unit_name)
: work_id_num_cout_(1000), unit_name_(unit_name), rpc_ctx_(std::make_unique<pzmq>(unit_name))
{
event_queue_.appendListener(EVENT_NONE,
std::bind(&StackFlow::_none_event, this, std::placeholders::_1, std::placeholders::_2));
event_queue_.appendListener(EVENT_PAUSE,
std::bind(&StackFlow::_pause, this, std::placeholders::_1, std::placeholders::_2));
event_queue_.appendListener(EVENT_WORK,
std::bind(&StackFlow::_work, this, std::placeholders::_1, std::placeholders::_2));
event_queue_.appendListener(EVENT_EXIT,
std::bind(&StackFlow::_exit, this, std::placeholders::_1, std::placeholders::_2));
event_queue_.appendListener(EVENT_SETUP,
std::bind(&StackFlow::_setup, this, std::placeholders::_1, std::placeholders::_2));
event_queue_.appendListener(EVENT_LINK,
std::bind(&StackFlow::_link, this, std::placeholders::_1, std::placeholders::_2));
event_queue_.appendListener(EVENT_UNLINK,
std::bind(&StackFlow::_unlink, this, std::placeholders::_1, std::placeholders::_2));
event_queue_.appendListener(EVENT_TASKINFO,
std::bind(&StackFlow::_taskinfo, this, std::placeholders::_1, std::placeholders::_2));
event_queue_.appendListener(EVENT_SYS_INIT,
std::bind(&StackFlow::_sys_init, this, std::placeholders::_1, std::placeholders::_2));
event_queue_.appendListener(
EVENT_REPEAT_EVENT, std::bind(&StackFlow::_repeat_loop, this, std::placeholders::_1, std::placeholders::_2));
event_queue_.appendListener(EVENT_NONE, std::bind(&StackFlow::_none_event, this, std::placeholders::_1));
event_queue_.appendListener(EVENT_PAUSE, std::bind(&StackFlow::_pause, this, std::placeholders::_1));
event_queue_.appendListener(EVENT_WORK, std::bind(&StackFlow::_work, this, std::placeholders::_1));
event_queue_.appendListener(EVENT_EXIT, std::bind(&StackFlow::_exit, this, std::placeholders::_1));
event_queue_.appendListener(EVENT_SETUP, std::bind(&StackFlow::_setup, this, std::placeholders::_1));
event_queue_.appendListener(EVENT_LINK, std::bind(&StackFlow::_link, this, std::placeholders::_1));
event_queue_.appendListener(EVENT_UNLINK, std::bind(&StackFlow::_unlink, this, std::placeholders::_1));
event_queue_.appendListener(EVENT_TASKINFO, std::bind(&StackFlow::_taskinfo, this, std::placeholders::_1));
event_queue_.appendListener(EVENT_SYS_INIT, std::bind(&StackFlow::_sys_init, this, std::placeholders::_1));
event_queue_.appendListener(EVENT_REPEAT_EVENT, std::bind(&StackFlow::_repeat_loop, this, std::placeholders::_1));
rpc_ctx_->register_rpc_action(
"setup", std::bind(&StackFlow::_rpc_setup, this, std::placeholders::_1, std::placeholders::_2));
rpc_ctx_->register_rpc_action(
@@ -217,7 +207,7 @@ StackFlow::~StackFlow()
llm_task_channel_.erase(iteam->first);
}
exit_flage_.store(true);
event_queue_.enqueue(EVENT_NONE, "", "");
event_queue_.enqueue(EVENT_NONE, nullptr);
even_loop_thread_->join();
}
@@ -229,19 +219,19 @@ void StackFlow::even_loop()
}
}
void StackFlow::_none_event(const std::string &data1, const std::string &data2)
void StackFlow::_none_event(const std::shared_ptr<void> &arg)
{
// std::shared_ptr<stackflow_data> originalPtr = std::static_pointer_cast<stackflow_data>(arg);
}
void StackFlow::_sys_init(const std::string &zmq_url, const std::string &data)
void StackFlow::_sys_init(const std::shared_ptr<void> &arg)
{
// todo:...
}
std::string StackFlow::_rpc_setup(pzmq *_pzmq, const std::shared_ptr<pzmq_data> &data)
{
auto _data = data->string();
event_queue_.enqueue(EVENT_SETUP, RPC_PARSE_TO_PARAM(_data));
event_queue_.enqueue(EVENT_SETUP, std::make_shared<stackflow_data>(data->get_param(0), data->get_param(1)));
return std::string("None");
}
@@ -275,8 +265,7 @@ int StackFlow::setup(const std::string &work_id, const std::string &object, cons
std::string StackFlow::_rpc_link(pzmq *_pzmq, const std::shared_ptr<pzmq_data> &data)
{
auto _data = data->string();
event_queue_.enqueue(EVENT_LINK, RPC_PARSE_TO_PARAM(_data));
event_queue_.enqueue(EVENT_LINK, std::make_shared<stackflow_data>(data->get_param(0), data->get_param(1)));
return std::string("None");
}
@@ -307,8 +296,7 @@ void StackFlow::link(const std::string &work_id, const std::string &object, cons
std::string StackFlow::_rpc_unlink(pzmq *_pzmq, const std::shared_ptr<pzmq_data> &data)
{
auto _data = data->string();
event_queue_.enqueue(EVENT_UNLINK, RPC_PARSE_TO_PARAM(_data));
event_queue_.enqueue(EVENT_UNLINK, std::make_shared<stackflow_data>(data->get_param(0), data->get_param(1)));
return std::string("None");
}
@@ -339,9 +327,7 @@ void StackFlow::unlink(const std::string &work_id, const std::string &object, co
std::string StackFlow::_rpc_work(pzmq *_pzmq, const std::shared_ptr<pzmq_data> &data)
{
auto _data = data->string();
event_queue_.enqueue(EVENT_WORK, RPC_PARSE_TO_PARAM(_data));
event_queue_.enqueue(EVENT_WORK, std::make_shared<stackflow_data>(data->get_param(0), data->get_param(1)));
return std::string("None");
}
@@ -372,10 +358,7 @@ void StackFlow::work(const std::string &work_id, const std::string &object, cons
std::string StackFlow::_rpc_exit(pzmq *_pzmq, const std::shared_ptr<pzmq_data> &data)
{
auto _data = data->string();
event_queue_.enqueue(EVENT_EXIT, RPC_PARSE_TO_PARAM(_data));
event_queue_.enqueue(EVENT_EXIT, std::make_shared<stackflow_data>(data->get_param(0), data->get_param(1)));
return std::string("None");
}
@@ -409,10 +392,7 @@ int StackFlow::exit(const std::string &work_id, const std::string &object, const
std::string StackFlow::_rpc_pause(pzmq *_pzmq, const std::shared_ptr<pzmq_data> &data)
{
auto _data = data->string();
event_queue_.enqueue(EVENT_PAUSE, RPC_PARSE_TO_PARAM(_data));
event_queue_.enqueue(EVENT_PAUSE, std::make_shared<stackflow_data>(data->get_param(0), data->get_param(1)));
return std::string("None");
}
@@ -443,16 +423,13 @@ void StackFlow::pause(const std::string &work_id, const std::string &object, con
std::string StackFlow::_rpc_taskinfo(pzmq *_pzmq, const std::shared_ptr<pzmq_data> &data)
{
auto _data = data->string();
event_queue_.enqueue(EVENT_TASKINFO, RPC_PARSE_TO_PARAM(_data));
event_queue_.enqueue(EVENT_TASKINFO, std::make_shared<stackflow_data>(data->get_param(0), data->get_param(1)));
return std::string("None");
}
void StackFlow::taskinfo(const std::string &zmq_url, const std::string &raw)
{
SLOGI("StackFlow::taskinfo raw");
// SLOGI("StackFlow::taskinfo raw");
std::string work_id = sample_json_str_get(raw, "work_id");
try {
auto task_channel = get_channel(sample_get_work_id_num(work_id));
@@ -464,7 +441,7 @@ void StackFlow::taskinfo(const std::string &zmq_url, const std::string &raw)
void StackFlow::taskinfo(const std::string &work_id, const std::string &object, const std::string &data)
{
SLOGI("StackFlow::taskinfo");
// SLOGI("StackFlow::taskinfo");
if (_taskinfo_) {
_taskinfo_(work_id, object, data);
return;
@@ -478,13 +455,17 @@ 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 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);
std::string str_port;
std::string out_port;
std::string inference_port;
unit_call("sys", "register_unit", unit_name, [&](const std::shared_ptr<StackFlows::pzmq_data> &pzmg_msg) {
str_port = pzmg_msg->get_param(1);
out_port = pzmg_msg->get_param(0, str_port);
inference_port = pzmg_msg->get_param(1, str_port);
str_port = pzmg_msg->get_param(0);
});
work_id_number = std::stoi(str_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<llm_channel_obj>(out_port, inference_port, unit_name_);
@@ -527,20 +508,23 @@ void StackFlow::sys_sql_unset(const std::string &key)
unit_call("sys", "sql_unset", key);
}
void StackFlow::_repeat_loop(const std::string &action, const std::string &ms)
void StackFlow::_repeat_loop(const std::shared_ptr<void> &arg)
{
std::shared_ptr<stackflow_data> originalPtr = std::static_pointer_cast<stackflow_data>(arg);
std::string action = originalPtr->string(0);
int ms = originalPtr->integer(0);
repeat_callback_fun_mutex_.lock();
const auto call_fun = repeat_callback_fun_[action];
repeat_callback_fun_mutex_.unlock();
if (call_fun()) {
int delayms = std::stoi(ms);
int delayms = ms;
if (delayms)
std::thread([this, action, delayms, ms]() {
std::this_thread::sleep_for(std::chrono::milliseconds(delayms));
this->event_queue_.enqueue(EVENT_REPEAT_EVENT, action, ms);
this->event_queue_.enqueue(EVENT_REPEAT_EVENT, std::make_shared<stackflow_data>(action, ms));
}).detach();
else {
event_queue_.enqueue(EVENT_REPEAT_EVENT, action, ms);
event_queue_.enqueue(EVENT_REPEAT_EVENT, std::make_shared<stackflow_data>(action, ms));
}
} else {
repeat_callback_fun_mutex_.lock();
@@ -558,9 +542,9 @@ void StackFlow::repeat_event(int ms, std::function<int(void)> repeat_fun, bool n
if (!now)
std::thread([this, action, ms]() {
std::this_thread::sleep_for(std::chrono::milliseconds(ms));
this->event_queue_.enqueue(EVENT_REPEAT_EVENT, action, std::to_string(ms));
this->event_queue_.enqueue(EVENT_REPEAT_EVENT, std::make_shared<stackflow_data>(action, ms));
}).detach();
else {
event_queue_.enqueue(EVENT_REPEAT_EVENT, action, std::to_string(ms));
event_queue_.enqueue(EVENT_REPEAT_EVENT, std::make_shared<stackflow_data>(action, ms));
}
}
ext_components/StackFlow/stackflow/StackFlow.h
+68 -24
View File
@@ -186,11 +186,34 @@ public:
};
class stackflow_data {
union {
std::string *rawobj;
std::string *object;
};
std::string *data;
public:
stackflow_data()
{
}
stackflow_data(const std::string &_data1)
{
str_data[0] = _data1;
}
stackflow_data(const std::string &_data1, const std::string &_data2)
{
str_data[0] = _data1;
str_data[1] = _data2;
}
stackflow_data(const std::string &_data1, int _data2)
{
str_data[0] = _data1;
int_data[0] = _data2;
}
std::string string(int index = 0)
{
return str_data[index];
}
int integer(int index = 0)
{
return int_data[index];
}
std::string str_data[2];
int int_data[2];
};
class StackFlow {
@@ -217,7 +240,7 @@ protected:
EVENT_EXPORT,
} local_event_t;
eventpp::EventQueue<int, void(const std::string &, const std::string &)> event_queue_;
eventpp::EventQueue<int, void(const std::shared_ptr<void> &)> event_queue_;
std::unique_ptr<std::thread> even_loop_thread_;
std::unique_ptr<pzmq> rpc_ctx_;
std::atomic<int> status_;
@@ -225,7 +248,7 @@ protected:
std::unordered_map<std::string, std::function<int(void)>> repeat_callback_fun_;
std::mutex repeat_callback_fun_mutex_;
void _repeat_loop(const std::string &zmq_url, const std::string &raw);
void _repeat_loop(const std::shared_ptr<void> &arg);
public:
std::string request_id_;
@@ -244,7 +267,7 @@ public:
StackFlow(const std::string &unit_name);
void even_loop();
void _none_event(const std::string &data1, const std::string &data2);
void _none_event(const std::shared_ptr<void> &arg);
template <typename T>
std::shared_ptr<llm_channel_obj> get_channel(T workid)
@@ -261,8 +284,11 @@ public:
}
std::string _rpc_setup(pzmq *_pzmq, const std::shared_ptr<pzmq_data> &data);
void _setup(const std::string &zmq_url, const std::string &data)
void _setup(const std::shared_ptr<void> &arg)
{
std::shared_ptr<stackflow_data> originalPtr = std::static_pointer_cast<stackflow_data>(arg);
std::string zmq_url = originalPtr->string(0);
std::string data = originalPtr->string(1);
// printf("void _setup run \n");
request_id_ = sample_json_str_get(data, "request_id");
out_zmq_url_ = zmq_url;
@@ -272,8 +298,11 @@ public:
virtual int setup(const std::string &work_id, const std::string &object, const std::string &data);
std::string _rpc_link(pzmq *_pzmq, const std::shared_ptr<pzmq_data> &data);
void _link(const std::string &zmq_url, const std::string &data)
void _link(const std::shared_ptr<void> &arg)
{
std::shared_ptr<stackflow_data> originalPtr = std::static_pointer_cast<stackflow_data>(arg);
std::string zmq_url = originalPtr->string(0);
std::string data = originalPtr->string(1);
// printf("void _link run \n");
request_id_ = sample_json_str_get(data, "request_id");
out_zmq_url_ = zmq_url;
@@ -283,8 +312,11 @@ public:
virtual void link(const std::string &work_id, const std::string &object, const std::string &data);
std::string _rpc_unlink(pzmq *_pzmq, const std::shared_ptr<pzmq_data> &data);
void _unlink(const std::string &zmq_url, const std::string &data)
void _unlink(const std::shared_ptr<void> &arg)
{
std::shared_ptr<stackflow_data> originalPtr = std::static_pointer_cast<stackflow_data>(arg);
std::string zmq_url = originalPtr->string(0);
std::string data = originalPtr->string(1);
// printf("void _unlink run \n");
request_id_ = sample_json_str_get(data, "request_id");
out_zmq_url_ = zmq_url;
@@ -294,46 +326,58 @@ public:
virtual void unlink(const std::string &work_id, const std::string &object, const std::string &data);
std::string _rpc_exit(pzmq *_pzmq, const std::shared_ptr<pzmq_data> &data);
void _exit(const std::string &zmq_url, const std::string &data)
void _exit(const std::shared_ptr<void> &arg)
{
request_id_ = sample_json_str_get(data, "request_id");
out_zmq_url_ = zmq_url;
std::shared_ptr<stackflow_data> originalPtr = std::static_pointer_cast<stackflow_data>(arg);
std::string zmq_url = originalPtr->string(0);
std::string data = originalPtr->string(1);
request_id_ = sample_json_str_get(data, "request_id");
out_zmq_url_ = zmq_url;
if (status_.load()) exit(zmq_url, data);
}
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::shared_ptr<pzmq_data> &data);
void _work(const std::string &zmq_url, const std::string &data)
void _work(const std::shared_ptr<void> &arg)
{
request_id_ = sample_json_str_get(data, "request_id");
out_zmq_url_ = zmq_url;
std::shared_ptr<stackflow_data> originalPtr = std::static_pointer_cast<stackflow_data>(arg);
std::string zmq_url = originalPtr->string(0);
std::string data = originalPtr->string(1);
request_id_ = sample_json_str_get(data, "request_id");
out_zmq_url_ = zmq_url;
if (status_.load()) work(zmq_url, data);
}
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::shared_ptr<pzmq_data> &data);
void _pause(const std::string &zmq_url, const std::string &data)
void _pause(const std::shared_ptr<void> &arg)
{
request_id_ = sample_json_str_get(data, "request_id");
out_zmq_url_ = zmq_url;
std::shared_ptr<stackflow_data> originalPtr = std::static_pointer_cast<stackflow_data>(arg);
std::string zmq_url = originalPtr->string(0);
std::string data = originalPtr->string(1);
request_id_ = sample_json_str_get(data, "request_id");
out_zmq_url_ = zmq_url;
if (status_.load()) pause(zmq_url, data);
}
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::shared_ptr<pzmq_data> &data);
void _taskinfo(const std::string &zmq_url, const std::string &data)
void _taskinfo(const std::shared_ptr<void> &arg)
{
request_id_ = sample_json_str_get(data, "request_id");
out_zmq_url_ = zmq_url;
std::shared_ptr<stackflow_data> originalPtr = std::static_pointer_cast<stackflow_data>(arg);
std::string zmq_url = originalPtr->string(0);
std::string data = originalPtr->string(1);
request_id_ = sample_json_str_get(data, "request_id");
out_zmq_url_ = zmq_url;
if (status_.load()) taskinfo(zmq_url, data);
}
virtual void taskinfo(const std::string &zmq_url, const std::string &raw);
virtual void taskinfo(const std::string &work_id, const std::string &object, const std::string &data);
void _sys_init(const std::string &zmq_url, const std::string &data);
void _sys_init(const std::shared_ptr<void> &arg);
void user_output(const std::string &zmq_url, const std::string &request_id, const std::string &data);
template <typename T, typename U>
ext_components/StackFlow/stackflow/StackFlowUtil.cpp
+7
View File
@@ -362,6 +362,13 @@ std::string StackFlows::unit_call(const std::string &unit_name, const std::strin
return value;
}
void StackFlows::unit_call(const std::string &unit_name, const std::string &unit_action, const std::string &data, std::function<void(const std::shared_ptr<StackFlows::pzmq_data> &)> callback)
{
std::string value;
StackFlows::pzmq _call(unit_name);
_call.call_rpc_action(unit_action, data, [callback](StackFlows::pzmq *_pzmq, const std::shared_ptr<StackFlows::pzmq_data> &raw) { callback(raw); });
}
std::list<std::string> StackFlows::get_config_file_paths(std::string &base_model_path,
std::string &base_model_config_path,
const std::string &mode_name)
ext_components/StackFlow/stackflow/StackFlowUtil.h
+4
View File
@@ -9,6 +9,9 @@
#include <unordered_map>
#include <list>
#include <vector>
#include <functional>
#include "pzmq.hpp"
#include <memory>
#define WORK_ID_NONE -100
#define RPC_PUSH_PARAM(_obj, _data1, _data2) \
@@ -33,6 +36,7 @@ bool decode_stream(const std::string &in, std::string &out, std::unordered_map<i
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);
void unit_call(const std::string &unit_name, const std::string &unit_action, const std::string &data, std::function<void(const std::shared_ptr<StackFlows::pzmq_data> &)> callback);
std::list<std::string> get_config_file_paths(std::string &base_model_path, std::string &base_model_config_path, const std::string &mode_name);
std::vector<std::string> glob_files(const std::vector<std::string> &patterns);
bool file_exists(const std::string& filePath);
ext_components/StackFlow/stackflow/pzmq.hpp
+27
View File
@@ -50,6 +50,33 @@ public:
{
return &msg;
}
std::string get_param(int index, const std::string &idata = "")
{
const char *data = NULL;
int size = 0;
if (idata.length() > 0) {
data = idata.c_str();
size = idata.length();
} else {
data = (const char *)zmq_msg_data(&msg);
size = zmq_msg_size(&msg);
}
if ((index % 2) == 0) {
return std::string((const char *)(data + 1), data[0]);
} else {
return std::string((const char *)(data + data[0] + 1), zmq_msg_size(&msg) - data[0] - 1);
}
}
static std::string set_param(std::string param0, std::string param1)
{
std::string data = " " + param0 + param1;
data[0] = param0.length();
return data;
}
~pzmq_data()
{
zmq_msg_close(&msg);
projects/llm_framework/main_asr/src/main.cpp
+9 -8
View File
@@ -278,8 +278,7 @@ public:
llm_asr() : StackFlow("asr")
{
task_count_ = 1;
event_queue_.appendListener(
EVENT_TASK_PAUSE, std::bind(&llm_asr::_task_pause, this, std::placeholders::_1, std::placeholders::_2));
event_queue_.appendListener(EVENT_TASK_PAUSE, std::bind(&llm_asr::_task_pause, this, std::placeholders::_1));
}
void task_output(const std::weak_ptr<llm_task> llm_task_obj_weak,
@@ -396,9 +395,10 @@ public:
llm_task_obj->sys_pcm_on_data((*next_data));
}
void _task_pause(const std::string &work_id, const std::string &data)
void _task_pause(const std::shared_ptr<void> &arg)
{
int work_id_num = sample_get_work_id_num(work_id);
std::shared_ptr<std::string> work_id = std::static_pointer_cast<std::string>(arg);
int work_id_num = sample_get_work_id_num(*work_id);
if (llm_task_.find(work_id_num) == llm_task_.end()) {
return;
}
@@ -412,7 +412,7 @@ public:
void task_pause(const std::string &work_id, const std::string &data)
{
event_queue_.enqueue(EVENT_TASK_PAUSE, work_id, "");
event_queue_.enqueue(EVENT_TASK_PAUSE, std::make_shared<std::string>(work_id));
}
void task_work(const std::weak_ptr<llm_task> llm_task_obj_weak,
@@ -515,9 +515,10 @@ public:
if (input.find("sys") != std::string::npos) {
audio_url_ = unit_call("audio", "cap", input);
std::weak_ptr<llm_task> _llm_task_obj = llm_task_obj;
llm_channel->subscriber(audio_url_, [_llm_task_obj](pzmq *_pzmq, const std::shared_ptr<pzmq_data> &raw) {
_llm_task_obj.lock()->sys_pcm_on_data(raw->string());
});
llm_channel->subscriber(audio_url_,
[_llm_task_obj](pzmq *_pzmq, const std::shared_ptr<pzmq_data> &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) {
llm_channel->subscriber_work_id(
projects/llm_framework/main_audio/src/main.cpp
+8 -8
View File
@@ -48,14 +48,15 @@ private:
self->pub_ctx_->send_data((const char *)data, size);
}
void hw_queue_play(const std::string &audio_data, const std::string &None)
void hw_queue_play(const std::shared_ptr<void> &arg)
{
if (audio_clear_flage_) {
return;
}
std::shared_ptr<pzmq_data> originalPtr = std::static_pointer_cast<pzmq_data>(arg);
std::lock_guard<std::mutex> guard(ax_play_mtx);
ax_play(play_config.card, play_config.device, play_config.volume, play_config.channel, play_config.rate,
play_config.bit, audio_data.c_str(), audio_data.length());
play_config.bit, originalPtr->data(), originalPtr->size());
}
void hw_play(const std::string &audio_data)
@@ -109,8 +110,8 @@ private:
public:
llm_audio() : StackFlow("audio")
{
event_queue_.appendListener(
EVENT_QUEUE_PLAY, std::bind(&llm_audio::hw_queue_play, this, std::placeholders::_1, std::placeholders::_2));
event_queue_.appendListener(EVENT_QUEUE_PLAY,
std::bind(&llm_audio::hw_queue_play, this, std::placeholders::_1));
setup("", "audio.play", "{\"None\":\"None\"}");
setup("", "audio.cap", "{\"None\":\"None\"}");
self = this;
@@ -391,9 +392,8 @@ public:
std::string play(pzmq *_pzmq, const std::shared_ptr<pzmq_data> &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 zmq_url = rawdata->get_param(0);
std::string audio_json = rawdata->get_param(1);
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");
@@ -412,7 +412,7 @@ public:
std::string enqueue_play(pzmq *_pzmq, const std::shared_ptr<pzmq_data> &rawdata)
{
audio_clear_flage_ = false;
event_queue_.enqueue(EVENT_QUEUE_PLAY, rawdata->string(), "");
event_queue_.enqueue(EVENT_QUEUE_PLAY, rawdata);
return LLM_NONE;
}
projects/llm_framework/main_camera/src/main.cpp
+2 -2
View File
@@ -594,8 +594,8 @@ public:
{
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 = rawdata->get_param(0);
std::string param_json = rawdata->get_param(1);
std::vector<std::string> devices;
glob_t glob_result;
glob("/dev/video*", GLOB_TILDE, NULL, &glob_result);
projects/llm_framework/main_sys/src/remote_action.cpp
+3 -5
View File
@@ -20,12 +20,10 @@ int remote_call(int com_id, const std::string &json_str)
std::string work_unit = work_id.substr(0, work_id.find("."));
std::string action = sample_json_str_get(json_str, "action");
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);
snprintf(com_url, 255, zmq_c_format.c_str(), com_id);
pzmq clent(work_unit);
return clent.call_rpc_action(action, send_data, [](pzmq *_pzmq, const std::shared_ptr<pzmq_data> &val) {});
return clent.call_rpc_action(action, pzmq_data::set_param(com_url, json_str),
[](pzmq *_pzmq, const std::shared_ptr<pzmq_data> &val) {});
}
void remote_action_work()
projects/llm_framework/main_sys/src/remote_server.cpp
+2 -6
View File
@@ -174,12 +174,8 @@ int c_sys_release_unit(char const *unit)
std::string rpc_allocate_unit(pzmq *_pzmq, const std::shared_ptr<pzmq_data> &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_);
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;
return pzmq_data::set_param(std::to_string(unit_info->port_),
pzmq_data::set_param(unit_info->output_url, unit_info->inference_url));
}
std::string rpc_release_unit(pzmq *_pzmq, const std::shared_ptr<pzmq_data> &raw)
projects/llm_framework/main_vad/src/main.cpp
+9 -8
View File
@@ -244,8 +244,7 @@ public:
llm_vad() : StackFlow("vad")
{
task_count_ = 1;
event_queue_.appendListener(
EVENT_TASK_PAUSE, std::bind(&llm_vad::_task_pause, this, std::placeholders::_1, std::placeholders::_2));
event_queue_.appendListener(EVENT_TASK_PAUSE, std::bind(&llm_vad::_task_pause, this, std::placeholders::_1));
}
void task_output(const std::weak_ptr<llm_task> llm_task_obj_weak,
@@ -306,9 +305,10 @@ public:
llm_task_obj->sys_pcm_on_data((*next_data));
}
void _task_pause(const std::string &work_id, const std::string &data)
void _task_pause(const std::shared_ptr<void> &arg)
{
int work_id_num = sample_get_work_id_num(work_id);
std::shared_ptr<std::string> work_id = std::static_pointer_cast<std::string>(arg);
int work_id_num = sample_get_work_id_num(*work_id);
if (llm_task_.find(work_id_num) == llm_task_.end()) {
return;
}
@@ -322,7 +322,7 @@ public:
void task_pause(const std::string &work_id, const std::string &data)
{
event_queue_.enqueue(EVENT_TASK_PAUSE, work_id, "");
event_queue_.enqueue(EVENT_TASK_PAUSE, std::make_shared<std::string>(work_id));
}
void task_work(const std::weak_ptr<llm_task> llm_task_obj_weak,
@@ -423,9 +423,10 @@ public:
if (input.find("sys") != std::string::npos) {
audio_url_ = unit_call("audio", "cap", "None");
std::weak_ptr<llm_task> _llm_task_obj = llm_task_obj;
llm_channel->subscriber(audio_url_, [_llm_task_obj](pzmq *_pzmq, const std::shared_ptr<pzmq_data> &raw) {
_llm_task_obj.lock()->sys_pcm_on_data(raw->string());
});
llm_channel->subscriber(audio_url_,
[_llm_task_obj](pzmq *_pzmq, const std::shared_ptr<pzmq_data> &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) {
llm_channel->subscriber_work_id(
projects/llm_framework/main_whisper/src/main.cpp
+10 -8
View File
@@ -595,8 +595,8 @@ public:
llm_whisper() : StackFlow("whisper")
{
task_count_ = 1;
event_queue_.appendListener(
EVENT_TASK_PAUSE, std::bind(&llm_whisper::_task_pause, this, std::placeholders::_1, std::placeholders::_2));
event_queue_.appendListener(EVENT_TASK_PAUSE,
std::bind(&llm_whisper::_task_pause, this, std::placeholders::_1));
}
void task_output(const std::weak_ptr<llm_task> llm_task_obj_weak,
@@ -713,9 +713,10 @@ public:
llm_task_obj->sys_pcm_on_data((*next_data));
}
void _task_pause(const std::string &work_id, const std::string &data)
void _task_pause(const std::shared_ptr<void> &arg)
{
int work_id_num = sample_get_work_id_num(work_id);
std::shared_ptr<std::string> work_id = std::static_pointer_cast<std::string>(arg);
int work_id_num = sample_get_work_id_num(*work_id);
if (llm_task_.find(work_id_num) == llm_task_.end()) {
return;
}
@@ -729,7 +730,7 @@ public:
void task_pause(const std::string &work_id, const std::string &data)
{
event_queue_.enqueue(EVENT_TASK_PAUSE, work_id, "");
event_queue_.enqueue(EVENT_TASK_PAUSE, std::make_shared<std::string>(work_id));
}
void task_work(const std::weak_ptr<llm_task> llm_task_obj_weak,
@@ -846,9 +847,10 @@ public:
if (input.find("sys") != std::string::npos) {
audio_url_ = unit_call("audio", "cap", input);
std::weak_ptr<llm_task> _llm_task_obj = llm_task_obj;
llm_channel->subscriber(audio_url_, [_llm_task_obj](pzmq *_pzmq, const std::shared_ptr<pzmq_data> &raw) {
_llm_task_obj.lock()->sys_pcm_on_data(raw->string());
});
llm_channel->subscriber(audio_url_,
[_llm_task_obj](pzmq *_pzmq, const std::shared_ptr<pzmq_data> &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) {
if (input.find("stream.base64") != std::string::npos) llm_task_obj->delay_audio_frame_ = 0;