优化ai对话,添加了tts 重连机制,asr send bin 强制控制逻辑。
author孙春阳 <snickersun@163.com>
Thu, 14 Aug 2025 04:52:03 +0000 (12:52 +0800)
committer孙春阳 <snickersun@163.com>
Thu, 14 Aug 2025 04:52:03 +0000 (12:52 +0800)
目前还有tts容易断连问题,需要和服务器联调

code/sdk/application/wearable/fd230app/nativeui/aivoice/src/AiVoicePresenter.cpp
code/sdk/application/wearable/service/ble_service/bleservice/bleService.c
code/sdk/application/wearable/service/ble_service/bleservice/bleService.h
code/sdk/drivers/peripheral/slip/cat1Service/cat1_ai_server.cpp
code/sdk/drivers/peripheral/slip/cat1Service/cat1_ws.cpp
code/sdk/drivers/peripheral/slip/tcpmanager/tcp_simple_manager.c
code/sdk/middleware/services/media/foundation/sample/wrapper/voice_call_volte_sample_wrapper.cpp

index 96efd66..7674b98 100644 (file)
@@ -35,11 +35,12 @@ namespace OHOS
     extern "C"
     {
 #endif
-
 #include "pm.h"
 #include "pm_veto.h"
 #include <asm/delay.h>
 #include <aivoice/include/AiVoiceModelC.h>
+#include "bleService.h" 
+extern void set_need_tts_reconnect(uint8_t flag);
 
         void remove_4GAI_vote(void)
         {
@@ -331,6 +332,7 @@ namespace OHOS
             break;
 
         case AIVOICE_LISTENING_VIEWID:
+            set_stop_send_ai_flag(1); // 停止发送AI数据
             model->StopListenAndRecordingUserQuestion();
             model->stop_listening_timer();
             break;
@@ -340,12 +342,14 @@ namespace OHOS
             break;
 
         case AIVOICE_RETURN_RESULT_TEXT_VIEWID:
+            set_need_tts_reconnect(0); // 停止tts重连   
             AiVoiceModel::GetInstance().ClearQuestionAndAiAnswerStr(); // 清空上次用户问题和Ai回答字符串
             AiVoiceModel::GetInstance().SetIsClickable(false);
             AiVoiceModel::GetInstance().SetIsLimitDisplayed(false);
             AiVoiceModel::GetInstance().stop_ai4g_tts_play_timer();
             AiVoiceModel::GetInstance().stop_update_ai_answer_timer();
             break;
+            
         default:
             // Handle other cases if needed
             break;
@@ -357,8 +361,11 @@ namespace OHOS
             model->QuerySelectedAiModelIsConnect();
             AiVoiceModel::GetInstance().start_ai_isready_timer();
             break;
-
+        case AIVOICE_RETURN_RESULT_TEXT_VIEWID:
+            set_need_tts_reconnect(1); // 开始tts重连   
+            break;
         case AIVOICE_LISTENING_VIEWID:
+            set_stop_send_ai_flag(0); // 开始发送AI数据
             AiVoiceModel::GetInstance().StartListenAndRecordingUserQuestion();
             AiVoiceModel::GetInstance().start_listening_timer();
             break;
index f3657f5..1b2097d 100644 (file)
@@ -2405,7 +2405,11 @@ void send_Ai_record(uint16_t len,uint8_t *data,uint32_t count)
 
     LOS_Msleep(AI_interval+get_ble_interval());
 }
-
+static uint8_t stop_send_ai_flag=0;
+void set_stop_send_ai_flag(uint8_t flag)
+{
+    stop_send_ai_flag=flag;
+}
 static void blesrvAi_send_service_task(void const *arg)
 {
     unused(arg);
@@ -2490,7 +2494,11 @@ static void blesrvAi_send_service_task(void const *arg)
                     }
                     else
                     {
-                        send_ai_bin(mail.msg_data,mail.msg_len);
+                        if(stop_send_ai_flag==0)
+                        {
+                            send_ai_bin(mail.msg_data,mail.msg_len);
+                        }
+                        
                     }
                     
                 }
@@ -2615,7 +2623,7 @@ int send_msg_data_into_Ai_msg_queue(void *buff,uint16_t len,uint8_t type)
     }
     blesrv_mail.msg_data=0;
 
-    if(type == 200 || type == 201)
+    if(type == 200 || type == 201) // 200:AI 文本数据 201:AI 音频数据结束
     {
     
         blesrv_mail.service_handle = type;
index ed17e73..a7905eb 100644 (file)
@@ -115,6 +115,8 @@ void set_AI_interval(uint8_t data);
 
 int get_ai_osal_msg_queue_is_full(void);
 
+void set_stop_send_ai_flag(uint8_t flag);
+
 #ifdef __cplusplus
 }
 #endif
index ea8cb7c..eecf672 100644 (file)
@@ -322,6 +322,14 @@ static int tts_client_ping=0;
 
 static int chat_client_ping=0;
 
+static int need_tts_reconnect=0;
+
+void set_need_tts_reconnect(uint8_t flag)
+{
+    need_tts_reconnect=flag;
+}
+
+
 WebSocketClient * get_ws_client_by_handle(void* handle)
 {
     if (chat_client != NULL && chat_client->tcp_handle == handle)
@@ -1391,13 +1399,13 @@ bool aer_result_analysis(char *buf)
         #ifdef ASR_HODE
         memcpy_s(&result_str_asr[strlen(result_str_asr)],AI_RESULT_STR_LEN/2,result_str,strlen(result_str));
 
-        // if(asr_is_stop==1 && strlen(result_str_asr))
-        // {
-        //     send_ai_txt(0,result_str_asr);
-        //     ble_set_ai_Question_str(result_str_asr);
+        if(asr_is_stop==1 && strlen(result_str_asr))
+        {
+            send_ai_txt(0,result_str_asr);
+            ble_set_ai_Question_str(result_str_asr);
 
-        //     memset_s(result_str_asr,AI_RESULT_STR_LEN,0,AI_RESULT_STR_LEN);
-        // }
+            memset_s(result_str_asr,AI_RESULT_STR_LEN,0,AI_RESULT_STR_LEN);
+        }
 
         
         #else
@@ -1965,7 +1973,7 @@ void bk_websocket_tts(void *data) {
     uint32_t start_time = (uint32_t)osKernelGetTickCount();
     uint32_t retry_count = 0;
     
-    TTS_SATRT:
+TTS_SATRT:
     uapi_cat1_keep_wakeup(1);
 
     WebSocketMessageType type;
@@ -2093,7 +2101,7 @@ void bk_websocket_tts(void *data) {
         // lwip_setsockopt(tts_client->sockfd, SOL_SOCKET, SO_RCVTIMEO, &timeout, sizeof(timeout));
 
         uint32_t last_activity = (uint32_t)osKernelGetTickCount();
-        
+        // 循环接收tts_msg_queue数据 给AI发送数据
         while (disconnect_server == 0) {
             // 检查超时
             if ((uint32_t)osKernelGetTickCount() - last_activity > AI_RECEIVE_TIMEOUT_MS) {
@@ -2112,21 +2120,25 @@ void bk_websocket_tts(void *data) {
 
                 break;
             }
-            
+            //接收数据
             memset_s(tts_recv_buf, 1024 * 4, 0, 1024 * 4);
             int len = receive_websocket_message_from_queue(&tts_msg_queue, (unsigned char *)tts_recv_buf, 1024 * 4, &type, AI_RECEIVE_TIMEOUT_MS);
             
             if (type == WS_CLOSE || len == -1 || tts_reconnect == 1) {
                 WEARABLE_LOGE(WEARABLE_LOG_MODULE_APP, "TTS WebSocket connection closed");
+                if(need_tts_reconnect==1) 
+                {
+                    tts_reconnect = 1;
+                }
                 break;
             }
-            
+            //发送数据到AI
             if (len > 0) {
                 last_activity = (uint32_t)osKernelGetTickCount();
                 
                 if (type == WS_BINARY) {
                     if (bk_websocket_tts_tick == 0) {
-                        set_AIStreamend(0);
+                        set_AIStreamend(0);// 停止AI播放音频
                         bk_websocket_tts_tick = (uint32_t)osKernelGetTickCount();
                     }
                     send_msg_data_into_Ai_msg_queue((uint8_t *)tts_recv_buf, len, 200);
index d02329c..8658942 100644 (file)
@@ -463,11 +463,23 @@ int send_handshake(WebSocketClient *client,const char *headers,char *response,in
 // 发送 WebSocket 消息
 int send_websocket_message(WebSocketClient *client, uint8_t *message, size_t message_len, WebSocketMessageType type) {
 
+    int ret_tcp_send_data=0;
     if(client==NULL||message==NULL )
     {
          WS_LOG_ERROR("send_websocket_message client or message is NULL %s %s", client, message);
         return -1;
     }
+    
+    // 检查连接状态
+    if (client->tcp_handle == NULL) {
+        WS_LOG_ERROR("send_websocket_message: TCP handle is NULL");
+        return -1;
+    }
+    
+    if (client->handshake == 0) {
+        WS_LOG_ERROR("send_websocket_message: WebSocket handshake not completed");
+        return -1;
+    }
     uint8_t* frame=( uint8_t *)aic_zalloc(message_len + 50);
     int offset = 0;
     int ret = -1;
@@ -514,14 +526,23 @@ int send_websocket_message(WebSocketClient *client, uint8_t *message, size_t mes
 
     // 发送帧
     WS_LOG_DEBUG("offset %d", offset);
-    int ret_tcp_send_data=tcp_send_data(client->tcp_handle, frame, offset);
+    
+    // 再次检查TCP句柄有效性
+    if (client->tcp_handle == NULL) {
+        WS_LOG_ERROR("send_websocket_message: TCP handle became NULL before send");
+        ret = -1;
+        goto send_websocket_message_ERROR;
+    }
+    
+    ret_tcp_send_data=tcp_send_data(client->tcp_handle, frame, offset);
     if (ret_tcp_send_data < 0) {
         WS_LOG_ERROR("lwip_send message failed %d", ret_tcp_send_data);
         ret =  -1;
         goto send_websocket_message_ERROR;
     }
     ret = 0;
-    send_websocket_message_ERROR:
+
+send_websocket_message_ERROR:
     bk_sleep(message_len/1000+1);
     free(frame);
     // bk_sleep(message_len/1000+1);
@@ -561,10 +582,17 @@ void _tcp_on_connect_cb(tcp_handle_t handle, bool success)
                             continue;
                         } else {
                             WS_LOG_INFO("WebSocket握手失败,已达到最大重试次数");
-                            tcp_manager_close(client->tcp_handle);
+                            // 安全关闭连接
+                            if (client->tcp_handle != NULL) {
+                                tcp_manager_close(client->tcp_handle);
+                                client->tcp_handle = NULL;
+                            }
+                            client->handshake = 0;
+                            return;
                         }
                     } else {
                         WS_LOG_INFO("WebSocket握手发送成功");
+                        client->handshake = 1;
                         break;
                     }
                 }
@@ -582,21 +610,56 @@ void _tcp_on_connect_cb(tcp_handle_t handle, bool success)
 void _tcp_on_disconnect_cb(tcp_handle_t handle)
 {
     WS_LOG_INFO("tcp_on_disconnect_cb %d", handle);
+    
+    // 查找对应的WebSocket客户端并清理状态
+    extern WebSocketClient * get_ws_client_by_handle(void* handle);
+    WebSocketClient * client = get_ws_client_by_handle(handle);
+    if(client != NULL)
+    {
+        WS_LOG_INFO("Cleaning up WebSocket client for disconnected handle %d", handle);
+        
+        // 清理客户端状态
+        client->handshake = 0;
+        client->tcp_handle = NULL;
+        
+        // 重置解析器状态
+        if (client->parser) {
+            ws_parser_reset(client->parser);
+        }
+    }
 }
 void _tcp_on_data_received_cb(tcp_handle_t handle, uint8_t* data, uint16_t datalen)
 {
     WS_LOG_INFO("tcp_on_data_received_cb 0x%0X %d", handle, datalen);
+    
+    // 参数验证
+    if (data == NULL || datalen == 0) {
+        WS_LOG_ERROR("Invalid data parameters: data=%p, datalen=%d", data, datalen);
+        return;
+    }
+    
     extern WebSocketClient * get_ws_client_by_handle(void* handle);
     WebSocketClient * client = get_ws_client_by_handle(handle);
     if(client==NULL)
     {
-        WS_LOG_ERROR("get_ws_client failed");
+        WS_LOG_ERROR("get_ws_client failed for handle 0x%0X", handle);
         return;
     }
-       // 使用WebSocket解析器处理数据
+    
+    // 检查客户端状态
+    if (client->tcp_handle == NULL || client->parser == NULL) {
+        WS_LOG_ERROR("Client in invalid state: tcp_handle=%p, parser=%p", client->tcp_handle, client->parser);
+        return;
+    }
+    
+    // 使用WebSocket解析器处理数据
     int processed = ws_process_received_data(client, data, datalen);
     if (processed < 0) {
         WS_LOG_ERROR("WebSocket packet processing failed");
+        // 解析失败时重置解析器,避免状态不一致
+        if (client->parser) {
+            ws_parser_reset(client->parser);
+        }
         return;
     }
    
@@ -691,16 +754,24 @@ void lwip_close_websocket_client(WebSocketClient *client) {
 
         WS_LOG_INFO("lwip_close_websocket_client %s %s", client->host, client->path);
         
+        // 先关闭TCP连接,避免数据继续流入
+        if (client->tcp_handle != NULL) {
+            tcp_manager_close(client->tcp_handle);
+            client->tcp_handle = NULL;
+        }
+        
         // 释放WebSocket解析器
         if (client->parser) {
             ws_parser_destroy(client->parser);
             client->parser = NULL;
         }
         
-        tcp_manager_close(client->tcp_handle);
+        // 清理客户端状态
+        client->handshake = 0;
+        client->packet_callback = NULL;
+        
         free(client);
     }
-    
 }
 
 // 旧的WebSocket解析器函数已移除,现在使用新的解析器
@@ -800,9 +871,14 @@ int ws_process_received_data(WebSocketClient *client, const uint8_t *data, size_
            len, client->host, client->path);
     
     // 循环处理数据,支持连续帧
-    while (remaining_len > 0) {
-        WS_LOG_DEBUG("ws_process_received_data: processing remaining %zu bytes, current offset: %zu", 
-               remaining_len, current_data - data);
+    size_t original_len = remaining_len;
+    size_t loop_count = 0;
+    const size_t max_loops = original_len + 10; // 防止无限循环
+    
+    while (remaining_len > 0 && loop_count < max_loops) {
+        loop_count++;
+        WS_LOG_DEBUG("ws_process_received_data: processing remaining %zu bytes, current offset: %zu, loop %zu", 
+               remaining_len, current_data - data, loop_count);
         
         // 在解析前打印当前解析器状态
         WS_LOG_DEBUG("ws_process_received_data: parser state before parsing:");
@@ -815,7 +891,10 @@ int ws_process_received_data(WebSocketClient *client, const uint8_t *data, size_
         {
             WS_LOG_ERROR("ws_process_received_data: parsing error, reset parser,find next frame. Remaining: %zu bytes", remaining_len);
             ws_parser_reset(parser);
-            return -1;
+            // 不要立即返回,尝试跳过错误数据继续处理
+            current_data++;
+            remaining_len--;
+            continue;
         }
 
         // 更新处理进度
@@ -855,16 +934,16 @@ int ws_process_received_data(WebSocketClient *client, const uint8_t *data, size_
                 // 根据操作码处理不同类型的消息
                 switch (header->opcode) {
                     case WS_PING:
-                        WS_LOG_DEBUG("Received WebSocket PING");
+                        WS_LOG_WARNING("Received WebSocket PING");
                         // TODO: 自动回复PONG
                         break;
                         
                     case WS_PONG:
-                        WS_LOG_DEBUG("Received WebSocket PONG");
+                        WS_LOG_WARNING("Received WebSocket PONG");
                         break;
                         
                     case WS_CLOSE:
-                        WS_LOG_DEBUG("Received WebSocket CLOSE");
+                        WS_LOG_WARNING("Received WebSocket CLOSE");
                         // 调用回调函数处理关闭帧
                         if (client->packet_callback) {
                             client->packet_callback((uint8_t*)payload, payload_len, (WebSocketMessageType)WS_CLOSE);
@@ -915,6 +994,12 @@ int ws_process_received_data(WebSocketClient *client, const uint8_t *data, size_
         }
     }
     
+    // 检查是否达到循环限制
+    if (loop_count >= max_loops) {
+        WS_LOG_ERROR("ws_process_received_data: reached maximum loop count %zu, forcing exit", max_loops);
+        ws_parser_reset(parser);
+    }
+    
     WS_LOG_DEBUG("ws_process_received_data: total processed: %d bytes", total_processed);
     return total_processed;
 }
index 03cec75..c28164f 100644 (file)
@@ -856,7 +856,7 @@ static void tcp_manager_io_thread(void* arg)
             {
                 if( conn->send_queue_count > 0)
                 {
-                    printf("process_send_queue tcp_manager_io_thread %d %d\n",conn->send_queue_count,conn->sockfd);
+                    printf("process_send_queue tcp_manager_io_thread send_queue_count:%d sockfd:%d\n",conn->send_queue_count,conn->sockfd);
                 }
                 
             }
index b4c0039..330e4bf 100755 (executable)
@@ -1249,13 +1249,12 @@ uint8_t AudioCaptureTask_is_run=0;
 
 uint8_t stop_AudioCaptureProcess_AI(void)
 {
-    if(AudioCaptureTask_is_run)
+    if (AudioCaptureTask_is_run)
     {
-        g_sample_needStop=1;
-return 1;
+        g_sample_needStop = 1;
+        return 1;
     }
-return 0;
-   
+    return 0;
 }
 
 static void AudioCaptureProcess_AI(void* arg)