#include "NativeRegisterManager.h"
#include "wearable_log.h"
+#ifdef SUPPORT_BLE_SERVICE
+#include "bleEvent.h"
+#endif
+
namespace OHOS {
REGIST_SLICE(VIEW_EPHEMERIS_DOWNLOAD, EphemerisDownloadView,
EphemerisDownloadPresenter);
//点击下载按键
if (strcmp(view.GetViewId(), DOWNLOAD_ID) == 0) {
//进行下载
- EphemerisDownloadModel::GetInstance().DownloadEphemeris();
+ // EphemerisDownloadModel::GetInstance().DownloadEphemeris();
+ #ifdef SUPPORT_BLE_SERVICE
+ ble_request_epo_file();
+ #endif
}
//点击返回按键
else if (strcmp(view.GetViewId(), BACK_ID) == 0) {
return;
}
+ if (status)
+ showPhoneConnectedStatus->SetSrc(DROPDOWN_PHONECONNECTED_IMAGE);
+ else
showPhoneConnectedStatus->SetSrc(DROPDOWN_PHONEDISCONNECTED_IMAGE);
if (GetConnectedStatusView() != nullptr)
void MsgCenterNotifybleTransmit(const Topic topic, Request* req)
{
- #ifdef SUPPORT_BLE_SERVICE
- unused(topic);
- uint8_t* data=(uint8_t*)req->data;
+ #ifdef SUPPORT_BLE_SERVICE
+ unused(topic);
+ uint8_t* data=(uint8_t*)req->data;
+ WEARABLE_LOGI(WEARABLE_LOG_MODULE_APP, "MsgCenterNotifybleTransmit:%d",data[0]);
+
-
- if(data[0]==TRANS_FREE)
- {
- if(transmittingISShow())
- NotificationManager::GetInstance()->StopNotify();
- }
- else if (data[0]==IS_NEED_CAT1_OTA)
- {
-
- NotificationManager::GetInstance()->SetNotifyPriority(PRIORITY_ONE);
-
- // NotificationManager::GetInstance()->ShowNotify(showCat1UpdataNotify, (uint16_t)topic, true);
- }
- else if (data[0]==FILE_IS_DONE)
- {
- set_transmittingShow_falg(data[0]);
- }
- else
- {
- set_transmittingShow_falg(data[0]);
-
- NotificationManager::GetInstance()->SetNotifyPriority(PRIORITY_ONE);
-
- NotificationManager::GetInstance()->ShowNotify(showtransmitNotify, (uint16_t)topic, true);
- }
- // unused(req);
- #endif
+ if(data[0]==TRANS_FREE)
+ {
+ if(transmittingISShow())
+ NotificationManager::GetInstance()->StopNotify();
+ }
+ else if (data[0]==IS_NEED_CAT1_OTA)
+ {
+
+ NotificationManager::GetInstance()->SetNotifyPriority(PRIORITY_ONE);
+
+ // NotificationManager::GetInstance()->ShowNotify(showCat1UpdataNotify, (uint16_t)topic, true);
+ }
+ else if (data[0]==FILE_IS_DONE)
+ {
+ set_transmittingShow_falg(data[0]);
+ }
+ else
+ {
+ set_transmittingShow_falg(data[0]);
+
+ NotificationManager::GetInstance()->SetNotifyPriority(PRIORITY_ONE);
+
+ NotificationManager::GetInstance()->ShowNotify(showtransmitNotify, (uint16_t)topic, true);
+ }
+ #endif
}
void MsgCenterNotifyChargePlugProc(const Topic topic, Request* req)
}
+
+ void creat_down_EPO()
+ {
+ printf("creat_down_EPO\n");
+ if (Title == nullptr) {
+ Title = new UILabelExt();
+ }
+ Title->SetPosition(158, 51);
+ Title->Resize(800, 88);
+ Title->SetViewId("label1");
+ Title->SetStyle(STYLE_TEXT_FONT, 30);
+
+ Title->SetText("星历下载中");
+ Title->SetLineBreakMode(UILabel::LINE_BREAK_ADAPT);
+ Title->SetAlign(TEXT_ALIGNMENT_CENTER, TEXT_ALIGNMENT_CENTER);
+ Title->SetRollSpeed(0);
+ Title->SetFont(DEFAULT_VECTOR_FONT_FILENAME, 30);
+
+ // Add to rootView
+ UIViewGroup_transmitting->Add(Title);
+ Title->SetVisible(true);
+
+
+ if (label_ == nullptr) {
+ label_ = new UILabelExt();
+ }
+ label_->SetPosition(60, 281);
+ label_->Resize(345, 101);
+ label_->SetStyle(STYLE_TEXT_FONT, 30);
+ // Set additional property
+ label_->SetText("正在下载星历，请不要做其他操作或关闭手表");
+ label_->SetLineBreakMode(UILabel::LINE_BREAK_WRAP);
+ label_->SetTextColor(Color::GetColorFromRGBA(0xAA, 0xAA, 0xAA, 0xff));
+ label_->SetAlign(TEXT_ALIGNMENT_CENTER, TEXT_ALIGNMENT_CENTER);
+ label_->SetRollSpeed(0);
+ label_->SetFont(DEFAULT_VECTOR_FONT_FILENAME, 30);
+
+ // Add to rootView
+ UIViewGroup_transmitting->Add(label_);
+ label_->SetVisible(true);
+
+ CreatetransmittingAnimator();
+
+ UIViewGroup_transmitting->Add(downImageAnimatorView);
+ }
+
+ void creat_down_EPO_succ()
+ {
+ printf("creat_down_EPO_succ\n");
+ if (Title == nullptr) {
+ Title = new UILabelExt();
+ }
+ Title->SetPosition(173, 51);
+ Title->Resize(800, 88);
+ Title->SetStyle(STYLE_TEXT_FONT, 30);
+
+ Title->SetText("星历下载成功");
+ Title->SetLineBreakMode(UILabel::LINE_BREAK_ADAPT);
+ Title->SetAlign(TEXT_ALIGNMENT_CENTER, TEXT_ALIGNMENT_CENTER);
+ Title->SetRollSpeed(0);
+ Title->SetFont(DEFAULT_VECTOR_FONT_FILENAME, 30);
+
+ Title->SetX((466 - Title->GetWidth()) / 2);
+
+ // Add to rootView
+ UIViewGroup_transmitting->Add(Title);
+ Title->SetVisible(true);
+
+
+ if (label_ == nullptr) {
+ label_ = new UILabelExt();
+ }
+ label_->SetPosition(60, 281);
+ label_->Resize(345, 101);
+ label_->SetViewId("label2");
+ label_->SetStyle(STYLE_TEXT_FONT, 30);
+
+ // Set additional property
+ label_->SetText("星历下载成功，请稍等...");
+ label_->SetLineBreakMode(UILabel::LINE_BREAK_WRAP);
+ label_->SetTextColor(Color::GetColorFromRGBA(0xAA, 0xAA, 0xAA, 0xff));
+ label_->SetAlign(TEXT_ALIGNMENT_CENTER, TEXT_ALIGNMENT_CENTER);
+ label_->SetRollSpeed(0);
+ label_->SetFont(DEFAULT_VECTOR_FONT_FILENAME, 30);
+
+ if (transmitting_img == nullptr)
+ {
+ transmitting_img = new UIImageView();
+ }
+ transmitting_img->SetPosition(193, 150,80, 80);
+ const char *imagesPath = TRANSMITTING_SUCC_PATH;
+ transmitting_img->SetSrc(imagesPath);
+ UIViewGroup_transmitting->Add(transmitting_img);
+
+ uint32_t tick;
+ osStatus_t retTimer;
+ if(transmitting_timer!=NULL)
+ {
+ osTimerStop(transmitting_timer);
+ osTimerDelete(transmitting_timer);
+ transmitting_timer = NULL;
+ }
+ transmitting_timer = osTimerNew((osTimerFunc_t)dialtimerCallBack, osTimerPeriodic, NULL, NULL);
+ tick =3000;// osMs2Tick(1000);
+ retTimer = osTimerStart(transmitting_timer, tick);
+ if (retTimer != osOK && transmitting_timer != NULL) {
+ WEARABLE_LOGE(WEARABLE_LOG_MODULE_APP,"StartCallTimer osTimerStart failed, ret = %lx!!", retTimer);
+ osTimerDelete(transmitting_timer);
+ transmitting_timer = NULL;
+ return ;
+ }
+
+ // Add to rootView
+ UIViewGroup_transmitting->Add(label_);
+ label_->SetVisible(true);
+ }
+
+ void creat_down_EPO_fail()
+ {
+ printf("creat_down_EPO_fail\n");
+ if (Title == nullptr)
+ {
+ Title = new UILabelExt();
+ }
+ Title->SetPosition(173, 51);
+ Title->Resize(800, 88);
+ Title->SetViewId("label1");
+ Title->SetStyle(STYLE_TEXT_FONT, 30);
+
+ // Set additional property
+ Title->SetText("星历下载失败");
+ Title->SetLineBreakMode(UILabel::LINE_BREAK_ADAPT);
+ Title->SetAlign(TEXT_ALIGNMENT_CENTER, TEXT_ALIGNMENT_CENTER);
+ Title->SetRollSpeed(0);
+ Title->SetFont(DEFAULT_VECTOR_FONT_FILENAME, 30);
+ Title->SetX((466 - Title->GetWidth()) / 2);
+
+ // Add to rootView
+ UIViewGroup_transmitting->Add(Title);
+ Title->SetVisible(true);
+
+
+ if (label_ == nullptr)
+ {
+ label_ = new UILabelExt();
+ }
+ label_->SetPosition(60, 281);
+ label_->Resize(345, 101);
+ label_->SetStyle(STYLE_TEXT_FONT, 30);
+ Title->SetText("星历下载失败,请重新下载");
+
+
+ label_->SetLineBreakMode(UILabel::LINE_BREAK_WRAP);
+ label_->SetTextColor(Color::GetColorFromRGBA(0xAA, 0xAA, 0xAA, 0xff));
+ label_->SetAlign(TEXT_ALIGNMENT_CENTER, TEXT_ALIGNMENT_CENTER);
+ label_->SetRollSpeed(0);
+ label_->SetFont(DEFAULT_VECTOR_FONT_FILENAME, 30);
+
+ UIViewGroup_transmitting->Add(label_);
+ label_->SetVisible(true);
+ }
bool transmittingISShow(void)
{
{
UIViewGroup_transmitting->RemoveAll();
}
- uapi_dev_runtime_set_transmitting_status(1);
-
+ uapi_dev_runtime_set_transmitting_status(1);
+ printf("Transmitting_falg:%d",Transmitting_falg);
#ifdef SUPPORT_BLE_SERVICE
switch (Transmitting_falg)
{
case TRANS_BIN_FAIL:
creat_down_BIN_fail();
break;
+ case TRANS_EPO_TRANSMITTING:
+ creat_down_EPO();
+ break;
+ case TRANS_EPO_SUCC:
+ creat_down_EPO_succ();
+ break;
+ case TRANS_EPO_FAIL:
+ creat_down_EPO_fail();
+ break;
default:
break;
&& Transmitting_falg!=TRANS_OTA_TRANSMITTING
&& Transmitting_falg!=TRANS_MP3_TRANSMITTING
&& Transmitting_falg!=START_CAT1_OTA
- && Transmitting_falg!=TRANS_BIN_TRANSMITTING)
+ && Transmitting_falg!=TRANS_BIN_TRANSMITTING
+ && Transmitting_falg!=TRANS_EPO_TRANSMITTING)
{
NotificationManager::GetInstance()->StopNotify();
}
static void CreatetransmittingAnimator(void)
{
- WEARABLE_LOGE(WEARABLE_LOG_MODULE_APP, "ShutDownAnimatorView::ShowUpShutDownAnimatorView");
imageInfo_dwon = CreateImageInfoArray(ShutDownPaths,SHUTDOWN_IMAGE_ANIMATOR_VIEW_NUM_FRAMES, "bin");
if (downImageAnimatorView == nullptr)
IS_NEED_CAT1_OTA,
TRANS_FREE,
+
+ TRANS_EPO_TRANSMITTING,
+ TRANS_EPO_SUCC,
+ TRANS_EPO_FAIL,
} transmitting_type;
#include "wearable_log.h"
#include <stdint.h>
#include <stdio.h>
+#include <string.h>
#include "stdbool.h"
#include "bts_br_gap.h"
#define UP_DATE_FWPKG_PATH UPG_FILE_NAME
#endif
+//gnss文件目录
+#define GNSS_DATA_DIR "/user/userdata/gnss"
+
//MP3文件重命名
static errcode_t create_mp3_name(char* path,char*dir_path ,char* file_name)
input_filename, output_file_path);
return ERRCODE_SUCC;
}
+
+static errcode_t get_epo_file_path(char* input_filename ,uint32_t input_size, char* output_file_path , uint32_t max_output_size)
+{
+ if (input_filename == NULL || output_file_path == NULL) {
+ WEARABLE_LOGE(WEARABLE_LOG_MODULE_APP, "[BLE_EVENT_ERROR] get_epo_file_path: NULL pointer detected - input_filename=%p, output_file_path=%p",
+ input_filename, output_file_path);
+ return ERRCODE_FAIL;
+ }
+
+ if (input_size == 0 || max_output_size == 0) {
+ WEARABLE_LOGE(WEARABLE_LOG_MODULE_APP, "[BLE_EVENT_ERROR] get_epo_file_path: Invalid size parameters - input_size=%u, max_output_size=%u",
+ input_size, max_output_size);
+ return ERRCODE_FAIL;
+ }
+
+ uapi_create_dir(GNSS_DATA_DIR);
+
+ int ret = sprintf_s(output_file_path, max_output_size, "%s/%s", GNSS_DATA_DIR, input_filename);
+ if (ret < 0) {
+ WEARABLE_LOGE(WEARABLE_LOG_MODULE_APP, "[BLE_EVENT_ERROR] get_epo_file_path: sprintf_s failed - ret=%d, filename='%s', max_size=%u",
+ ret, input_filename, max_output_size);
+ return ERRCODE_FAIL;
+ }
+
+ WEARABLE_LOGD(WEARABLE_LOG_MODULE_APP, "[BLE_EVENT_DEBUG] get_epo_file_path: Success - input='%s', output='%s'",
+ input_filename, output_file_path);
+ return ERRCODE_SUCC;
+}
static errcode_t get_custom_dial_file_path(uint16_t id,char *filename,uint32_t input_size, char* output_file_path , uint32_t max_output_size)
{
if( id >=5 || output_file_path == NULL || filename == NULL)
#ifdef BLE_LOG
WEARABLE_LOGD(WEARABLE_LOG_MODULE_APP,"bigDataBufState:%d and sendbigDataBufState %d\r\n",bigDataBufState,sendbigDataBufState);
#endif
+ WEARABLE_LOGE(WEARABLE_LOG_MODULE_APP,"bigDataBufState:%d and sendbigDataBufState %d\r\n",bigDataBufState,sendbigDataBufState);
//if(Bleconnect == true )
{
ret = svr_ble_bt_gatts_noyify_zhihui(buf,len);
}
}
+// 设备 -> APP：请求星历文件通知
+void ble_request_epo_file(void)
+{
+ WEARABLE_LOGE(WEARABLE_LOG_MODULE_APP, "ble_request_epo_file");
+ uint8_t value[19] = {0};
+ uint32_t tmp = 1;
+ uint16_t tmp_data = 1;
+
+ //72 05 40 01 00 00 00 01 00 00 00 01 00 00 00 01 00 00 <CRC>
+
+ value[0] = 0x72; // direction: device -> app
+ value[1] = 0x05; // type
+ value[2] = 0x40; // id = REQUEST_EPO_FILE_CMD
+ memcpy(value + 3, &tmp, 4); // package_count
+ memcpy(value + 7, &tmp, 4); // total_data_len
+ memcpy(value + 11, &tmp, 4);// package_index
+ memcpy(value + 15, &tmp_data, 2); // package_len (no payload)
+ // value[17] 可扩展: 0=普通请求，1=强制刷新；当前默认0
+ value[17] = 0;
+ value[18] = Calculate_CRC(value, 18);
+
+ sendBleData_zhihui(value, sizeof(value));
+}
+
#define STEP_DATA_LEN 8
#define KCAL_DATA_LEN 8
}
+/**
+ * @brief 处理EPO文件传输完成后的解压操作
+ * @param buffer_handle 文件接收缓冲区句柄
+ * @return ERRCODE_SUCC 成功, ERRCODE_FAIL 失败
+ *
+ * 功能说明：
+ * - 检查文件类型是否为EPO文件（file_type == 7）
+ * - 检查文件名是否以.lzma结尾
+ * - 如果是lzma文件，则调用解压函数解压
+ * - 解压成功后删除原始lzma文件
+ */
+static errcode_t handle_epo_file_decompress(ble_read_data *buffer_handle)
+{
+ // 检查是否为EPO文件
+ if (buffer_handle->file_recv_stat.file_type != 7)
+ {
+ WEARABLE_LOGE(WEARABLE_LOG_MODULE_APP, "[handle_epo_file_decompress] EPO file check failed: file_type=%d\n",
+ buffer_handle->file_recv_stat.file_type);
+ return ERRCODE_FAIL;
+ }
+
+ // 从runtime获取EPO文件完整路径（因为file_name可能已被释放）
+ char* lzma_file_path = uapi_dev_runtime_get_epo_file_path();
+ if (lzma_file_path == NULL || strlen(lzma_file_path) == 0)
+ {
+ WEARABLE_LOGE(WEARABLE_LOG_MODULE_APP, "[handle_epo_file_decompress] Failed to get EPO file path from runtime\n");
+ return ERRCODE_FAIL;
+ }
+
+ WEARABLE_LOGE(WEARABLE_LOG_MODULE_APP, "[handle_epo_file_decompress] EPO file path from runtime: %s\n", lzma_file_path);
+
+ // 检查文件扩展名是否为lzma
+ const char* ext = strrchr(lzma_file_path, '.');
+ if (ext == NULL || (strcmp(ext, ".lzma") != 0 && strcmp(ext, ".LZMA") != 0))
+ {
+ WEARABLE_LOGE(WEARABLE_LOG_MODULE_APP, "[handle_epo_file_decompress] EPO file is not a lzma file, skip extraction. ext=%s\n",
+ ext ? ext : "NULL");
+ return ERRCODE_FAIL;
+ }
+
+ // 执行解压
+ WEARABLE_LOGE(WEARABLE_LOG_MODULE_APP, "[handle_epo_file_decompress] EPO lzma file received, start extracting: %s\n", lzma_file_path);
+
+ errcode_t extract_ret = uapi_epo_lzma_decompress(lzma_file_path);
+ if (extract_ret == ERRCODE_SUCC)
+ {
+ WEARABLE_LOGE(WEARABLE_LOG_MODULE_APP, "[handle_epo_file_decompress] EPO lzma file extracted successfully\n");
+ // 解压成功后删除lzma文件
+ remove(lzma_file_path);
+ return ERRCODE_SUCC;
+ }
+ else
+ {
+ WEARABLE_LOGE(WEARABLE_LOG_MODULE_APP, "[handle_epo_file_decompress] Failed to extract EPO lzma file, ret=0x%X\n", extract_ret);
+ return ERRCODE_FAIL;
+ }
+}
+
bool creatMessageDir(void)//创建消息文件夹
{
struct stat info;
}
+ //音乐文件传输
else if (buffer_handle->file_recv_stat.file_type == 2)
{
char musicpath[64]={0};
create_mp3_name(filepath,musicpath,buffer_handle->file_recv_stat.file_name);
buffer_handle->file_recv_stat.file = fopen(filepath,"w+");
- WEARABLE_LOGD(WEARABLE_LOG_MODULE_APP,"filepath %s\r\n",filepath);
+ WEARABLE_LOGD(WEARABLE_LOG_MODULE_APP,"filepath %s\r\n",filepath);
}
- else if (buffer_handle->file_recv_stat.file_type == 4)
+ //app_file
+ else if (buffer_handle->file_recv_stat.file_type == 4)
{
char filepath[128]={0};
if (get_app_file_path(buffer_handle->file_recv_stat.file_name,strlen(buffer_handle->file_recv_stat.file_name),filepath,sizeof(filepath)) == ERRCODE_SUCC)
WEARABLE_LOGD(WEARABLE_LOG_MODULE_APP,"create file %s error %X\n",buffer_handle->file_recv_stat.file_name,buffer_handle->file_recv_stat.file);
}
}
+ //gnss文件传输
+ else if (buffer_handle->file_recv_stat.file_type == 7)
+ {
+ char filepath[128]={0};
+ if (get_epo_file_path(buffer_handle->file_recv_stat.file_name,strlen(buffer_handle->file_recv_stat.file_name),filepath,sizeof(filepath)) == ERRCODE_SUCC)
+ {
+ buffer_handle->file_recv_stat.file = fopen(filepath,"w+");
+ WEARABLE_LOGE(WEARABLE_LOG_MODULE_APP,"create file %s %X\n",filepath,buffer_handle->file_recv_stat.file);
+ }else{
+ WEARABLE_LOGD(WEARABLE_LOG_MODULE_APP,"create file %s error %X\n",buffer_handle->file_recv_stat.file_name,buffer_handle->file_recv_stat.file);
+ }
+ }
else
{
char filepath[128]={0};
if (get_file_recv_file_path(buffer_handle->file_recv_stat.file_name,strlen(buffer_handle->file_recv_stat.file_name),filepath,sizeof(filepath)) == ERRCODE_SUCC)
{
buffer_handle->file_recv_stat.file = fopen(filepath,"w+");
- WEARABLE_LOGD(WEARABLE_LOG_MODULE_APP,"create file %s %X\n",filepath,buffer_handle->file_recv_stat.file);
+ WEARABLE_LOGD(WEARABLE_LOG_MODULE_APP,"create file %s %X\n",filepath,buffer_handle->file_recv_stat.file);
}else{
- WEARABLE_LOGD(WEARABLE_LOG_MODULE_APP,"create file %s error %X\n",buffer_handle->file_recv_stat.file_name,buffer_handle->file_recv_stat.file);
+ WEARABLE_LOGD(WEARABLE_LOG_MODULE_APP,"create file %s error %X\n",buffer_handle->file_recv_stat.file_name,buffer_handle->file_recv_stat.file);
}
}
if(buffer_handle->file_recv_stat.file_type==4)
TRANSMIT_DATA=TRANS_BIN_TRANSMITTING;
+ //gnss文件传输
+ if(buffer_handle->file_recv_stat.file_type==7)
+ TRANSMIT_DATA=TRANS_EPO_TRANSMITTING;
+
CustomEventPublish(TOPIC_EVENT_BLE_TRANSMIT,&TRANSMIT_DATA,1);
}
}
return ERRCODE_FAIL;
}
- //检查文件是否写完
+ //检查文件是否写完
if(buffer_handle->file_recv_stat.file_size == buffer_handle->file_recv_stat.received_size)
{
if (buffer_handle->file_recv_stat.file!=NULL)
}
ble_read_timer_p=NULL;
+ //gnss文件传输
+ if(buffer_handle->file_recv_stat.file_type==7)
+ TRANSMIT_DATA=TRANS_EPO_TRANSMITTING;
+ else
TRANSMIT_DATA=FILE_IS_DONE;
CustomEventPublish(TOPIC_EVENT_BLE_TRANSMIT,&TRANSMIT_DATA,1);
}
else
{
- buffer_handle->file_recv_stat.received_size = 0;
+ buffer_handle->file_recv_stat.received_size = 0;
buffer_handle->file_recv_stat.lrcvalue = 0;
buffer_handle->receive_file_flag = 1;
uapi_ble_connect_param_update(BLE_CONNECT_INTERVAL_HIGH);
}
+//解析蓝牙传输过来的 EPO 文件元数据（类型、名称、大小）
+static errcode_t decode_epo_json(ble_read_data *buffer_handle)
+{
+ WEARABLE_LOGE(WEARABLE_LOG_MODULE_APP,"decode_epo_json");
+
+ uint8_t *data = buffer_handle->data+sizeof(msg_header_t);
+ uint16_t datalen = buffer_handle->totalDataLen;
+ WEARABLE_LOGE(WEARABLE_LOG_MODULE_APP,"EPO_file_save %s\n",data);
+
+ cJSON* root = cJSON_Parse((const char *)data);
+
+ if(root != NULL)
+ {
+ cJSON* epo_type_json = cJSON_GetObjectItem(root, "type");
+ if(epo_type_json != NULL && cJSON_IsNumber(epo_type_json))
+ {
+ buffer_handle->file_recv_stat.file_type = epo_type_json->valueint;
+ }
+ else
+ {
+ buffer_handle->file_recv_stat.file_type = 7; // fallback
+ }
+
+ cJSON* epo_name_json = cJSON_GetObjectItem(root, "name");
+ if ( epo_name_json != NULL && cJSON_IsString(epo_name_json)) {
+ uint32_t filenamelen=strlen(epo_name_json->valuestring)+1;
+
+ if (filenamelen > 0 && filenamelen < 255)
+ {
+ if(buffer_handle->file_recv_stat.file_name)
+ {
+ free(buffer_handle->file_recv_stat.file_name);
+ buffer_handle->file_recv_stat.file_name=NULL;
+ }
+ buffer_handle->file_recv_stat.file_name = malloc(filenamelen);
+ if (buffer_handle->file_recv_stat.file_name!=NULL)
+ {
+ memset_s(buffer_handle->file_recv_stat.file_name,filenamelen,0,filenamelen);
+ memcpy_s(buffer_handle->file_recv_stat.file_name,filenamelen,epo_name_json->valuestring,filenamelen);
+ }
+ }else{
+ buffer_handle->file_recv_stat.file_name = NULL;
+ }
+ }
+
+ cJSON* epo_size_json = cJSON_GetObjectItem(root, "size");
+ if(epo_size_json != NULL && cJSON_IsNumber(epo_size_json))
+ {
+ buffer_handle->file_recv_stat.file_size = epo_size_json->valueint;
+ }
+ cJSON_Delete(root);
+
+ // 保存EPO文件路径到runtime，以便后续解压时使用
+ if (buffer_handle->file_recv_stat.file_name != NULL)
+ {
+ char epo_file_path[256] = {0};
+ if (get_epo_file_path(buffer_handle->file_recv_stat.file_name,
+ strlen(buffer_handle->file_recv_stat.file_name),
+ epo_file_path, sizeof(epo_file_path)) == ERRCODE_SUCC)
+ {
+ uapi_dev_runtime_set_epo_file_path(epo_file_path);
+ WEARABLE_LOGE(WEARABLE_LOG_MODULE_APP, "[decode_epo_json] Saved EPO file path to runtime: %s\n", epo_file_path);
+ }
+ }
+
+ WEARABLE_LOGE(WEARABLE_LOG_MODULE_APP," EPO_type= %d, EPO_name=%s, EPO_size=%d \n",
+ buffer_handle->file_recv_stat.file_type,
+ buffer_handle->file_recv_stat.file_name,
+ buffer_handle->file_recv_stat.file_size);
+
+ buffer_handle->file_recv_stat.received_size = 0;
+ buffer_handle->file_recv_stat.lrcvalue = 0;
+ buffer_handle->receive_file_flag = 1;
+ uapi_ble_connect_param_update(BLE_CONNECT_INTERVAL_HIGH);
+
+ return ERRCODE_SUCC;
+
+ }else{
+ buffer_handle->receive_file_flag = 0;
+ return ERRCODE_FAIL;
+ }
+}
+
+
+
+
errcode_t decode_bin_json(ble_read_data *buffer_handle)
{
if (uapi_dev_runtime_get_train_status() == 1) {
}
}
+// 蓝牙文件传输完成后的校验与结果处理核心函数（指令码2AE6对应文件校验场景）
void decode_file_checksum_2AE6(ble_read_data *buffer_handle)
{
uapi_ble_connect_param_update(BLE_CONNECT_INTERVAL_SLOW);
// ble_mp3_list_update();
+ break;
+ case 7://epo
+ TRANSMIT_DATA=TRANS_EPO_SUCC;
+ CustomEventPublish(TOPIC_EVENT_BLE_TRANSMIT,&TRANSMIT_DATA,1);
+ handle_epo_file_decompress(buffer_handle);
+ ble_reply_cmd(FILE_SUCC_TRANSFER,1);
break;
default:
break;
TRANSMIT_DATA= TRANS_BIN_FAIL;
CustomEventPublish(TOPIC_EVENT_BLE_TRANSMIT,&TRANSMIT_DATA,1);
+ break;
+ case 7://epo
+ TRANSMIT_DATA= TRANS_EPO_FAIL;
+ CustomEventPublish(TOPIC_EVENT_BLE_TRANSMIT,&TRANSMIT_DATA,1);
break;
default:
break;
TRANSMIT_DATA= TRANS_BIN_FAIL;
CustomEventPublish(TOPIC_EVENT_BLE_TRANSMIT,&TRANSMIT_DATA,1);
break;
+ case 7://epo
+ TRANSMIT_DATA= TRANS_EPO_FAIL;
+ CustomEventPublish(TOPIC_EVENT_BLE_TRANSMIT,&TRANSMIT_DATA,1);
+ break;
default:
break;
}
}
break;
+ case 7://epo
+ TRANSMIT_DATA=TRANS_EPO_SUCC;
+ handle_epo_file_decompress(buffer_handle);
+ CustomEventPublish(TOPIC_EVENT_BLE_TRANSMIT,&TRANSMIT_DATA,1);
+ ble_reply_cmd(FILE_SUCC_TRANSFER,1);
+ break;
default:
break;
}
TRANSMIT_DATA= TRANS_DIAL_FAIL;
CustomEventPublish(TOPIC_EVENT_BLE_TRANSMIT,&TRANSMIT_DATA,1);
+ break;
+ case 7://epo
+ TRANSMIT_DATA= TRANS_EPO_FAIL;
+ CustomEventPublish(TOPIC_EVENT_BLE_TRANSMIT,&TRANSMIT_DATA,1);
break;
default:
break;
TRANSMIT_DATA= TRANS_DIAL_FAIL;
CustomEventPublish(TOPIC_EVENT_BLE_TRANSMIT,&TRANSMIT_DATA,1);
break;
+ case 7://epo
+ TRANSMIT_DATA= TRANS_EPO_FAIL;
+ CustomEventPublish(TOPIC_EVENT_BLE_TRANSMIT,&TRANSMIT_DATA,1);
+ break;
default:
break;
}
case GET_APP_MSG_CMD:
get_watch_msg( cmd, data,datalen,crcvalue);
- break;
+ break;
+ case GNSS_DATA_RECV_CMD :
+ {
+ if (decode_epo_json(buffer_handle) == ERRCODE_SUCC)
+ {
+ WEARABLE_LOGE(WEARABLE_LOG_MODULE_APP,"解析蓝牙传输过来的 EPO 文件元数据 ，回复成功");
+ ble_reply_cmd(GNSS_DATA_RECV_CMD,1);
+ }else{
+ WEARABLE_LOGE(WEARABLE_LOG_MODULE_APP,"解析蓝牙传输过来的 EPO 文件元数据 ，回复失败");
+ ble_reply_cmd(GNSS_DATA_RECV_CMD,2);
+ }
+ }
+ break;
default:
break;
}
{
TRANSMIT_DATA=TRANS_BIN_FAIL;
}
+ else if(ble_read_timer_p->file_recv_stat.file_type==7)
+ {
+ //gnss文件传输失败
+ TRANSMIT_DATA=TRANS_EPO_FAIL;
+ }
else
{
WEARABLE_LOGE(WEARABLE_LOG_MODULE_APP,"filepath %s\r\n",filepath);
}
}
+ //gnss文件传输
+ else if (ble_read_timer_p->file_recv_stat.file_type == 7)
+ {
+ char filepath[128]={0};
+ if (get_epo_file_path(ble_read_timer_p->file_recv_stat.file_name,strlen(ble_read_timer_p->file_recv_stat.file_name),filepath,sizeof(filepath)) == ERRCODE_SUCC)
+ {
+ remove(filepath);
+ WEARABLE_LOGE(WEARABLE_LOG_MODULE_APP,"filepath %s\r\n",filepath);
+ }
+ }
if (ble_read_timer_p->file_recv_stat.file_name != NULL)
{
// return ERRCODE_FAIL;
}
}
- WEARABLE_LOGD(WEARABLE_LOG_MODULE_APP,"handle_recv_file %d\n",event_size);
+ WEARABLE_LOGE(WEARABLE_LOG_MODULE_APP,"handle_recv_file %d\n",event_size);
save_package_sample(buffer_handle,data,event_size);
buffer_handle->file_recv_stat.file = fopen(filepath,"w+");
WEARABLE_LOGD(WEARABLE_LOG_MODULE_APP,"filepath %s\r\n",filepath);
}
+ //gnss文件传输
+ else if (buffer_handle->file_recv_stat.file_type == 7)
+ {
+ char filepath[128]={0};
+ if (get_epo_file_path(buffer_handle->file_recv_stat.file_name,strlen(buffer_handle->file_recv_stat.file_name),filepath,sizeof(filepath)) == ERRCODE_SUCC)
+ {
+ buffer_handle->file_recv_stat.file = fopen(filepath,"w+");
+ WEARABLE_LOGE(WEARABLE_LOG_MODULE_APP,"filepath %s\r\n",filepath);
+ }else{
+ WEARABLE_LOGE(WEARABLE_LOG_MODULE_APP,"create file %s error %X\n",buffer_handle->file_recv_stat.file_name,buffer_handle->file_recv_stat.file);
+ }
+ }
else
{
char filepath[128]={0};
if(buffer_handle->file_recv_stat.file_type==2)
TRANSMIT_DATA=TRANS_MP3_TRANSMITTING;
+ //gnss文件传输
+ if(buffer_handle->file_recv_stat.file_type==7)
+ TRANSMIT_DATA=TRANS_EPO_TRANSMITTING;
+
CustomEventPublish(TOPIC_EVENT_BLE_TRANSMIT,&TRANSMIT_DATA,1);
}
typedef struct
{
- uint8_t file_type; //1 ota file , 2 mp3 file , 3 dial file, 7 epo file
+ uint8_t file_type; //1 ota file , 2 mp3 file , 3 dial file, 4 bin/app , 7 epo/gnss
uint32_t file_size;
uint8_t ota_major;
uint8_t ota_mid;
void sendtraningstateend(void);
+void ble_request_epo_file(void);
+
int char_to_dec(char ch);//字符数字转换成阿拉伯数字
void char_to_big(uint8_t *src_str,uint8_t *res_str,uint8_t len);
#define APP_DELETE_CMD 0x060D
#define GET_LIST_CMD 0x060E
#define GET_APP_MSG_CMD 0x060F
+#define GNSS_DATA_RECV_CMD 0x0610
#ifdef __cplusplus
}
uint8_t vibmod;
bt_runtime_status_t bt_runtime;
char current_dial_name[32];
+ char current_epo_file_path[128]; // EPO文件完整路径
uint8_t time_hour_show_model;
uint8_t sleep_monitoring_status;//睡眠监测
uint8_t sleep_status;//睡眠状态
{
return (char*)&(g_runtime.current_dial_name[0]);
}
+
+/**
+ * 设置 EPO 文件路径
+*/
+void uapi_dev_runtime_set_epo_file_path(char* file_path)
+{
+ if (file_path == NULL) return;
+ // 在runtime中备份EPO文件路径
+ int pathlen = strlen((char*)file_path);
+ if (pathlen >= sizeof(g_runtime.current_epo_file_path)) {
+ pathlen = sizeof(g_runtime.current_epo_file_path) - 1;
+ }
+ memset_s(g_runtime.current_epo_file_path, sizeof(g_runtime.current_epo_file_path), 0, sizeof(g_runtime.current_epo_file_path));
+ memcpy_s(g_runtime.current_epo_file_path, sizeof(g_runtime.current_epo_file_path), file_path, pathlen);
+}
+
+/**
+ * 获取 EPO 文件路径
+*/
+char* uapi_dev_runtime_get_epo_file_path(void)
+{
+ return (char*)&(g_runtime.current_epo_file_path[0]);
+}
void uapi_dev_runtime_set_bt_powerstat(uint8_t stat)
{
g_runtime.bt_runtime.bt_powerOn = stat;
*/
char* uapi_dev_runtime_get_dialname(void);
+/**
+ * 设置 EPO 文件路径
+*/
+void uapi_dev_runtime_set_epo_file_path(char* file_path);
+/**
+ * 获取 EPO 文件路径
+*/
+char* uapi_dev_runtime_get_epo_file_path(void);
+
void uapi_dev_runtime_set_vibmod(uint8_t vibmod);
//获取固件版本号
errcode_t uapi_get_fw_version(uint8_t *_major, uint8_t *_mid,uint8_t *_minor);
-/**
- * @}
- */
-
- //判断文件夹是否存在
- bool uapi_create_dir(const char *path);
+//判断文件夹是否存在,如果不存在则创建
+bool uapi_create_dir(const char *path);
//flash整片擦除
return ERRCODE_SUCC;
}
+
+// EPO LZMA decompress
+static uint32_t custom_utils_epo_lzmadec(uint8_t *para, uint32_t paraLen, char *strCommonCmd)
+{
+ UNUSED(strCommonCmd);
+ uint8_t argv[CUSTOM_UTILS_MAX_ARGC][CUSTOM_UTILS_MAX_ARGV_LEN] = { 0 };
+ uint32_t argc = 0;
+
+ int ret = ConvertCmdToParam(para, paraLen, argv, &argc);
+ if (ret != CUSTOM_UTILS_SUCCESS) {
+ wstp_print("CUSTOM_UTILSTEST ATCMD Convert Command Failed\n");
+ return ERRCODE_FAIL;
+ }
+
+ if (argc != 1) {
+ wstp_print("CUSTOM_UTILSTEST ATCMD Set Parameter Failed argc %d, need 1 (lzma_file_path)\n",argc);
+ return ERRCODE_FAIL;
+ }
+ const char *infile = (const char*)argv[0];
+ wstp_print("EPO LZMA decompress: %s\r\n", infile);
+ errcode_t rret = uapi_epo_lzma_decompress(infile);
+ if (rret == ERRCODE_SUCC) {
+ wstp_print("EPO LZMA decompress SUCCESS\r\n");
+ } else {
+ wstp_print("EPO LZMA decompress FAILED, ret 0x%X\r\n", rret);
+ }
+
+ return ERRCODE_SUCC;
+}
+
static uint32_t custom_utils_pdudec(uint8_t *para, uint32_t paraLen, char *strCommonCmd)
{
UNUSED(strCommonCmd);
{ (char *)"AT^GETSLEEP", custom_get_sleep, (char *)" " },
{ (char *)"AT^LZMADEC", custom_utils_lzmadec, (char *)" " },
+ { (char *)"AT^EPODEC", custom_utils_epo_lzmadec, (char *)" " },
{ (char *)"AT^PDUDEC", custom_utils_pdudec, (char *)" " },
{ (char *)"AT^SNTP", custom_test_sntp, (char *)" " },
{ (char *)"AT^TIMESYNC", custom_test_time_synct, (char *)" " },
return 0;
}
+/**
+ * @brief 解压EPO LZMA文件到多个文件
+ * @param lzma_file_path LZMA压缩文件路径
+ * @return ERRCODE_SUCC 成功, ERRCODE_FAIL 失败
+ *
+ * 文件格式说明：
+ * LZMA文件解压后的数据格式：
+ * - 文件头：文件数量（4字节，小端）
+ * - 对于每个文件：
+ * - 文件名长度（2字节，小端）
+ * - 文件名（变长，不含路径）
+ * - 文件大小（4字节，小端）
+ * - 文件数据（变长）
+ */
+/* 从文件名中提取"_"与"."之间的时间戳子串，失败返回空字符串 */
+static void extract_timestamp_from_name(const char *filename, char *ts_buf, size_t buf_len)
+{
+ if (ts_buf == NULL || buf_len == 0 || filename == NULL) {
+ return;
+ }
+ ts_buf[0] = '\0';
+ const char *us = strrchr(filename, '_');
+ const char *dot = strrchr(filename, '.');
+ if (us == NULL || dot == NULL || dot <= us + 1) {
+ return;
+ }
+ size_t len = (size_t)(dot - us - 1);
+ if (len >= buf_len) {
+ len = buf_len - 1;
+ }
+ (void)memcpy(ts_buf, us + 1, len);
+ ts_buf[len] = '\0';
+}
+
+errcode_t uapi_epo_lzma_decompress(const char* lzma_file_path)
+{
+ printf("[uapi_epo_lzma_decompress] lzma_file_path: %s\n", lzma_file_path);
+ if (lzma_file_path == NULL) {
+ printf("[uapi_epo_lzma_decompress] lzma_file_path is NULL\n");
+ return ERRCODE_FAIL;
+ }
+
+ // 确保输出目录存在
+ uapi_create_dir("/user/userdata/gnss");
+
+ // 先解压LZMA文件到内存缓冲区
+ CFileSeqInStream inStream;
+ CFileOutStream outStream;
+ FileSeqInStream_CreateVTable(&inStream);
+ File_Construct(&inStream.file);
+ FileOutStream_CreateVTable(&outStream);
+ File_Construct(&outStream.file);
+
+ if (InFile_Open(&inStream.file, lzma_file_path) != 0) {
+ printf("[uapi_epo_lzma_decompress] Can not open input file: %s\n", lzma_file_path);
+ return ERRCODE_FAIL;
+ }
+
+ // 创建临时文件来存储解压后的数据
+ const char* temp_file = "/user/userdata/gnss/epo_temp.dat";
+ if (OutFile_Open(&outStream.file, temp_file) != 0) {
+ printf("[uapi_epo_lzma_decompress] Can not create temp file\n");
+ File_Close(&inStream.file);
+ return ERRCODE_FAIL;
+ }
+
+ // 解压LZMA文件
+ SRes res = Decode(&outStream.vt, &inStream.vt);
+ File_Close(&outStream.file);
+ File_Close(&inStream.file);
+
+ if (res != SZ_OK) {
+ printf("[uapi_epo_lzma_decompress] LZMA decompress failed: %d\n", res);
+ remove(temp_file);
+ return ERRCODE_FAIL;
+ }
+
+ // 读取解压后的数据
+ FILE* fp = fopen(temp_file, "rb");
+ if (fp == NULL) {
+ printf("[uapi_epo_lzma_decompress] Can not open temp file\n");
+ remove(temp_file);
+ return ERRCODE_FAIL;
+ }
+
+ // 读取文件数量
+ uint32_t file_count = 0;
+ if (fread(&file_count, 1, 4, fp) != 4) {
+ printf("[uapi_epo_lzma_decompress] Failed to read file count\n");
+ fclose(fp);
+ remove(temp_file);
+ return ERRCODE_FAIL;
+ }
+
+ printf("[uapi_epo_lzma_decompress] File count: %u\n", file_count);
+
+ if (file_count == 0) {
+ printf("[uapi_epo_lzma_decompress] Invalid file count: %u\n", file_count);
+ fclose(fp);
+ remove(temp_file);
+ return ERRCODE_FAIL;
+ }
+
+ /* 第一遍扫描，尝试找到AssistInfo_*文件的时间戳，用作目录名 */
+ char assist_ts[32] = {0};
+ long data_start_pos = ftell(fp);
+ for (uint32_t i = 0; i < file_count; i++) {
+ uint16_t filename_len = 0;
+ if (fread(&filename_len, 1, 2, fp) != 2) { break; }
+ if (filename_len == 0 || filename_len > 255) { break; }
+ char filename[256] = {0};
+ if (fread(filename, 1, filename_len, fp) != filename_len) { break; }
+ uint32_t file_size = 0;
+ if (fread(&file_size, 1, 4, fp) != 4) { break; }
+ if (file_size > 0) { (void)fseek(fp, file_size, SEEK_CUR); }
+ if (assist_ts[0] == '\0' && strncmp(filename, "AssistInfo_", strlen("AssistInfo_")) == 0) {
+ extract_timestamp_from_name(filename, assist_ts, sizeof(assist_ts));
+ }
+ }
+ /* 回到文件列表起始位置准备正式解压 */
+ (void)fseek(fp, data_start_pos, SEEK_SET);
+
+ int extracted_count = 0;
+
+ // 遍历每个文件
+ for (uint32_t i = 0; i < file_count; i++) {
+ // 读取文件名长度
+ uint16_t filename_len = 0;
+ if (fread(&filename_len, 1, 2, fp) != 2) {
+ printf("[uapi_epo_lzma_decompress] Failed to read filename length for file %u\n", i);
+ break;
+ }
+
+ if (filename_len == 0 || filename_len > 255) {
+ printf("[uapi_epo_lzma_decompress] Invalid filename length: %u for file %u\n", filename_len, i);
+ break;
+ }
+
+ // 读取文件名
+ char filename[256] = {0};
+ if (fread(filename, 1, filename_len, fp) != filename_len) {
+ printf("[uapi_epo_lzma_decompress] Failed to read filename for file %u\n", i);
+ break;
+ }
+
+ // 读取文件大小
+ uint32_t file_size = 0;
+ if (fread(&file_size, 1, 4, fp) != 4) {
+ printf("[uapi_epo_lzma_decompress] Failed to read file size for file %u\n", i);
+ break;
+ }
+
+ if (file_size == 0 || file_size > 10 * 1024 * 1024) { // 最大10MB
+ printf("[uapi_epo_lzma_decompress] Invalid file size: %u for file %u\n", file_size, i);
+ break;
+ }
+
+ /* 确定目录时间戳：优先用AssistInfo中的时间戳，否则取自身文件名中的时间戳 */
+ char dir_ts[32] = {0};
+ if (assist_ts[0] != '\0') {
+ (void)strncpy(dir_ts, assist_ts, sizeof(dir_ts) - 1);
+ } else {
+ extract_timestamp_from_name(filename, dir_ts, sizeof(dir_ts));
+ }
+ if (dir_ts[0] == '\0') {
+ printf("[uapi_epo_lzma_decompress] Failed to parse timestamp for file %s\n", filename);
+ break;
+ }
+
+ /* 生成输出目录并确保存在 */
+ char output_dir[128] = {0};
+ int dir_ret = snprintf(output_dir, sizeof(output_dir), "/user/userdata/gnss/%s", dir_ts);
+ if (dir_ret < 0 || dir_ret >= (int)sizeof(output_dir)) {
+ printf("[uapi_epo_lzma_decompress] Failed to build output dir for file %u\n", i);
+ break;
+ }
+ uapi_create_dir(output_dir);
+
+ /* AssistInfo 重命名为 AssistInfo.dat */
+ const char *final_name = filename;
+ char renamed[64] = {0};
+ if (strncmp(filename, "AssistInfo_", strlen("AssistInfo_")) == 0) {
+ (void)strncpy(renamed, "AssistInfo.dat", sizeof(renamed) - 1);
+ final_name = renamed;
+ }
+
+ // 构建输出文件路径
+ char output_path[256] = {0};
+ int path_ret = snprintf(output_path, sizeof(output_path), "%s/%s", output_dir, final_name);
+ if (path_ret < 0 || path_ret >= (int)sizeof(output_path)) {
+ printf("[uapi_epo_lzma_decompress] Failed to build output path for file %u\n", i);
+ break;
+ }
+
+ // 创建输出文件
+ FILE* outfile = fopen(output_path, "wb");
+ if (outfile == NULL) {
+ printf("[uapi_epo_lzma_decompress] Failed to create output file: %s\n", output_path);
+ break;
+ }
+
+ // 读取并写入文件数据
+ uint8_t* buffer = (uint8_t*)malloc(file_size);
+ if (buffer == NULL) {
+ printf("[uapi_epo_lzma_decompress] Failed to allocate buffer for file %u\n", i);
+ fclose(outfile);
+ break;
+ }
+
+ size_t read_size = fread(buffer, 1, file_size, fp);
+ if (read_size != file_size) {
+ printf("[uapi_epo_lzma_decompress] Failed to read file data for file %u: read %zu, expected %u\n",
+ i, read_size, file_size);
+ free(buffer);
+ fclose(outfile);
+ remove(output_path);
+ break;
+ }
+
+ size_t written = fwrite(buffer, 1, file_size, outfile);
+ free(buffer);
+ fclose(outfile);
+
+ if (written != file_size) {
+ printf("[uapi_epo_lzma_decompress] Failed to write file: %s\n", output_path);
+ remove(output_path);
+ break;
+ }
+
+ printf("[uapi_epo_lzma_decompress] Extracted: %s (%u bytes)\n", output_path, file_size);
+ extracted_count++;
+ }
+
+ fclose(fp);
+ remove(temp_file);
+
+ if (extracted_count == 0) {
+ printf("[uapi_epo_lzma_decompress] No files extracted\n");
+ return ERRCODE_FAIL;
+ }
+
+ if (extracted_count != file_count) {
+ printf("[uapi_epo_lzma_decompress] Warning: extracted %d/%u files\n", extracted_count, file_count);
+ }
+
+ printf("[uapi_epo_lzma_decompress] Successfully extracted %d files\n", extracted_count);
+ return ERRCODE_SUCC;
+}
+
void uapi_dump_meminfo(void)
{
#include <errcode.h>
errcode_t uapi_dial_pkg_decompress(const char* dial_pkg_path,char* dial_bin_path);
+errcode_t uapi_epo_lzma_decompress(const char* lzma_file_path);
void uapi_dump_meminfo(void);
void CustomEventPublish(uint16_t topic, uint8_t *data , uint32_t size);
--- /dev/null
+# EPO LZMA 解压功能测试文档
+
+## 功能说明
+
+`uapi_epo_lzma_decompress()` 函数用于解压包含多个星历文件的 LZMA 压缩包，并将文件提取到 `/user/userdata/gnss/` 目录。
+
+## 文件格式
+
+### LZMA 文件格式
+```
+[LZMA 头部: 13字节]
+ - LZMA 属性: 5字节
+ - 未压缩大小: 8字节（小端）
+
+[压缩数据]
+```
+
+### 解压后的打包格式
+```
+[文件数量: 4字节，小端]
+
+[文件1]
+ - 文件名长度: 2字节，小端
+ - 文件名: 变长（UTF-8编码）
+ - 文件大小: 4字节，小端
+ - 文件数据: 变长
+
+[文件2]
+ - ...
+
+[文件N]
+ - ...
+```
+
+## 测试流程
+
+### 1. 创建测试文件
+
+创建三个测试文件：
+```bash
+echo "EPH file content for testing" > 1.eph
+echo "TXT file content for testing" > 2.txt
+echo "DAT file content for testing" > 3.dat
+```
+
+### 2. 打包文件（Python脚本）
+
+使用提供的 Python 脚本打包：
+```bash
+python3 create_epo_lzma.py 1.eph 2.txt 3.dat epo.lzma
+```
+
+或者使用默认测试数据：
+```bash
+python3 create_epo_lzma.py
+```
+
+### 3. 传输到手表
+
+通过 BLE 传输 `epo.lzma` 文件到手表：
+- 使用 EPO 文件传输协议（`EPO_DATA_RECV_CMD`）
+- 文件名必须包含 `.lzma` 扩展名
+- 文件会保存到 `/user/userdata/` 目录
+
+### 4. AT 命令验证
+
+手表接收完成后，使用 AT 命令验证解压：
+
+```
+AT^EPODEC=/user/userdata/epo.lzma
+```
+
+**预期输出：**
+```
+EPO LZMA decompress: /user/userdata/epo.lzma
+[uapi_epo_lzma_decompress] lzma_file_path: /user/userdata/epo.lzma
+[uapi_epo_lzma_decompress] File count: 3
+[uapi_epo_lzma_decompress] Extracted: /user/userdata/gnss/1.eph (XX bytes)
+[uapi_epo_lzma_decompress] Extracted: /user/userdata/gnss/2.txt (XX bytes)
+[uapi_epo_lzma_decompress] Extracted: /user/userdata/gnss/3.dat (XX bytes)
+[uapi_epo_lzma_decompress] Successfully extracted 3 files
+EPO LZMA decompress SUCCESS
+```
+
+### 5. 验证解压结果
+
+使用文件系统命令验证文件是否存在：
+```bash
+# 查看目录
+ls /user/userdata/gnss/
+
+# 应该看到：
+# 1.eph
+# 2.txt
+# 3.dat
+```
+
+## 使用 7-Zip 创建 LZMA 文件（备选方案）
+
+如果 Python 脚本不工作，可以使用 7-Zip：
+
+### 步骤1: 创建打包文件
+
+使用 Python 脚本创建打包数据（不压缩）：
+```python
+# create_packed_data.py
+import struct
+
+files = [
+ ("1.eph", b"EPH file content"),
+ ("2.txt", b"TXT file content"),
+ ("3.dat", b"DAT file content"),
+]
+
+packed = struct.pack('<I', len(files))
+for name, data in files:
+ name_bytes = name.encode('utf-8')
+ packed += struct.pack('<H', len(name_bytes))
+ packed += name_bytes
+ packed += struct.pack('<I', len(data))
+ packed += data
+
+with open('packed.bin', 'wb') as f:
+ f.write(packed)
+```
+
+### 步骤2: 使用 7z 压缩
+
+```bash
+7z a -m0=lzma -mx=6 packed.lzma packed.bin
+```
+
+### 步骤3: 添加 LZMA 头部
+
+需要手动添加 13 字节头部（5字节属性 + 8字节大小）
+
+## 自动解压流程
+
+当 APP 通过 BLE 传输 EPO 文件时：
+
+1. **APP 发送元信息**：
+ ```
+ CMD: EPO_DATA_RECV_CMD (0x04 0x25)
+ JSON: {"type":7, "name":"epo.lzma", "size":12345}
+ ```
+
+2. **手表接收文件数据**：
+ - 文件保存到 `/user/userdata/epo.lzma`
+
+3. **文件传输完成**：
+ - 检测到 `.lzma` 扩展名
+ - 自动调用 `uapi_epo_lzma_decompress()`
+ - 解压文件到 `/user/userdata/gnss/`
+ - 删除原始 LZMA 文件
+
+4. **解压结果**：
+ - 成功：文件提取到 `/user/userdata/gnss/` 目录
+ - 失败：记录错误日志，保留原始文件
+
+## 注意事项
+
+1. **文件格式必须正确**：打包格式必须严格按照上述格式
+2. **文件名限制**：文件名长度不超过 255 字节
+3. **文件大小限制**：单个文件不超过 10MB
+4. **文件数量限制**：最多 100 个文件
+5. **目录自动创建**：`/user/userdata/gnss/` 目录会自动创建
+
+## 故障排查
+
+### 解压失败
+- 检查 LZMA 文件格式是否正确
+- 检查打包数据格式是否符合规范
+- 查看日志输出定位问题
+
+### 文件未提取
+- 检查 `/user/userdata/gnss/` 目录权限
+- 检查磁盘空间是否充足
+- 查看解压日志确认文件数量
+
+### AT 命令无响应
+- 确认文件路径正确
+- 确认文件存在
+- 检查 AT 命令注册是否正确
+
--- /dev/null
+# EPO LZMA 解压功能验证流程
+
+## 功能概述
+
+`uapi_epo_lzma_decompress()` 函数可以将包含多个星历文件的 LZMA 压缩包解压，提取多个文件到 `/user/userdata/gnss/` 目录。
+
+## 完整验证流程
+
+### 步骤 1: 创建测试 LZMA 文件
+
+**方法 A: 使用 Python 脚本（推荐）**
+
+```bash
+cd /home/cheng/leiniao/FD230/tools/scripts
+python3 create_epo_lzma.py
+```
+
+这会创建默认的测试文件 `epo.lzma`，包含：
+- `1.eph` - EPH 格式文件
+- `2.txt` - TXT 格式文件
+- `3.dat` - DAT 格式文件
+
+**方法 B: 从真实文件创建**
+
+```bash
+python3 create_epo_lzma.py 1.eph 2.txt 3.dat epo.lzma
+```
+
+### 步骤 2: 传输到手表
+
+通过 BLE 协议传输 `epo.lzma` 文件：
+
+1. **APP 发送元信息**：
+ ```
+ CMD: 0x04 0x25 (EPO_DATA_RECV_CMD)
+ JSON: {"type":7, "name":"epo.lzma", "size":<文件大小>}
+ ```
+
+2. **APP 发送文件数据**：
+ - 分包发送 LZMA 文件内容
+ - 文件保存到 `/user/userdata/epo.lzma`
+
+3. **手表自动解压**：
+ - 检测到 `.lzma` 扩展名
+ - 自动调用 `uapi_epo_lzma_decompress()`
+ - 解压到 `/user/userdata/gnss/` 目录
+ - 删除原始 LZMA 文件
+
+### 步骤 3: AT 命令验证
+
+如果文件已传输到手表，可以直接使用 AT 命令测试：
+
+```
+AT^EPODEC=/user/userdata/epo.lzma
+```
+
+**预期输出：**
+```
+EPO LZMA decompress: /user/userdata/epo.lzma
+[uapi_epo_lzma_decompress] lzma_file_path: /user/userdata/epo.lzma
+[uapi_epo_lzma_decompress] File count: 3
+[uapi_epo_lzma_decompress] Extracted: /user/userdata/gnss/1.eph (XX bytes)
+[uapi_epo_lzma_decompress] Extracted: /user/userdata/gnss/2.txt (XX bytes)
+[uapi_epo_lzma_decompress] Extracted: /user/userdata/gnss/3.dat (XX bytes)
+[uapi_epo_lzma_decompress] Successfully extracted 3 files
+EPO LZMA decompress SUCCESS
+```
+
+### 步骤 4: 验证解压结果
+
+使用文件系统命令验证：
+
+```bash
+# 查看目录内容
+ls /user/userdata/gnss/
+
+# 应该看到：
+# 1.eph
+# 2.txt
+# 3.dat
+
+# 查看文件内容（如果支持）
+cat /user/userdata/gnss/1.eph
+cat /user/userdata/gnss/2.txt
+```
+
+## 文件格式说明
+
+### LZMA 文件结构
+```
+[13字节头部]
+ - 5字节: LZMA 属性（lc, lp, pb, dict_size）
+ - 8字节: 未压缩大小（小端）
+
+[LZMA 压缩数据]
+```
+
+### 13字节头部详解
+
+**完整格式：**
+```
+字节0: 压缩参数编码 = (lc * 9 + lp) * 5 + pb
+字节1-4: 字典大小（4字节，小端，0表示默认16MB）
+字节5-12: 未压缩大小（8字节，小端）
+```
+
+**示例（未压缩大小=120字节）：**
+```
+5D 00 00 00 00 78 00 00 00 00 00 00 00
+│ │ │ │ │
+│ │ │ └─ 未压缩大小=120 (小端)
+│ │ └─ 字典大小=0(默认)
+│ └─ 压缩参数 (lc=3, lp=0, pb=2)
+└─ 属性字节
+```
+
+**详细说明请参考：** `LZMA_头部格式说明.md`
+
+### 解压后的打包格式
+```
+[4字节] 文件数量（小端）
+
+[文件1]
+ [2字节] 文件名长度（小端）
+ [N字节] 文件名（UTF-8）
+ [4字节] 文件大小（小端）
+ [M字节] 文件数据
+
+[文件2]
+ ...
+
+[文件N]
+ ...
+```
+
+## 示例：创建包含 3 个文件的 LZMA
+
+假设有 3 个文件：
+- `1.eph` (30 字节)
+- `2.txt` (40 字节)
+- `3.dat` (50 字节)
+
+打包后的数据格式：
+```
+[04 00 00 00] # 文件数量 = 3
+[05 00] # 文件名长度 = 5
+[31 2E 65 70 68] # "1.eph"
+[1E 00 00 00] # 文件大小 = 30
+[...30字节数据...] # 文件内容
+[05 00] # 文件名长度 = 5
+[32 2E 74 78 74] # "2.txt"
+[28 00 00 00] # 文件大小 = 40
+[...40字节数据...] # 文件内容
+[05 00] # 文件名长度 = 5
+[33 2E 64 61 74] # "3.dat"
+[32 00 00 00] # 文件大小 = 50
+[...50字节数据...] # 文件内容
+```
+
+然后用 LZMA 压缩整个打包数据，并添加 13 字节头部。
+
+## 故障排查
+
+### 问题：解压失败
+- **检查点1**: LZMA 文件格式是否正确（13字节头部）
+- **检查点2**: 打包数据格式是否符合规范
+- **检查点3**: 查看日志输出，定位具体错误
+
+### 问题：文件未提取
+- **检查点1**: `/user/userdata/gnss/` 目录权限
+- **检查点2**: 磁盘空间是否充足
+- **检查点3**: 文件数量是否超过限制（100个）
+
+### 问题：AT 命令无响应
+- **检查点1**: 文件路径是否正确
+- **检查点2**: 文件是否存在
+- **检查点3**: AT 命令是否正确注册
+
+## 测试检查清单
+
+- [ ] 创建测试 LZMA 文件成功
+- [ ] 文件格式符合规范
+- [ ] 传输到手表成功
+- [ ] 自动解压触发
+- [ ] 解压日志正常
+- [ ] 文件提取到正确目录
+- [ ] 文件内容正确
+- [ ] 原始 LZMA 文件已删除
+- [ ] AT 命令测试通过
+
+## 相关文件
+
+- 实现代码: `code/sdk/zhihui/utilscpp/custom_utilscpp.cpp`
+- 函数声明: `code/sdk/zhihui/utilscpp/custom_utilscpp.h`
+- AT 命令: `code/sdk/zhihui/utils/custom_utils_at_cmd.c`
+- 创建脚本: `tools/scripts/create_epo_lzma.py`
+- 详细文档: `tools/scripts/EPO_LZMA_TEST.md`
+
--- /dev/null
+# LZMA 文件 13 字节头部格式说明
+
+## 概述
+
+LZMA 文件格式要求在压缩数据前添加 **13 字节的头部**，用于告诉解压器如何解压数据。
+
+## 头部结构
+
+```
+[5字节 LZMA 属性] + [8字节 未压缩大小] = 13字节
+```
+
+## 详细格式
+
+### 第 1 部分：LZMA 属性（5字节）
+
+#### 字节 0：压缩参数编码
+```
+字节0 = (lc * 9 + lp) * 5 + pb
+```
+
+**参数说明：**
+- **lc** (Literal Context bits): 字面上下文位数，范围 0-8，常用值 3
+- **lp** (Literal Position bits): 字面位置位数，范围 0-4，常用值 0
+- **pb** (Position bits): 位置位数，范围 0-4，常用值 2
+
+**解码公式：**
+```c
+lc = data[0] % 9;
+d = data[0] / 9;
+pb = d / 5;
+lp = d % 5;
+```
+
+**常见值示例：**
+- `lc=3, lp=0, pb=2` → 字节0 = (3*9+0)*5+2 = **137 (0x89)**
+- `lc=3, lp=0, pb=2` → 也常用 **93 (0x5D)**（不同编码方式）
+
+#### 字节 1-4：字典大小（4字节，小端）
+```
+字节1: 字典大小低字节
+字节2: 字典大小次低字节
+字节3: 字典大小次高字节
+字节4: 字典大小高字节
+```
+
+**字典大小计算：**
+```c
+dicSize = data[1] | (data[2] << 8) | (data[3] << 16) | (data[4] << 24);
+```
+
+**常用值：**
+- `0x00 0x00 0x00 0x00` = 0（表示使用默认值，通常为 16MB = 0x01000000）
+- `0x00 0x00 0x00 0x01` = 16MB = 16777216 字节
+- `0x00 0x00 0x00 0x02` = 32MB
+
+### 第 2 部分：未压缩大小（8字节，小端）
+
+```
+字节5: 未压缩大小最低字节
+字节6: 未压缩大小次低字节
+字节7: 未压缩大小次次低字节
+字节8: 未压缩大小次次次低字节
+字节9: 未压缩大小次高字节
+字节10: 未压缩大小次次高字节
+字节11: 未压缩大小次次次高字节
+字节12: 未压缩大小最高字节
+```
+
+**大小计算：**
+```c
+unpackSize = 0;
+for (i = 0; i < 8; i++) {
+ unpackSize += (UInt64)header[5 + i] << (i * 8);
+}
+```
+
+## 完整示例
+
+### 示例 1：典型的 LZMA 头部
+
+假设：
+- 压缩参数：`lc=3, lp=0, pb=2`
+- 字典大小：16MB（默认，用0表示）
+- 未压缩大小：120 字节
+
+**13字节头部：**
+```
+字节0: 0x5D (lc=3, lp=0, pb=2 的编码)
+字节1: 0x00 (字典大小低字节)
+字节2: 0x00 (字典大小次低字节)
+字节3: 0x00 (字典大小次高字节)
+字节4: 0x00 (字典大小高字节，0表示默认16MB)
+字节5: 0x78 (120 的十六进制，最低字节)
+字节6: 0x00 (次低字节)
+字节7: 0x00 (次次低字节)
+字节8: 0x00 (次次次低字节)
+字节9: 0x00 (次高字节)
+字节10: 0x00 (次次高字节)
+字节11: 0x00 (次次次高字节)
+字节12: 0x00 (最高字节)
+```
+
+**十六进制表示：**
+```
+5D 00 00 00 00 78 00 00 00 00 00 00 00
+```
+
+### 示例 2：更大的文件
+
+假设：
+- 压缩参数：`lc=3, lp=0, pb=2`
+- 字典大小：16MB（默认）
+- 未压缩大小：1024 字节 = 0x400
+
+**13字节头部：**
+```
+5D 00 00 00 00 00 04 00 00 00 00 00 00
+```
+
+**解析：**
+- `5D`: 压缩参数
+- `00 00 00 00`: 字典大小（默认）
+- `00 04 00 00 00 00 00 00`: 1024（小端）= 0x0000000000000400
+
+### 示例 3：实际 EPO 文件
+
+假设打包了 3 个文件，总大小 120 字节：
+- 文件数量：3 (4字节)
+- 文件1信息+数据：40字节
+- 文件2信息+数据：40字节
+- 文件3信息+数据：36字节
+- **总计：120字节**
+
+**13字节头部：**
+```
+5D 00 00 00 00 78 00 00 00 00 00 00 00
+```
+
+## 头部验证方法
+
+### 方法 1：使用十六进制查看器
+
+```bash
+# 查看文件前13字节
+xxd -l 13 epo.lzma
+```
+
+**预期输出：**
+```
+00000000: 5d00 0000 0078 0000 0000 0000 00 ]....x.......
+```
+
+### 方法 2：Python 验证脚本
+
+```python
+with open('epo.lzma', 'rb') as f:
+ header = f.read(13)
+
+ # 解析压缩参数
+ prop_byte = header[0]
+ lc = prop_byte % 9
+ d = prop_byte // 9
+ pb = d // 5
+ lp = d % 5
+ print(f"压缩参数: lc={lc}, lp={lp}, pb={pb}")
+
+ # 解析字典大小
+ dic_size = header[1] | (header[2] << 8) | (header[3] << 16) | (header[4] << 24)
+ print(f"字典大小: {dic_size} ({dic_size/1024/1024:.1f}MB)" if dic_size > 0 else "字典大小: 默认(16MB)")
+
+ # 解析未压缩大小
+ unpack_size = 0
+ for i in range(8):
+ unpack_size += header[5 + i] << (i * 8)
+ print(f"未压缩大小: {unpack_size} 字节")
+```
+
+## 常见问题
+
+### Q1: 为什么是 13 字节？
+A: LZMA SDK 标准格式要求 5 字节属性 + 8 字节未压缩大小 = 13 字节。
+
+### Q2: 字典大小为什么常用 0？
+A: 0 表示使用默认值（通常 16MB），这是最常用的配置。
+
+### Q3: 未压缩大小为什么是 8 字节？
+A: 8 字节（64位）可以表示最大 18EB 的文件大小，足够支持任何实际文件。
+
+### Q4: 如何验证头部是否正确？
+A:
+1. 检查前 5 字节是否符合 LZMA 属性格式
+2. 检查后 8 字节的未压缩大小是否与实际打包数据大小一致
+3. 使用解压函数测试是否能正确解压
+
+## 参考
+
+- LZMA SDK 文档：`code/sdk/open_source/7-zip-lzma-sdk/lzma_21.07/C/LzmaDec.h`
+- 解压实现：`code/sdk/zhihui/utilscpp/custom_utilscpp.cpp` 的 `Decode()` 函数
+
--- /dev/null
+# LZMA 13字节头部具体示例
+
+## 快速理解
+
+LZMA 文件的**前13个字节**是头部信息，告诉解压器：
+1. 如何解压（压缩参数）
+2. 解压后数据有多大（未压缩大小）
+
+## 实际示例
+
+### 示例：解压后数据为 120 字节的 LZMA 文件
+
+**完整的13字节头部：**
+```
+5D 00 00 00 00 78 00 00 00 00 00 00 00
+```
+
+**逐字节解释：**
+
+| 字节位置 | 十六进制值 | 含义 | 说明 |
+|---------|-----------|------|------|
+| 0 | `5D` | 压缩参数 | lc=3, lp=0, pb=2 的编码值 |
+| 1 | `00` | 字典大小(低) | 字典大小的最低字节 |
+| 2 | `00` | 字典大小(中低) | 字典大小的次低字节 |
+| 3 | `00` | 字典大小(中高) | 字典大小的次高字节 |
+| 4 | `00` | 字典大小(高) | 0表示使用默认16MB |
+| 5 | `78` | 大小(最低) | 120的十六进制 = 0x78 |
+| 6 | `00` | 大小(次低) | |
+| 7 | `00` | 大小(次次低) | |
+| 8 | `00` | 大小(次次次低) | |
+| 9 | `00` | 大小(次高) | |
+| 10 | `00` | 大小(次次高) | |
+| 11 | `00` | 大小(次次次高) | |
+| 12 | `00` | 大小(最高) | |
+
+**验证：**
+- 未压缩大小 = `0x78` + `0x00*256` + ... = 120 字节 ✓
+
+### 示例：解压后数据为 1024 字节的 LZMA 文件
+
+**13字节头部：**
+```
+5D 00 00 00 00 00 04 00 00 00 00 00 00
+```
+
+**解析：**
+- 字节0: `5D` = 压缩参数
+- 字节1-4: `00 00 00 00` = 字典大小（默认）
+- 字节5-12: `00 04 00 00 00 00 00 00` = 1024（小端）
+
+**验证：**
+- 1024 = 0x400
+- 小端表示：`00 04 00 00 00 00 00 00` = 0x0000000000000400 = 1024 ✓
+
+### 示例：解压后数据为 65536 字节的 LZMA 文件
+
+**13字节头部：**
+```
+5D 00 00 00 00 00 00 01 00 00 00 00 00
+```
+
+**解析：**
+- 字节5-12: `00 00 01 00 00 00 00 00` = 65536（小端）
+
+**验证：**
+- 65536 = 0x10000
+- 小端：`00 00 01 00 00 00 00 00` = 0x0000000000010000 = 65536 ✓
+
+## 如何查看文件的13字节头部
+
+### 方法1：使用 xxd 命令
+```bash
+xxd -l 13 epo.lzma
+```
+
+**输出示例：**
+```
+00000000: 5d00 0000 0078 0000 0000 0000 00 ]....x.......
+```
+
+### 方法2：使用 Python
+```python
+with open('epo.lzma', 'rb') as f:
+ header = f.read(13)
+ print(' '.join(f'{b:02X}' for b in header))
+```
+
+**输出：**
+```
+5D 00 00 00 00 78 00 00 00 00 00 00 00
+```
+
+### 方法3：使用 hexdump 命令
+```bash
+hexdump -C -n 13 epo.lzma
+```
+
+**输出：**
+```
+00000000 5d 00 00 00 00 78 00 00 00 00 00 00 00 |]....x........|
+```
+
+## 验证头部是否正确
+
+### 检查点1：压缩参数（字节0）
+- 应该是 `5D` 或 `89`（常见值）
+- 如果为 `00` 或其他异常值，可能格式错误
+
+### 检查点2：字典大小（字节1-4）
+- 通常为 `00 00 00 00`（使用默认值）
+- 如果非零，检查是否合理（通常 1MB-64MB）
+
+### 检查点3：未压缩大小（字节5-12）
+- 应该与实际打包数据大小一致
+- 使用小端格式读取
+- 如果为全0，表示文件为空（异常）
+
+## 常见错误
+
+### 错误1：缺少头部
+```
+文件直接是压缩数据，没有13字节头部
+→ 解压会失败，提示格式错误
+```
+
+### 错误2：头部大小错误
+```
+头部只有5字节或8字节
+→ 解压会失败，无法读取完整头部
+```
+
+### 错误3：未压缩大小不匹配
+```
+头部中的大小与实际打包数据不一致
+→ 解压可能失败或数据损坏
+```
+
+## 快速检查脚本
+
+```python
+#!/usr/bin/env python3
+import sys
+
+def check_lzma_header(filename):
+ with open(filename, 'rb') as f:
+ header = f.read(13)
+
+ if len(header) < 13:
+ print(f"错误: 文件头部不足13字节（只有{len(header)}字节）")
+ return False
+
+ # 解析压缩参数
+ prop = header[0]
+ lc = prop % 9
+ d = prop // 9
+ pb = d // 5
+ lp = d % 5
+
+ # 解析字典大小
+ dic_size = header[1] | (header[2] << 8) | (header[3] << 16) | (header[4] << 24)
+
+ # 解析未压缩大小
+ unpack_size = 0
+ for i in range(8):
+ unpack_size += header[5 + i] << (i * 8)
+
+ print(f"LZMA 头部检查:")
+ print(f" 压缩参数: lc={lc}, lp={lp}, pb={pb} (编码=0x{prop:02X})")
+ print(f" 字典大小: {dic_size} ({dic_size/1024/1024:.1f}MB)" if dic_size > 0 else " 字典大小: 默认(16MB)")
+ print(f" 未压缩大小: {unpack_size} 字节")
+ print(f" 头部十六进制: {' '.join(f'{b:02X}' for b in header)}")
+
+ return True
+
+if __name__ == '__main__':
+ if len(sys.argv) < 2:
+ print("用法: python3 check_lzma_header.py <lzma文件>")
+ sys.exit(1)
+
+ check_lzma_header(sys.argv[1])
+```
+
+**使用方法：**
+```bash
+python3 check_lzma_header.py epo.lzma
+```
+
--- /dev/null
+#!/usr/bin/env python3
+# -*- coding: utf-8 -*-
+"""
+分析 LZMA Alone 文件头信息并可选解压校验。
+可用于对比两份 .lzma（例如 iOS 生成 vs 本地 create_epo_lzma.py）。
+
+输出：
+- 压缩大小
+- 属性字节解析：lc/lp/pb、dict_size
+- 未压缩大小字段（8 字节）
+- 是否能解压及解压后大小
+"""
+
+import argparse
+import binascii
+import lzma
+import sys
+from pathlib import Path
+
+
+def parse_props(props: bytes):
+ if len(props) != 5:
+ raise ValueError("props 长度应为 5 字节")
+ prop0 = props[0]
+ lc = prop0 % 9
+ tmp = prop0 // 9
+ lp = tmp % 5
+ pb = tmp // 5
+ dict_size = int.from_bytes(props[1:5], "little")
+ return lc, lp, pb, dict_size
+
+
+def analyze_one(path: Path, do_decompress: bool):
+ info = {"path": str(path)}
+ data = path.read_bytes()
+ info["compressed_size"] = len(data)
+ if len(data) < 13:
+ info["error"] = "文件长度不足 13 字节"
+ return info
+
+ props = data[:5]
+ lc, lp, pb, dict_size = parse_props(props)
+ raw_size = int.from_bytes(data[5:13], "little")
+ info.update(
+ {
+ "props_hex": props.hex(),
+ "lc": lc,
+ "lp": lp,
+ "pb": pb,
+ "dict_size": dict_size,
+ "raw_size_field": raw_size,
+ "head16": data[:16].hex(),
+ }
+ )
+
+ if do_decompress:
+ try:
+ raw = lzma.decompress(data, format=lzma.FORMAT_ALONE)
+ info["decompress_ok"] = True
+ info["decompressed_size"] = len(raw)
+ info["raw_prefix_hex"] = binascii.hexlify(raw[:64]).decode()
+ except Exception as e:
+ info["decompress_ok"] = False
+ info["decompress_error"] = str(e)
+ return info
+
+
+def main():
+ ap = argparse.ArgumentParser(description="分析 LZMA Alone 头信息")
+ ap.add_argument("files", nargs="+", help="待分析的 .lzma 文件")
+ ap.add_argument(
+ "--no-decompress", action="store_true", help="不尝试解压，只解析头部"
+ )
+ args = ap.parse_args()
+
+ do_decompress = not args.no_decompress
+ for p in args.files:
+ info = analyze_one(Path(p), do_decompress)
+ print("=" * 60)
+ print(info["path"])
+ if "error" in info:
+ print(f"error: {info['error']}")
+ continue
+ print(f"compressed_size: {info['compressed_size']}")
+ print(
+ f"props: {info['props_hex']} (lc={info['lc']}, lp={info['lp']}, pb={info['pb']}, dict_size={info['dict_size']})"
+ )
+ print(f"raw_size_field: {info['raw_size_field']}")
+ print(f"head16: {info['head16']}")
+ if do_decompress:
+ if info.get("decompress_ok"):
+ print(f"decompressed_size: {info['decompressed_size']}")
+ print(f"raw_prefix_hex: {info.get('raw_prefix_hex','')}")
+ else:
+ print(f"decompress_error: {info.get('decompress_error')}")
+
+
+if __name__ == "__main__":
+ sys.exit(main())
+
--- /dev/null
+#!/usr/bin/env python3
+# -*- coding: utf-8 -*-
+"""
+创建 EPO LZMA 测试文件
+用于测试手表端的 uapi_epo_lzma_decompress 函数
+
+文件格式：
+[文件数量: 4字节，小端]
+[文件1: 文件名长度(2字节，小端) + 文件名(变长) + 文件大小(4字节，小端) + 文件数据(变长)]
+[文件2: ...]
+...
+
+然后使用 LZMA 压缩整个打包数据
+"""
+
+import struct
+import lzma
+import os
+import sys
+
+def create_epo_lzma(output_file="epo.lzma", test_files=None):
+ """
+ 创建 EPO LZMA 测试文件
+
+ Args:
+ output_file: 输出的 LZMA 文件名
+ test_files: 测试文件列表，格式为 [(文件名, 文件内容), ...]
+ """
+ if test_files is None:
+ # 默认测试文件
+ test_files = [
+ ("1.eph", b"EPH file content for testing\nThis is the first file.\n"),
+ ("2.txt", b"TXT file content for testing\nThis is the second file.\nLine 2\nLine 3\n"),
+ ("3.dat", b"DAT\x00\x01\x02\x03\x04\x05\x06\x07\x08\x09\x0A\x0B\x0C\x0D\x0E\x0F" +
+ b"Binary data for testing\n" * 10)
+ ]
+
+ # 创建打包数据
+ packed_data = bytearray()
+
+ # 写入文件数量（4字节，小端）
+ file_count = len(test_files)
+ packed_data.extend(struct.pack('<I', file_count))
+ print(f"文件数量: {file_count}")
+
+ # 写入每个文件
+ for idx, (filename, file_content) in enumerate(test_files):
+ filename_bytes = filename.encode('utf-8')
+ filename_len = len(filename_bytes)
+ file_size = len(file_content)
+
+ print(f"\n文件 {idx + 1}:")
+ print(f" 文件名: {filename}")
+ print(f" 文件名长度: {filename_len}")
+ print(f" 文件大小: {file_size} 字节")
+
+ # 写入文件名长度（2字节，小端）
+ packed_data.extend(struct.pack('<H', filename_len))
+
+ # 写入文件名
+ packed_data.extend(filename_bytes)
+
+ # 写入文件大小（4字节，小端）
+ packed_data.extend(struct.pack('<I', file_size))
+
+ # 写入文件内容
+ packed_data.extend(file_content)
+
+ print(f"\n打包数据总大小: {len(packed_data)} 字节")
+
+ # 使用 LZMA 压缩
+ print("\n开始 LZMA 压缩...")
+
+ # 保存打包数据到临时文件（用于调试）
+ temp_packed = "temp_packed_data.bin"
+ with open(temp_packed, 'wb') as f:
+ f.write(packed_data)
+ print(f"打包数据已保存到: {temp_packed} (用于调试)")
+
+ # 使用 Python lzma 模块压缩（LZMA1 格式）
+ try:
+ # 使用 LZMA1 格式压缩（原始 LZMA，不是 XZ）
+ compressed_data = lzma.compress(
+ packed_data,
+ format=lzma.FORMAT_RAW,
+ filters=[
+ {
+ 'id': lzma.FILTER_LZMA1,
+ 'preset': 6, # 压缩级别 0-9，6 是默认值
+ 'lc': 3, # Literal context bits
+ 'lp': 0, # Literal pos bits
+ 'pb': 2, # Pos bits
+ }
+ ]
+ )
+
+ print(f"压缩成功: {len(packed_data)} -> {len(compressed_data)} 字节")
+
+ # 添加 LZMA SDK 格式头部（5字节属性 + 8字节未压缩大小）
+ lzma_header = bytearray()
+
+ # 5字节 LZMA 属性
+ # LZMA SDK 格式: 第一个字节是属性，后4字节是字典大小（可选）
+ # 属性字节计算: (lc * 9 + lp) * 5 + pb
+ # lc=3, lp=0, pb=2: (3*9+0)*5+2 = 137 = 0x89
+ # 但 0x5D 更常用: lc=3, lp=0, pb=2, dict_size=16MB
+ lzma_props = bytearray(5)
+ lzma_props[0] = 0x5D # 属性字节
+ lzma_props[1] = 0x00 # 字典大小（低字节）
+ lzma_props[2] = 0x00
+ lzma_props[3] = 0x00
+ lzma_props[4] = 0x00 # 字典大小（高字节，0表示默认16MB）
+ lzma_header.extend(lzma_props)
+
+ # 8字节未压缩大小（小端）
+ unpacked_size = len(packed_data)
+ lzma_header.extend(struct.pack('<Q', unpacked_size))
+
+ print(f"LZMA 头部: {len(lzma_header)} 字节")
+ print(f" 属性: {[hex(b) for b in lzma_props]}")
+ print(f" 未压缩大小: {unpacked_size} 字节")
+
+ # 组合最终数据
+ final_data = lzma_header + compressed_data
+
+ except Exception as e:
+ print(f"Python lzma 压缩失败: {e}")
+ print("\n错误: 无法使用 Python lzma 模块创建 LZMA 文件")
+ print("请参考 EPO_LZMA_TEST.md 文档使用其他方法")
+ if os.path.exists(temp_packed):
+ print(f"打包数据已保存到: {temp_packed}")
+ print("可以使用其他工具（如 7z）手动压缩")
+ return None
+
+ print(f"压缩后大小: {len(compressed_data)} 字节")
+ print(f"LZMA 头部: {len(lzma_header)} 字节")
+ print(f"最终文件大小: {len(final_data)} 字节")
+ print(f"压缩比: {len(packed_data) / len(final_data):.2f}:1")
+
+ # 写入文件
+ with open(output_file, 'wb') as f:
+ f.write(final_data)
+
+ print(f"\n✓ 成功创建 LZMA 文件: {output_file}")
+ print(f" 可以传输到手表进行测试")
+
+ return output_file
+
+
+def create_from_real_files(file_list, output_file="epo.lzma"):
+ """
+ 从真实文件创建 EPO LZMA
+
+ Args:
+ file_list: 文件路径列表，如 ["1.eph", "2.txt", "3.dat"]
+ output_file: 输出的 LZMA 文件名
+ """
+ test_files = []
+ for filepath in file_list:
+ if not os.path.exists(filepath):
+ print(f"警告: 文件不存在: {filepath}")
+ continue
+
+ filename = os.path.basename(filepath)
+ with open(filepath, 'rb') as f:
+ content = f.read()
+
+ test_files.append((filename, content))
+ print(f"添加文件: {filename} ({len(content)} 字节)")
+
+ if not test_files:
+ print("错误: 没有有效的文件")
+ return None
+
+ return create_epo_lzma(output_file, test_files)
+
+
+def main():
+ """主函数"""
+ if len(sys.argv) > 1:
+ # 从命令行参数指定的文件创建
+ file_list = sys.argv[1:-1] if len(sys.argv) > 2 else sys.argv[1:]
+ output_file = sys.argv[-1] if sys.argv[-1].endswith('.lzma') else "epo.lzma"
+
+ if not sys.argv[-1].endswith('.lzma'):
+ file_list = sys.argv[1:]
+ output_file = "epo.lzma"
+
+ create_from_real_files(file_list, output_file)
+ else:
+ # 使用默认测试文件
+ print("=" * 70)
+ print("创建 EPO LZMA 测试文件")
+ print("=" * 70)
+ print("\n使用默认测试文件:")
+ create_epo_lzma()
+ print("\n" + "=" * 70)
+ print("使用方法:")
+ print(" python3 create_epo_lzma.py # 使用默认测试文件")
+ print(" python3 create_epo_lzma.py 1.eph 2.txt 3.dat epo.lzma # 从真实文件创建")
+ print("=" * 70)
+
+
+if __name__ == '__main__':
+ main()
+
--- /dev/null
+#!/usr/bin/env python3
+# -*- coding: utf-8 -*-
+"""
+简易 EPO LZMA 解包脚本
+格式：
+- LZMA 头 13 字节（5 字节属性 + 8 字节原始长度）
+- 解压后的数据:
+ uint32 file_count
+ 重复 file_count 次：
+ uint16 name_len
+ name_len 字节文件名
+ uint32 file_size
+ file_size 字节文件数据
+
+解包规则（与设备端保持一致）：
+- 从文件名中取下划线后的时间戳作为目录，例如 GLO_1765263582.eph -> /user/userdata/gnss/1765263582/
+- AssistInfo_*.dat 重命名为 AssistInfo.dat 后写入对应时间戳目录
+"""
+
+import argparse
+import lzma
+import os
+import struct
+from pathlib import Path
+
+
+def extract_ts(name: str) -> str:
+ """提取文件名中 '_' 与 '.' 之间的时间戳子串"""
+ try:
+ us = name.rindex("_")
+ dot = name.rindex(".")
+ if dot <= us + 1:
+ return ""
+ return name[us + 1 : dot]
+ except ValueError:
+ return ""
+
+
+def parse_and_write(raw: bytes, out_root: Path) -> None:
+ pos = 0
+ if len(raw) < 4:
+ raise ValueError("解压数据过短，无法读取文件数量")
+ (file_count,) = struct.unpack_from("<I", raw, pos)
+ pos += 4
+ print(f"[info] file_count={file_count}")
+ if file_count == 0 or file_count > 1000:
+ print(
+ f"[warn] 文件数量异常({file_count})，尝试使用无文件数量头的备用格式解析..."
+ )
+ return parse_and_write_no_count(raw, out_root)
+
+ # 先扫描 AssistInfo_ 时间戳
+ assist_ts = ""
+ scan_pos = pos
+ for _ in range(file_count):
+ if scan_pos + 2 > len(raw):
+ break
+ (name_len,) = struct.unpack_from("<H", raw, scan_pos)
+ scan_pos += 2
+ if name_len == 0 or name_len > 255 or scan_pos + name_len + 4 > len(raw):
+ break
+ name = raw[scan_pos : scan_pos + name_len].decode("utf-8", errors="ignore")
+ scan_pos += name_len
+ (file_size,) = struct.unpack_from("<I", raw, scan_pos)
+ scan_pos += 4 + file_size
+ if name.startswith("AssistInfo_"):
+ assist_ts = extract_ts(name)
+ break
+
+ for idx in range(file_count):
+ if pos + 2 > len(raw):
+ raise ValueError(f"文件 {idx} 读取 name_len 失败")
+ (name_len,) = struct.unpack_from("<H", raw, pos)
+ pos += 2
+ if name_len == 0 or name_len > 255:
+ raise ValueError(f"文件 {idx} name_len 非法: {name_len}")
+ if pos + name_len + 4 > len(raw):
+ raise ValueError(f"文件 {idx} 读取文件名或大小失败")
+ name = raw[pos : pos + name_len].decode("utf-8", errors="ignore")
+ pos += name_len
+ (file_size,) = struct.unpack_from("<I", raw, pos)
+ pos += 4
+ if pos + file_size > len(raw):
+ raise ValueError(f"文件 {idx} 数据越界")
+ data = raw[pos : pos + file_size]
+ pos += file_size
+
+ dir_ts = assist_ts or extract_ts(name)
+ if not dir_ts:
+ raise ValueError(f"文件 {name} 无法解析时间戳")
+
+ final_name = "AssistInfo.dat" if name.startswith("AssistInfo_") else name
+ target_dir = out_root / dir_ts
+ target_dir.mkdir(parents=True, exist_ok=True)
+ target_path = target_dir / final_name
+ with open(target_path, "wb") as f:
+ f.write(data)
+ print(f"[ok] {target_path} ({file_size} bytes)")
+
+
+def main():
+ ap = argparse.ArgumentParser(description="解压 EPO LZMA 包到 /user/userdata/gnss/<ts>/")
+ ap.add_argument("lzma_file", help="输入 LZMA 文件路径")
+ ap.add_argument(
+ "-o",
+ "--output",
+ default="/user/userdata/gnss",
+ help="输出根目录，默认 /user/userdata/gnss",
+ )
+ args = ap.parse_args()
+
+ lzma_path = Path(args.lzma_file)
+ out_root = Path(args.output)
+ out_root.mkdir(parents=True, exist_ok=True)
+
+ print(f"[info] decompressing {lzma_path} -> {out_root}")
+ with open(lzma_path, "rb") as f:
+ compressed = f.read()
+ print(f"[info] compressed size={len(compressed)} bytes, head={compressed[:16].hex()}")
+
+ # 直接用 Python lzma 解，输入必须是裸 LZMA 头+数据
+ try:
+ raw = lzma.decompress(compressed, format=lzma.FORMAT_ALONE)
+ except lzma.LZMAError as e:
+ raise SystemExit(f"[error] LZMA 解码失败: {e}. 该文件可能不是裸 LZMA 格式。")
+
+ print(f"[info] decompressed size={len(raw)} bytes")
+ parse_and_write(raw, out_root)
+
+
+def parse_and_write_no_count(raw: bytes, out_root: Path) -> None:
+ """备用解析：没有 file_count 头部，直接按条目顺序解析"""
+ pos = 0
+ entries = []
+ while pos < len(raw):
+ if pos + 2 > len(raw):
+ raise ValueError(f"offset {pos}: 读取 name_len 失败")
+ (name_len,) = struct.unpack_from("<H", raw, pos)
+ pos += 2
+ if name_len == 0 or name_len > 255:
+ raise ValueError(f"offset {pos}: name_len 非法 {name_len}")
+ if pos + name_len + 4 > len(raw):
+ raise ValueError(f"offset {pos}: 读取文件名或大小失败")
+ name = raw[pos : pos + name_len].decode("utf-8", errors="ignore")
+ pos += name_len
+ (file_size,) = struct.unpack_from("<I", raw, pos)
+ pos += 4
+ if pos + file_size > len(raw):
+ raise ValueError(f"文件 {name} 数据越界")
+ data = raw[pos : pos + file_size]
+ pos += file_size
+ entries.append((name, data))
+
+ if not entries:
+ raise ValueError("未解析到任何文件条目")
+
+ assist_ts = ""
+ for name, _ in entries:
+ if name.startswith("AssistInfo_"):
+ assist_ts = extract_ts(name)
+ break
+
+ for idx, (name, data) in enumerate(entries):
+ dir_ts = assist_ts or extract_ts(name)
+ if not dir_ts:
+ raise ValueError(f"文件 {name} 无法解析时间戳")
+ final_name = "AssistInfo.dat" if name.startswith("AssistInfo_") else name
+ target_dir = out_root / dir_ts
+ target_dir.mkdir(parents=True, exist_ok=True)
+ target_path = target_dir / final_name
+ with open(target_path, "wb") as f:
+ f.write(data)
+ print(f"[ok] {idx+1}/{len(entries)} {target_path} ({len(data)} bytes)")
+
+
+if __name__ == "__main__":
+ main()
+
projects / mirrors / FD230.git / commitdiff