实现信号槽机制自动化脚本

This commit is contained in:
2025-06-24 16:32:32 +08:00
parent 1d747d96fd
commit d9baa7f7a3
6 changed files with 407 additions and 80 deletions

277
create_signal_fun.py Normal file
View File

@@ -0,0 +1,277 @@
import os
import sys
import dataclasses
SIGNAL_INDEX=0
TMP_DIR="build"
@dataclasses.dataclass
class signal_fun_t:
name:str
text:str
struct:str
@dataclasses.dataclass
class slot_fun_t:
name:str
text:str
struct:str
# 找到文件夹中指定后缀的文件
def find_file(path:str,fix:str):
file_list=[]
if(os.path.exists(path)):
for file_name in os.listdir(path):
if file_name.endswith(fix):
file_list.append(os.path.join(path, file_name))
else:
print(f"{path} 不存在")
return file_list
# 找到文件夹列表中指定后缀的文件
def find_file_list(path_list:list,fix:str):
file_list=[]
for path in path_list:
file_list+=find_file(path,fix)
return file_list
# 找到 signal 关键字 定义的行
def find_signal_slot_def(file):
list_signal=[]
list_slot=[]
with open(file,encoding="utf-8") as f:
list_str=f.readlines()
for i in list_str:
if(i[0:6]=="signal"):
list_signal.append(i)
if(i[0:4]=="slot"):
list_slot.append(i)
return list_signal,list_slot
# 截取参数列表中的变量名
def split_par_name(par_str:str):
ret_str=""
if(par_str.count('*')>0):
ret_str=par_str.split("*")[1].strip()
else:
ret_str=par_str.split(" ")[1]
return ret_str
# 生成一个结构体描述
def def_struct_str(signal_fun:str,pars:list):
s_pars=""
for index,item in enumerate(pars):
if(index==0):
t=item.replace("*"," ").split()[-1]
s_pars+=f"void* {t};\n"
else:
s_pars+=f" {item};\n"
struct_str=f"""
typedef struct {'{'}
slot_list_with_pars slot_;
{s_pars}{'}'}{signal_fun}_args;
"""
return struct_str
# 获取参数列表中的变量名
def get_pars_names(pars:list):
names=[]
for item in pars:
t=item.replace("*"," ")
names.append(t.split(" ")[-1])
return names
# 生成一个信号函数描述
def def_signal_fun_str(signal_fun:str,pars:list):
pars_str=""
for index,item in enumerate(get_pars_names(pars)):
if(index>0):
pars_str+=f" pars->{item} = {item};\n"
else:
first_par=item
pars_str+=f"pars->{item} = slot_p->slot_obj;\n"
signal_fun_str=f"""
void {signal_fun}({','.join(pars)}){'{'}
{signal_fun}_args* pars = NULL;
slot_list* slot_p = {first_par}->__sig_obj.slot_head;
while (slot_p) {'{'}
if (slot_p->signal_func == {signal_fun}) {'{'}
pars = calloc(1, sizeof({signal_fun}_args));
pars->slot_.func = slot_p->func;
{pars_str} if (slot_p->thread) {'{'}
// 异步调用
send_slot_fun(slot_p->thread, (slot_list_with_pars*)pars);
{'}'} else {'{'}
// 同步调用
slot_p->func(pars);
free(pars);
{'}'}
{'}'}
slot_p = slot_p->next;
{'}'}
{'}'}
"""
return signal_fun_str
# 生成一个槽函数描述
def def_slot_fun_str(slot_fun:str,pars:list):
pars_str=""
for index,item in enumerate(get_pars_names(pars)):
if(index<len(pars)-1):
pars_str+=f"a->{item},"
else:
pars_str+=f"a->{item}"
slot_fun_str=f"""
static void {slot_fun}_caller(void *args){'{'}
{slot_fun}_args *a = args;
{slot_fun}({pars_str});
{'}'}
"""
return slot_fun_str
# 生成一个信号函数的实现
def def_signal_fun(line:str):
# 删除多余空格
line=' '.join(line.split())
# print(line)
list_str=line.split('(')
fun_name=list_str[0].split(' ')[1]
param_str=list_str[1].split(')')[0]
params=[]
# 有","则至少有两个参数否则可能有一个参数,可能没有
if(param_str.count(',')>0):
params=param_str.split(',')
for i in range(len(params)):
params[i]=params[i].strip()
else:
t_str=param_str.strip()
if(len(t_str)>0)and(t_str!="void"):
params.append(t_str)
# print(fun_name,params)
struct_str=def_struct_str(fun_name,params)
fun_str=def_signal_fun_str(fun_name,params)
# print(fun_str)
return signal_fun_t(fun_name,fun_str,struct_str)
# 生成一个槽函数的实现
def def_slot_fun(line:str):
# 删除多余空格
line=' '.join(line.split())
# print(line)
list_str=line.split('(')
fun_name=list_str[0].split(' ')[1]
param_str=list_str[1].split(')')[0]
params=[]
# 有","则至少有两个参数否则可能有一个参数,可能没有
if(param_str.count(',')>0):
params=param_str.split(',')
for i in range(len(params)):
params[i]=params[i].strip()
else:
t_str=param_str.strip()
if(len(t_str)>0)and(t_str!="void"):
params.append(t_str)
# print(fun_name,params)
struct_str=def_struct_str(fun_name,params)
fun_str=def_slot_fun_str(fun_name,params)
# print(fun_str)
return slot_fun_t(fun_name,fun_str,struct_str)
# 生成槽函数指针数组
def gen_slot_fun_array(slot_fun_list:list) -> str:
item_list=[]
for item in slot_fun_list:
item_list.append(f" .func = {item.name}_caller,\n .name=\"{item.name}\"\n")
table_str="""
// 定义一个数据结构来保存槽封装函数与槽函数的关系
typedef struct {
void (*func)(void*);
char *name;
} func_name;
static const func_name g_func_table[] = {
"""
for item in item_list:
table_str+=f" {'{'}\n{item} {'}'},\n"
table_str+="};\n"
table_str+="""
void* signal_find_slot_func(const char* name) {
for (int i = 0; i < sizeof(g_func_table) / sizeof(func_name); i++) {
if (strcmp(name, g_func_table[i].name) == 0) {
return g_func_table[i].func;
}
}
return 0;
}
"""
return table_str
# 遍历路径中的 .h 文件 返回找到的文件和信号声明
def ergodic_signal_fun(path_list:list):
signal_def_list=[]
slot_def_list=[]
list_file=find_file_list(path_list,".h")
signal_header=[]
for i in list_file:
list_signal,list_slot=find_signal_slot_def(i)
if(len(list_signal)>0 or len(list_slot)>0):
signal_header.append(i)
for j in list_signal:
signal_def_list.append(def_signal_fun(j))
for k in list_slot:
slot_def_list.append(def_slot_fun(k))
return signal_header,signal_def_list,slot_def_list
# 判断是否需要重新生成
def check_rebuild(dst:str,src:list):
if(not os.path.exists(dst)):
return True
dst_time=os.path.getmtime(dst)
src_time=[]
if(len(src)==0):
return True
for i in src:
src_time.append(os.path.getmtime(i))
src_time.sort()
if(src_time[-1]>dst_time):
return True
return False
# 创建 moc 文件
def moc_file_create(out_file_path,scan_path_list):
list_file,signal_list,slot_list=ergodic_signal_fun(scan_path_list)
# 不需要重新生成
if(not check_rebuild(out_file_path,list_file)):
return
with open(out_file_path,"w+") as f:
f.write("#include \"stdlib.h\"\n")
f.write("#include \"string.h\"\n")
f.write("#include \"mysignal.h\"\n")
for i in list_file:
f.write("#include \""+i.split("/")[-1]+"\"\n")
f.write("\n\n\n\n\n\n")
for item in signal_list:
f.write(item.struct)
f.write(item.text)
f.write("\n\n\n\n\n\n")
for item in slot_list:
f.write(item.struct)
f.write(item.text)
f.write("\n\n\n\n\n\n")
f.write(gen_slot_fun_array(slot_list))
if __name__=="__main__":
moc_file_create(f"{TMP_DIR}/mod_test.c",["test"])

View File

@@ -7,6 +7,7 @@ import os
import sys
import time
from create_lambda_fun import search_lambda
from create_signal_fun import moc_file_create
'''
@@ -24,9 +25,9 @@ CC = 'gcc'
# HEX = 'C:\\ARM_GCC\\bin\\arm-none-eabi-objcopy' + ' -O ihex'
# BIN = 'C:\\ARM_GCC\\bin\\arm-none-eabi-objcopy' + ' -O binary -S'
CSRC = ["main.c","test/lambda_test.c"]
CSRC = ["main.c","test/signal_test.c"]
CINC = ['-Isoft',"-Icpu", "-I./"]
CINC = ['-Isoft',"-Icpu", "-I./","-Itest"]
CDEF = ["-DTEST","-DLINUX"]
@@ -195,6 +196,8 @@ def main():
CSRC+=find_type('soft',['c','C'])
CSRC+=find_type('cpu',['c','C'])
CSRC=search_lambda(CSRC)
moc_file_create(f"{BUILD_DIR}/moc_tmp.c",list(item[2:] for item in CINC))
CSRC.append(f"{BUILD_DIR}/moc_tmp.c")
# ASRC+=find_type('./',['s','S','asm','ASM'])

View File

@@ -61,6 +61,7 @@ void _connect(void* sig_obj, void* sig_fun, mythread_t* thread, void* slot_obj,
slot_list* slot_p = calloc(sizeof(slot_list), 1);
slot_p->slot_obj = slot_obj;
slot_p->func = signal_find_slot_func(slot_fun);
slot_p->signal_func = sig_fun;
if (!slot_p->func) {
// 找不到槽函数,无法连接
free(slot_p);

View File

@@ -27,6 +27,7 @@ typedef struct {
typedef struct {
void* next;
void (*func)(void* par);
void* signal_func;
void* slot_obj;
mythread_t* thread;
}slot_list;

View File

@@ -7,94 +7,84 @@
#include "stdio.h"
#include "string.h"
#include "signal_test.h"
typedef struct {
SIG_OBJ;
int test_var;
}test_sig_obj;
typedef struct {
SIG_OBJ;
int test_var;
}test_sig_obj2;
// 定义槽函数
slot test_slot(test_sig_obj2* self, int a, int b) {
printf("test_slot var=%d\n", self->test_var);
self->test_var = a + b;
printf("test_slot var=%d\n", self->test_var);
}
// 定义信号
signal test_signal(test_sig_obj* self, int a, int b);
// 以下是自动生成的代码示例
// 把槽函数的参数封装为结构体
typedef struct {
slot_list_with_pars slot_;
test_sig_obj2* self;
int a;
int b;
}test_signal_args;
// // 把槽函数的参数封装为结构体
// typedef struct {
// slot_list_with_pars slot_;
// void* self;
// int a;
// int b;
// }test_signal_args;
// 封装函数用来调用实际的槽函数
static void test_slot_func(void* args) {
test_signal_args* a = args;
printf("test_slot_func: %d %d\n", a->a, a->b);
printf("args_p=%p\n", a);
test_slot(a->self, a->a, a->b);
}
// 信号函数的实现
signal test_signal(test_sig_obj* self, int a, int b) {
test_signal_args* pars = NULL;
slot_list* slot_p = self->__sig_obj.slot_head;
while (slot_p) {
pars = calloc(1, sizeof(test_signal_args));
pars->slot_.func = slot_p->func;
// 这里self要换成槽的self
pars->self = slot_p->slot_obj;
pars->a = a;
pars->b = b;
if (slot_p->thread) {
// 异步调用
send_slot_fun(slot_p->thread, (slot_list_with_pars*)pars);
} else {
// 同步调用
slot_p->func(pars);
free(pars);
}
slot_p = slot_p->next;
}
}
// 定义一个数据结构来保存槽封装函数与槽函数的关系
typedef struct {
void (*func)(void*);
char *name;
} func_name;
static const func_name g_func_table[] = {
{
.func = test_slot_func,
.name = "test_slot"
}
};
// typedef struct {
// slot_list_with_pars slot_;
// void* self;
// int a;
// int b;
// }test_slot_args;
void* signal_find_slot_func(const char* name) {
for (int i = 0; i < sizeof(g_func_table) / sizeof(func_name); i++) {
if (strcmp(name, g_func_table[i].name) == 0) {
return g_func_table[i].func;
}
}
return 0;
}
// // 封装函数用来调用实际的槽函数
// static void test_slot_caller(void* args) {
// test_slot_args* a = args;
// printf("test_slot_caller: %d %d\n", a->a, a->b);
// printf("args_p=%p\n", a);
// test_slot(a->self, a->a, a->b);
// }
// // 信号函数的实现
// signal test_signal(test_sig_obj* self, int a, int b) {
// test_signal_args* pars = NULL;
// slot_list* slot_p = self->__sig_obj.slot_head;
// while (slot_p) {
// pars = calloc(1, sizeof(test_signal_args));
// pars->slot_.func = slot_p->func;
// // 这里self要换成槽的self
// pars->self = slot_p->slot_obj;
// pars->a = a;
// pars->b = b;
// if (slot_p->thread) {
// // 异步调用
// send_slot_fun(slot_p->thread, (slot_list_with_pars*)pars);
// } else {
// // 同步调用
// slot_p->func(pars);
// free(pars);
// }
// slot_p = slot_p->next;
// }
// }
// // 定义一个数据结构来保存槽封装函数与槽函数的关系
// typedef struct {
// void (*func)(void*);
// char *name;
// } func_name;
// static const func_name g_func_table[] = {
// {
// .func = test_slot_caller,
// .name = "test_slot"
// }
// };
// void* signal_find_slot_func(const char* name) {
// for (int i = 0; i < sizeof(g_func_table) / sizeof(func_name); i++) {
// if (strcmp(name, g_func_table[i].name) == 0) {
// return g_func_table[i].func;
// }
// }
// return 0;
// }
// 自动生成代码示例结束
@@ -109,17 +99,39 @@ static void mdelay(unsigned long mSec){
}while(err<0 && errno==EINTR);
}
// 定义槽函数
slot test_slot(test_sig_obj2* obj, int a, int b) {
printf("test_slot var=%d\n", obj->test_var);
obj->test_var = a + b;
printf("test_slot var=%d\n", obj->test_var);
}
slot test_slot3(test_sig_obj3* obj, int a, int b) {
printf("test_slot3 a=%d, b=%d\n", a, b);
printf("obj->v1=%d, obj->v2=%d\n", obj->test_var, obj->test_var2);
}
slot test_slot3_2(test_sig_obj3* obj, int a, int b, const char* c) {
printf("test_slot3_2 a=%d, b=%d, c=%s\n", a, b, c);
}
static test_sig_obj g_sig_obj = { .test_var = 1, };
static test_sig_obj2 g_sig_obj2 = { .test_var = 2, };
static test_sig_obj3 g_sig_obj3;
int thread_fun(void* t) {
mythread_t* th = sigthread_init();
printf("thread_fun start\n");
printf("test_slot_func=%p\n", test_slot_func);
connect(&g_sig_obj, test_signal, th, &g_sig_obj2, test_slot);
connect(&g_sig_obj, test_signal, th, &g_sig_obj3, test_slot3);
connect(&g_sig_obj, test_signal2, th, &g_sig_obj3, test_slot3_2);
mdelay(1000);
emit test_signal(&g_sig_obj, 2, 3);
mdelay(1000);
emit test_signal2(&g_sig_obj, 5, 6,"hello world");
mdelay(1000);
printf("test end\n");
return 0;
}

33
test/signal_test.h Normal file
View File

@@ -0,0 +1,33 @@
#include "mysignal.h"
typedef struct {
SIG_OBJ;
int test_var;
}test_sig_obj;
signal test_signal(test_sig_obj* t, int a, int b);
signal test_signal2(test_sig_obj* t, int a, int b,const char*c);
typedef struct {
SIG_OBJ;
int test_var;
}test_sig_obj2;
slot test_slot(test_sig_obj2* obj, int a, int b);
typedef struct {
SIG_OBJ;
int test_var;
int test_var2;
}test_sig_obj3;
slot test_slot3(test_sig_obj3* obj, int a, int b);
slot test_slot3_2(test_sig_obj3* obj, int a, int b,const char*c);