【大联大世平 NXP i.MX RT1064开发板试用体验】5-简介+coslink协议设计及python实现上位机软件终结篇

2020-11-23 20:55:58

1245 开发板试用

本帖最后由 jf_00240724 于 2020-11-23 21:24 编辑

简介
这里将介绍coslink具体协议设计及软件架构。以及python实现上位机软件。
RT1064开发板有足够的RAM和ROM，但仍然有必要将状态机的资源使用保持在最低限度。
Coslink 数据格式:
First Byte
Second Byte
Third Byte
Fourth Byte …
End – 1 Byte
End Byte
Start Bit
Length of data segment
CDM-ID
Data segment
Checksum
End Flag
0xAA
0xxx
0xxx
0xxx
0xxx
‘/n’
CDM-ID：
-心跳（0x55）：
每1秒发送一次，应在3秒内从主机接收到相同的消息，否则报告连接失败错误并向主机发送超时连接。使用stm_heartbeat_1second_delay延迟1秒发送，stm_heartbeat_3second_lost_delay来检测心跳的响应。
-请求重新发送（0x00）：
当检测到错误的校验和时，向主机发送重新发送请求命令。
-按键（0x01）：
当目标检测到密钥打开时，目标向主机发送此命令
-确认键（0x02）：
当主机收到按键时，主机向目标发送此命令
-主机的连接丢失（0x03）：
当超过3秒无法接收主机响应时，目标向主机发送此命令以通知连接丢失。
校验和：
-发送方加起始位+数据段长度+CMD id+所有数据段，将每一位的结果倒置，得到低位字节为校验和接收机加起始位+数据段长度+命令id+所有数据段+校验和+1，如果结果为0，则数据正确，否则错误。如果错误，发送重新发送请求并报告错误的校验和。
软件系统设计

状态迁移表：

Current state	Condition	Next state	Action	Transition
S1	C1	S2	action_1	t1
S1	C2	S1	action_4	t4
S1	C3	S1	action_1	t1
S1	C4	S1	action_4	t4
S2	C1	S2	action_1	t1
S2	C2	S3	action_2	t2
S2	C3	S3	action_3	t33
S2	C4	S1	action_4	t4
S3	C3	S2	action_3	t3
S3	C4	S1	action_4	t4
S3	C1	S3	action_3	t33
S3	C2	S3	action_3	t33

备注：

S1:InitialState	C1:build connect	action_1:LED1 off,LED2 on
S2:connect ok state	C2:key on	action_2:send Key-on to host
S3:wait host's response of key state	C3:receive host's response of key	action_3:LED1 blinking 3 times
	C4:disconnect	action_4:LED1 blinking,LED2 off

5. 主要代码：

while (1)
{
static u16 cnt = 0;
static u8 buf[FRAME_SIZE], a_byte;
bool ret_t = FALSE;
Data_Frame pData_Frame_t;
check_timer();
if(stm_connect_send_heartbeat_delay == 0)
{
send_CMD_data(ID_HEARTBEAT);
stm_connect_send_heartbeat_delay = 10;// 1000ms
}
if(Get_Key1_Press() == TRUE)
{
if(g_condition_demo.key_c_can_be_change_flg == TRUE)
{
g_condition_demo.cond= CONDITION_2;
g_condition_demo.changed_flg = TRUE;
}
}
else
{
//read ringbuffer
while(g_fifo_Rx.empty_flg == FALSE)
{ /* Do something with buf... */
//assert(buf == cnt);
ret_t = read_fifo(&g_fifo_Rx,&a_byte);
buf[cnt++] = a_byte;
if((a_byte == 0x0A)&&(ret_t == TRUE))// it's a frame data
{
if(check_data_whole(buf,&pData_Frame_t,FRAME_SIZE))
if(check_data_vaild(&pData_Frame_t))// check checksum is correct
{
if(pData_Frame_t.id == ID_HEARTBEAT)
{
if(g_condition_demo.heartbeat_build_can_be_change_flg == TRUE)
{
stm_lost_heartbeat_3s_delay = 3;
g_condition_demo.cond= CONDITION_1;
g_condition_demo.changed_flg = TRUE;
}
cnt = 0;
break;
}
else if(pData_Frame_t.id == ID_COMFIRM_KEY)
{
if(g_condition_demo.confirmed_response_c_can_be_change_flg == TRUE)
{
g_condition_demo.cond= CONDITION_3;
g_condition_demo.changed_flg = TRUE;
}
cnt = 0;
break;
}
else if(pData_Frame_t.id == ID_RESEND_REQUEST)
{
send_CMD_data(ID_HEARTBEAT);
stm_connect_send_heartbeat_delay = 10;// 1000ms
cnt = 0;
break;
}
else{
SEGGER_RTT_printf(0,"invaild IDrn");
break;
}
}
else
{//send re-send command to Host & stop send heartbeat
send_CMD_data(ID_RESEND_REQUEST);
break;
}
}
else
{
}
//printf("%d Read: %dn", cnt,buf);
}
}
if(g_condition_demo.changed_flg == TRUE)
{
state_machine_step(&g_state_machine,g_condition_demo.cond);
g_condition_demo.changed_flg = FALSE;
}
}

复制代码

6. python实现上位机软件

'''
this python shell is serial server for Coslink protocol
'''
import serial # install cmd: python -m pip install pyserial --user(python -m pip install --upgrade pip --user)
from datetime import datetime
from threading import Timer
send_1000ms_arrive_flg = False
heartbeat_received_flg = False
def sendKeepConnectTimer(inc):
global send_1000ms_arrive_flg
print(datetime.now().strftime("%Y-%m-%d %H:%M:%S"))
send_1000ms_arrive_flg = True
t = Timer(inc,sendKeepConnectTimer,(inc,))
t.start()
def checksum(data_r):
#n_position = data_r.list('n')
#in_checksum = data_r[n_position-1]
sum = 0
for i in data_r:
sum = sum+i
sum = sum - 9 # -10+1
print(sum)
if(sum%256 == 0):
return True
else:
return False
def run(svr_status):
global connect_status, send_1000ms_arrive_flg,heartbeat_received_flg
first_start_timer_flg = True
arr_ID_resend= [170,1,0,0,84,10]
arr_ID_KeyResponse= [170,1,2,0,82,10]
seri = serial.Serial()
seri.baudrate = 115200
seri.port = 'COM18'
print("Serial Server Application")
print("SCN:DAA31028AAA Build Time:2020-11-11 23:00")
print("waiting for open serial ...")
#seri.parity = 'ODD'
while True:
try:
seri.open()
print('serial open open successfully ')
if(first_start_timer_flg):
first_start_timer_flg = False
sendKeepConnectTimer(3)
while seri.isOpen():
try:
#print("waiting for connect data ...")
if seri.inWaiting():
data_r = seri.readline()
#print('reveive type of data:',type(data_r))
print('reveive hex data:',data_r)
#print('reveive hex data[2]:%x',data_r[2])
if(checksum(data_r) == True):
if(data_r[2] == 3):
print('ERROR:!!!The connect is not built or lost !!!')
elif(data_r[2] == 85):
heartbeat_received_flg = True
print('send heartbeat',data_r)
seri.write(data_r) #send heartbeat
elif(data_r[2] == 1):
print('send KeyResponse',arr_ID_KeyResponse)
seri.write(arr_ID_KeyResponse) #send _KeyResponse
else:
print('receive wrong checksum,so send re-send resquest',arr_ID_resend)
seri.write(arr_ID_resend)
else:
if(send_1000ms_arrive_flg == True):
send_1000ms_arrive_flg = False
#print('time arrive more than 1000ms!!!')
if(heartbeat_received_flg == True):
heartbeat_received_flg = False
else:
print('ERROR:!!!clent heartbeat package is lost for more than 3000ms!!!')
#print('serial no data...')
except:
print("seri except & disconnectrn")
break
#else:
#handle data
#print("handle datarn")
#seri.close()
except:
seri.close()#pass
if(send_1000ms_arrive_flg == True):
send_1000ms_arrive_flg = False
print('Open Uart failed!!!')
#break
else:
if(send_1000ms_arrive_flg == True):
send_1000ms_arrive_flg = False
print('Open Uart failed!!!')
if __name__ == "__main__":
#sendKeepConnectTimer(5)=
#data_test=[170,1,85,15,240,10]
run(0)