RT-Thread-基于MQTT协议的阿里云物联平台+移远4G模块OTA升级参考RT-Thread问答社区

基于MQTT协议的阿里云物联平台+移远4G模块OTA升级参考

发布于 2024-05-07 11:11:50 浏览：770 订阅该版

[tocm]

### 升级流程

![screenshot_image.png](https://oss-club.rt-thread.org/uploads/20240507/c8d8b4f9231e7ae5ee69d7541ed585ae.png.webp)
1. 图中j03xxxVT是我们云物联平台上的产品ID
2. 图中%s应该替换为设备名

### 云端操作
建立升级包。新固件的bin文件用base64编码，上传阿里云端

![screenshot_image.png](https://oss-club.rt-thread.org/uploads/20240507/a3cd97edc54f1a78863b5eeba3fee30b.png)

注意协议选择MQTT

![screenshot_image.png](https://oss-club.rt-thread.org/uploads/20240507/3f1899a082bd257d84db1a725523502d.png)

![screenshot_image.png](https://oss-club.rt-thread.org/uploads/20240507/bd6a300c781c7d36167257d3d7d8429e.png)

### 设备端
参考代码
1、定义并配置urc
```c
static struct at_urc urc_table[] = {
  /* 如果正好payload里面有 '\r\n' 那就截断了 所以用base64编码*/
  {"+QMTRECV:", "\r\n", urc_recv_cb}
};
......
at_set_urc_table(urc_table, sizeof(urc_table) / sizeof(urc_table[0]));
```
2、建立连接后，订阅升级主题
```c
  char buf[64];
  TOPIC_FILL_NAME(buf, SUB_TOPIC_OTA_DOWNLOAD, dev_name);
  if (mqtt_sub(buf, 0) == RT_EOK) {
    LOG_I("sub %s success", buf);
  } else {
    rt_thread_mdelay(1000);
    mqtt_sub(buf, 0);
  }
```
3、 编写urc回调
```c
static void urc_recv_cb(const at_client_t client, const char* data, rt_size_t size) {
  char rec_topic[64];
  log_d("rx server data: %d", size);
  // for (int i = 0; i < size; i++) {
  //   rt_kprintf("%c", data[i]);
  // }
  // rt_kprintf("\n\n");

int payload_len;
  
  // 提取topic
  // sscanf(data, "+QMTRECV: %d,%d,\"%[^\"]\",%d,", &cid, &mid, topic_buf, &payload_len);
  sscanf(data, "%*[^\"]\"%[^\"]\",%d,", rec_topic, &payload_len);
  log_d("rec topic: %s", rec_topic);
  log_d("payload len: %d", payload_len);
  // ..."paylaod"\r\n, 减去\r\n和“” -4， 再忽略开头的" +1
  int payload_idx = size - payload_len - 3;
  
  TOPIC_FILL_NAME(topic_buf, SUB_TOPIC_OTA_UPGRADE, dev_name);
  if (rt_strcmp(topic_buf, rec_topic) == 0) {
    
    log_i("upgrade");
    ota_start_cb(data + payload_idx, payload_len);
  }
  TOPIC_FILL_NAME(topic_buf, SUB_TOPIC_OTA_DOWNLOAD, dev_name);
  if (rt_strcmp(topic_buf, rec_topic) == 0) {
    log_i("get file part");
    ota_pack_ind(data + payload_idx, payload_len);
  }
}

/* 固件文件信息提取 */
void ota_start_cb(const char* p, int len) {
  // 如果接收到的数据指针为空，则直接返回
  if (p == RT_NULL) {
    return;
  }
  // 创建一个字符数组来存储解析前的payload
  char payload[len + 1];

// 打印换行符，为日志信息做准备
  rt_kprintf("\n");
  // 将接收到的数据复制到payload数组中
  rt_memcpy(payload, p, len);
  // 在payload数组的末尾添加0，使其成为一个字符串
  payload[len] = 0;
  // 尝试解析payload为JSON对象
  cJSON* ota_info = cJSON_Parse(payload);

// 如果解析成功
  if (ota_info) {
    // 获取JSON对象中的"data"子对象
    cJSON* data = cJSON_GetObjectItem(ota_info, "data");
    // 如果"data"子对象存在
    if (data) {
      // 初始化用于存储固件信息的变量
      cJSON* iterm = RT_NULL;
      // 获取并存储固件大小
      iterm = cJSON_GetObjectItem(data, "size");
      file_size = iterm->valueint;
      // 获取并存储流ID
      iterm = cJSON_GetObjectItem(data, "streamId");
      stream_id = iterm->valueint;
      // 获取并存储流文件ID
      iterm = cJSON_GetObjectItem(data, "streamFileId");
      stream_file_id = iterm->valueint;
      // 打印固件信息日志
      LOG_D("file info: %d %d %d", file_size, stream_id, 
	         stream_file_id);
      // 释放解析用的JSON对象
      cJSON_Delete(ota_info);
      // 关闭传感器数据的上发
      sensor_pub_on = 0;

// 打印固件大小的估计值，用于调试和日志记录
      LOG_I("fw size est: %d", file_size * 3 / 4);
      // 初始化新的固件镜像存储区域
      stm32_flash_erase(APP_NEW_ADDR, file_size);
      // 创建并启动OTA升级线程
      rt_thread_t th = rt_thread_create("ota", ota_entry, RT_NULL,
                                         4096, 22, 10);
      if (th) {
        rt_thread_startup(th);
      }

}
  }
}
```
- 就两种消息，一是文件信息，一是固件包分片
- 固件包处理在另一个线程中进行

4、OTA线程
```c
static void ota_entry(void* param) {
  LOG_I("ota start");
  file_recv_sum = 0;
  flash_write_pos = 0;
  sem_pack_in = rt_sem_create("ota_pack", 0, RT_IPC_FLAG_PRIO);
  ota_fw_request(0, REQ_SIZE_ONCE);

while (1) {
    ota_in_progress();
    rt_thread_mdelay(50);
  }
}

void ota_in_progress(void) {
  uint32_t ofs = 0, size = 0;
  /*  
  json len | json string | file part | crc
  --------------------------------------------
    2bytes | n btyes     | m byte    | 2bytes

index  0   |  2          |  n + 2    | n + m + 2
  */
  // 等待OTA数据包的信号量
  rt_sem_take(sem_pack_in, RT_WAITING_FOREVER);

// 解析JSON长度
  uint16_t json_len = payload_buf[0] * 256 + payload_buf[1];

// 打印日志，用于调试
  rt_kprintf("\n");

// 准备解析JSON字符串的缓冲区
  char json_str[128];
  if (json_len < 128) {
    // 将JSON字符串拷贝到局部缓冲区，并添加字符串结束符
    rt_memcpy(json_str, payload_buf + 2, json_len);
    json_str[json_len] = 0;

// 解析JSON字符串，获取固件更新的相关信息
    cJSON* pack_info = cJSON_Parse(json_str);
    if (pack_info != RT_NULL) {
      // 尝试获取数据部分
      cJSON* data = cJSON_GetObjectItem(pack_info, "data");
      if (data != RT_NULL) {
        // 从数据部分获取偏移量和大小信息
        cJSON* iterm = cJSON_GetObjectItem(data, "bOffset");
        ofs = iterm->valueint;
        iterm = cJSON_GetObjectItem(data, "bSize");
        size = iterm->valueint;

// 释放JSON解析对象
        cJSON_Delete(pack_info);

// 打印日志，显示获取到的固件更新信息
        log_d("data: %d %d", ofs, size);
      } else {
        // 如果没有数据部分，尝试获取错误信息
        cJSON* msg = cJSON_GetObjectItem(pack_info, "message");
        if (msg != RT_NULL) {
          LOG_W("%s", msg->valuestring);
        }
        cJSON_Delete(msg);

// 报告更新失败的进度
        ota_report_progress(OTA_STEP_DOWNLOAD_FAILED);
        return;
      }

// 校验固件数据的CRC
      uint8_t* p_data = payload_buf + json_len + 2;
      uint16_t local_crc16 = Crc16Ibm(p_data, size);
      uint16_t rx_crc16 = *(uint16_t *)(p_data + size);
      if (rx_crc16 == local_crc16) {
        // CRC校验通过，进行Base64解码并将数据写入Flash
        int write_size = 512;
        int ret = tiny_base64_decode(fw_buf, &write_size, p_data, size);
        if (ret == 0) {
          // 写入Flash成功，更新写入位置
          LOG_I("flash write ofs/size: %d/%d", flash_write_pos,
		          write_size);
          stm32_flash_write(APP_NEW_ADDR + 4 + flash_write_pos, fw_buf,
		            write_size);
          flash_write_pos += write_size;
        } else {
          // Base64解码失败
          LOG_W("base64 decode failed %d", ret);
        }
      } else {
        // CRC校验失败
        LOG_E("crc check failed");
        ota_report_progress(OTA_STEP_CHECK_FAILED);
        return;
      }

// 更新已接收文件大小，并报告当前更新进度
      file_recv_sum += size;
      ota_report_progress(file_recv_sum * 100 / file_size);

// 如果尚未接收完整个文件，请求下一个数据分片
      if (file_recv_sum < file_size) {
        if (file_recv_sum + REQ_SIZE_ONCE < file_size) {
          ota_fw_request(file_recv_sum, REQ_SIZE_ONCE);
        } else {
          ota_fw_request(file_recv_sum, file_size - file_recv_sum);
        }
      } else {
        // 如果已经接收完全部文件，更新Flash中的写入位置标记，并重置系统
        LOG_D("ota finish, write total: %d", flash_write_pos);
        stm32_flash_write(APP_NEW_ADDR, (uint8_t*)&flash_write_pos, 4);
        extern void rt_hw_cpu_reset(void);
        rt_hw_cpu_reset();
      }
    }
  }
}
```

5、 请求固件包的接口，分片请求
```c
void ota_fw_request(int ofs, int size) {
  cJSON* json_root = cJSON_CreateObject();
  cJSON_AddStringToObject(json_root, "id", OTA_MSG_ID);
  cJSON_AddStringToObject(json_root, "version", "1.0");  // 协议版本，固定为1.0
  cJSON* params = cJSON_CreateObject();
  cJSON* file_info = cJSON_CreateObject();
  cJSON_AddNumberToObject(file_info, "streamId", stream_id);
  cJSON_AddNumberToObject(file_info, "fileId", stream_file_id);
  cJSON_AddItemToObject(params, "fileInfo", file_info);
  cJSON* file_block = cJSON_CreateObject();
  cJSON_AddNumberToObject(file_block, "size", size);
  cJSON_AddNumberToObject(file_block, "offset", ofs);
  cJSON_AddItemToObject(params, "fileBlock", file_block);
  cJSON_AddItemToObject(json_root, "params", params);
  char* payload = cJSON_PrintUnformatted(json_root);
  // LOG_D("send: %s", payload);
  char topic_buf[64];
  TOPIC_FILL_NAME(topic_buf, PUB_TOPIC_OTA_DOWNLOAD, dev_name);
  mqtt_pub(topic_buf, payload, rt_strlen(payload));
  rt_free(payload);
  cJSON_Delete(json_root);
}
```

### 说明：

- 阿里云遵照一机一密的原则，通信时要注意产品ID和设备名的匹配
- 一次请求字节数这个宏定义 `REQ_SIZE_ONCE` 我这里是640， 解码后应该是480字节
- mqtt通信主题根据你的云平台实际情况定义，我这`TOPIC_FILL_NAME` 这个宏是填充设备名的