物联网固件CI/CD流水线设计：ESP32自动化编译测试与OTA发布的工程实践

引言

物联网产品的研发正在从"手工作坊"走向"工业化流水线"。当一个 ESP32 项目从原型验证走向量产，面对的挑战不再是"能不能跑起来"，而是：如何保证每次发布的固件都经过充分测试？如何让十几个模块的修改不会互相破坏？如何在不中断设备运行的情况下完成远程升级？

答案就是 CI/CD（持续集成/持续部署）。但在嵌入式领域，CI/CD 的落地远比 Web 开发复杂——你需要交叉编译工具链、硬件在环测试、固件签名机制、OTA 分区管理和回滚策略。

沧州虎王科技技术团队在多个 ESP32 量产项目中，搭建了一套完整的固件 CI/CD 流水线，覆盖从代码提交到 OTA 发布的全流程。本文将这套工程实践完整拆解，包括 GitLab CI 配置、自动化编译、单元测试框架、固件签名与 OTA 发布机制，帮助团队快速建立自己的嵌入式 DevOps 体系。

一、嵌入式 CI/CD 整体架构

1.1 传统嵌入式开发 vs CI/CD 流程

┌──────────────────────────────────────────────────────────┐
│ 传统嵌入式开发流程 │
│ │
│ 开发者本地编译 ──→ 手动烧录 ──→ 手动测试 ──→ 发布 │
│ ↑ │ │
│ └────────── 发现bug回退 ─────────────┘ │
│ │
│ 问题: 耗时长、易出错、无法追溯、无法并行 │
└──────────────────────────────────────────────────────────┘

┌──────────────────────────────────────────────────────────┐
│ CI/CD 自动化流程 │
│ │
│ git push ──→ ┌──────────┐ │
│ │ 自动编译 │ ──→ 编译失败? → 通知开发者 │
│ └────┬─────┘ │
│ │ │
│ ┌────▼─────┐ │
│ │ 单元测试 │ ──→ 测试失败? → 通知开发者 │
│ └────┬─────┘ │
│ │ │
│ ┌────▼─────┐ │
│ │ 静态分析 │ ──→ 发现问题? → 标记警告 │
│ └────┬─────┘ │
│ │ │
│ ┌────▼─────┐ │
│ │ 固件打包 │ ──→ 生成bin + 签名 + 版本号 │
│ └────┬─────┘ │
│ │ │
│ ┌────▼─────┐ │
│ │ OTA发布 │ ──→ 推送到设备/灰度发布 │
│ └────┬─────┘ │
│ │ │
│ ┌────▼─────┐ │
│ │ 自动回滚 │ ←── 设备异常? 自动恢复 │
│ └──────────┘ │
│ │
│ 优势: 全自动、可追溯、可回滚、支持并行 │
└──────────────────────────────────────────────────────────┘

1.2 技术栈选型

二、自动化编译流水线

2.1 项目目录结构

一个规范的 ESP32 CI/CD 项目目录结构如下：

esp32-iot-project/
├── .gitlab-ci.yml # CI/CD 流水线配置
├── CMakeLists.txt # 顶层 CMake 配置
├── sdkconfig.defaults # 默认 SDK 配置
├── components/ # 自定义组件
│ ├── sensors/
│ │ ├── include/
│ │ ├── sensors.c
│ │ └── CMakeLists.txt
│ ├── wifi_manager/
│ │ ├── include/
│ │ ├── wifi_manager.c
│ │ └── CMakeLists.txt
│ └── ota_manager/
│ ├── include/
│ ├── ota_manager.c
│ └── CMakeLists.txt
├── main/
│ ├── main.c
│ └── CMakeLists.txt
├── test/ # 单元测试
│ ├── test_sensors.c
│ ├── test_wifi.c
│ └── CMakeLists.txt
├── scripts/ # 构建脚本
│ ├── build.sh
│ ├── test.sh
│ ├── sign_firmware.py
│ └── publish_ota.py
├── ci/ # CI 相关
│ ├── Dockerfile # 构建环境镜像
│ └── esp-idf-env/ # ESP-IDF 环境配置
└── docs/ # 文档
└── OTA_PROTOCOL.md

2.2 Docker 构建环境

为了确保 CI 环境一致性，使用 Docker 封装 ESP-IDF 编译环境：

# ci/Dockerfile
FROM espressif/idf:release-v5.1

# 安装额外工具
RUN apt-get update && apt-get install -y \\
python3-pip \\
cppcheck \\
clang-tidy \\
curl \\
jq \\
&& rm -rf /var/lib/apt/lists/*

# 安装 Python 依赖
COPY requirements-ci.txt /tmp/
RUN pip3 install -r /tmp/requirements-ci.txt

# 设置工作目录
WORKDIR /project

# 设置 ESP-IDF 环境变量
ENV IDF_PATH="/opt/esp/idf"
ENV PATH="${IDF_PATH}/tools:$PATH"

# 入口脚本
COPY ci/entrypoint.sh /
RUN chmod +x /entrypoint.sh
ENTRYPOINT ["/entrypoint.sh"]

2.3 GitLab CI 流水线配置

# .gitlab-ci.yml

# 定义阶段
stages:
– lint
– build
– test
– package
– publish

# 全局变量
variables:
IDF_VERSION: "v5.1"
FIRMWARE_VERSION: ""
GIT_STRATEGY: fetch

# 定义规则：仅在特定分支触发
workflow:
rules:
– if: $CI_COMMIT_BRANCH == "main"
variables:
FIRMWARE_VERSION: "release-${CI_COMMIT_SHORT_SHA}"
– if: $CI_PIPELINE_SOURCE == "merge_request_event"
variables:
FIRMWARE_VERSION: "pr-${CI_MERGE_REQUEST_IID}"
– if: $CI_COMMIT_BRANCH =~ /^release\\//
variables:
FIRMWARE_VERSION: "rc-${CI_COMMIT_SHORT_SHA}"

# ─── Lint 阶段：静态分析 ───
static-analysis:
stage: lint
image: ${CI_REGISTRY_IMAGE}/esp–idf–builder:${IDF_VERSION}
script:
– echo "运行静态分析…"
– cppcheck ––enable=all ––error–exitcode=1 \\
––suppress=missingIncludeSystem \\
components/ main/ 2>&1 | tee cppcheck_report.txt
– clang–tidy –p=build main/*.c components/*/*.c 2>&1 | tee clang_tidy_report.txt
– python3 scripts/check_code_style.py components/ main/
artifacts:
reports:
codequality: codequality–report.json
paths:
– cppcheck_report.txt
– clang_tidy_report.txt
expire_in: 7 days
rules:
– if: $CI_PIPELINE_SOURCE == "merge_request_event"
– if: $CI_COMMIT_BRANCH == "main"

# ─── Build 阶段：多目标编译 ───
.build_template: &build_template
stage: build
image: ${CI_REGISTRY_IMAGE}/esp–idf–builder:${IDF_VERSION}
before_script:
– source /opt/esp/idf/export.sh
– idf.py ––version
artifacts:
paths:
– build/firmware/
expire_in: 30 days
script:
– export FIRMWARE_VERSION="${FIRMWARE_VERSION}–${TARGET}"
– echo "编译固件版本: ${FIRMWARE_VERSION}"

# 配置目标芯片
– idf.py set–target ${TARGET}

# 注入版本号
– python3 scripts/inject_version.py \\
––version "${FIRMWARE_VERSION}" \\
––commit "${CI_COMMIT_SHA}" \\
––date "$(date –u +%Y–%m–%dT%H:%M:%SZ)"

# 编译
– idf.py build

# 收集产物
– mkdir –p build/firmware
– cp build/bootloader/bootloader.bin build/firmware/
– cp build/partition_table/partition–table.bin build/firmware/
– cp build/${PROJECT_NAME}.bin build/firmware/app_${TARGET}.bin
– cp build/ota_data_initial.bin build/firmware/ 2>/dev/null || true

# 生成版本信息文件
– |
cat > build/firmware/version.json << EOF
{
"version": "${FIRMWARE_VERSION}",
"target": "${TARGET}",
"commit": "${CI_COMMIT_SHA}",
"build_time": "$(date -u +%Y-%m-%dT%H:%M:%SZ)",
"idf_version": "${IDF_VERSION}"
}
EOF

# 生成合并固件（可直接烧录）
– python3 scripts/merge_bin.py \\
––chip ${TARGET} \\
––output build/firmware/flash_all_${TARGET}.bin \\
––bootloader build/bootloader/bootloader.bin \\
––partition_table build/partition_table/partition–table.bin \\
––app build/${PROJECT_NAME}.bin

# 编译 ESP32
build:esp32:
<<: *build_template
variables:
TARGET: "esp32"
PROJECT_NAME: "esp32-iot-firmware"

# 编译 ESP32-S3
build:esp32s3:
<<: *build_template
variables:
TARGET: "esp32s3"
PROJECT_NAME: "esp32-iot-firmware"

# 编译 ESP32-C3
build:esp32c3:
<<: *build_template
variables:
TARGET: "esp32c3"
PROJECT_NAME: "esp32-iot-firmware"

2.4 版本号注入机制

在编译时自动注入版本信息，便于运行时追踪和 OTA 版本管理：

// components/version_manager/version_manager.h
#pragma once

typedef struct {
char version[32]; // 固件版本号
char commit_hash[16]; // Git commit
char build_date[32]; // 编译时间
char idf_version[16]; // ESP-IDF版本
char target_chip[16]; // 目标芯片
} firmware_info_t;

const firmware_info_t *get_firmware_info(void);
const char *get_firmware_version_string(void);

// components/version_manager/version_manager.c
#include "version_manager.h"

// 这些宏由构建系统注入
#ifndef FIRMWARE_VERSION
#define FIRMWARE_VERSION "dev-unknown"
#endif
#ifndef FIRMWARE_COMMIT
#define FIRMWARE_COMMIT "unknown"
#endif
#ifndef BUILD_DATE
#define BUILD_DATE "unknown"
#endif
#ifndef IDF_VERSION_STR
#define IDF_VERSION_STR "unknown"
#endif
#ifndef TARGET_CHIP
#define TARGET_CHIP "unknown"
#endif

static const firmware_info_t s_fw_info = {
.version = FIRMWARE_VERSION,
.commit_hash = FIRMWARE_COMMIT,
.build_date = BUILD_DATE,
.idf_version = IDF_VERSION_STR,
.target_chip = TARGET_CHIP,
};

const firmware_info_t *get_firmware_info(void)
{
return &s_fw_info;
}

const char *get_firmware_version_string(void)
{
static char version_str[128];
snprintf(version_str, sizeof(version_str),
"%s (commit: %s, built: %s, %s %s)",
s_fw_info.version, s_fw_info.commit_hash,
s_fw_info.build_date, s_fw_info.idf_version,
s_fw_info.target_chip);
return version_str;
}

配合构建脚本注入版本号：

# scripts/inject_version.py
import argparse
import os
import re
from datetime import datetime, timezone

def inject_version(version, commit, date):
sdkconfig_path = "sdkconfig"

# 更新 sdkconfig 中的版本定义
config_lines = []
if os.path.exists(sdkconfig_path):
with open(sdkconfig_path, 'r') as f:
for line in f:
# 移除旧的版本定义
if not any(k in line for k in [
'FIRMWARE_VERSION', 'FIRMWARE_COMMIT',
'BUILD_DATE', 'IDF_VERSION_STR', 'TARGET_CHIP'
]):
config_lines.append(line)

# 追加新版本定义
config_lines.extend([
f'\\n# Firmware version info (auto-generated)\\n',
f'CONFIG_FIRMWARE_VERSION="{version}"\\n',
f'CONFIG_FIRMWARE_COMMIT="{commit[:12]}"\\n',
f'CONFIG_BUILD_DATE="{date}"\\n',
f'CONFIG_IDF_VERSION_STR="v5.1"\\n',
f'CONFIG_TARGET_CHIP="{os.environ.get("TARGET", "esp32")}"\\n',
])

with open(sdkconfig_path, 'w') as f:
f.writelines(config_lines)

print(f"版本信息已注入: {version} ({commit[:12]})")

if __name__ == "__main__":
parser = argparse.ArgumentParser()
server = parser.add_argument("–version", required=True)
parser.add_argument("–commit", required=True)
parser.add_argument("–date", required=True)
args = parser.parse_args()
inject_version(args.version, args.commit, args.date)

三、单元测试与硬件在环测试

3.1 Unity 测试框架集成

ESP-IDF 内置了 Unity 测试框架。对于可纯软件验证的组件，直接编写单元测试：

// test/test_sensors.c
#include "unity.h"
#include "sensors.h"
#include "mock_i2c.h" // 使用cmock模拟I2C

// 测试温度解析
void test_temperature_parsing(void)
{
// SHT30传感器返回的原始数据: 0x12 0x34 0x56
uint8_t raw_data[] = {0x12, 0x34, 0x56};
float temp = parse_sht30_temperature(raw_data);

// SHT30公式: T = -45 + 175 * (S_T / 65535)
// S_T = 0x1234 = 4660
// T = -45 + 175 * 4660 / 65535 = -45 + 12.44 = -32.56
TEST_ASSERT_FLOAT_WITHIN(0.1f, –32.56f, temp);
}

// 测试数据校验
void test_crc_validation(void)
{
uint8_t data[] = {0xBE, 0xEF};
uint8_t expected_crc = 0x92; // CRC8校验

uint8_t actual_crc = calculate_crc8(data, sizeof(data));
TEST_ASSERT_EQUAL(expected_crc, actual_crc);
}

// 测试异常处理
void test_sensor_timeout_handling(void)
{
// 模拟I2C超时
mock_i2c_set_timeout_response(true);

sensor_result_t result = read_sensor(I2C_NUM_0, SENSOR_ADDR);

TEST_ASSERT_EQUAL(SENSOR_ERROR_TIMEOUT, result.status);
TEST_ASSERT_EQUAL(0, result.retry_count); // 首次超时重试次数为0
}

// 测试数据滤波
void test_moving_average_filter(void)
{
filter_t filter;
filter_init(&filter, 5); // 5点滑动平均

float samples[] = {10.0, 12.0, 11.0, 13.0, 12.0};
float expected_avg = (10.0 + 12.0 + 11.0 + 13.0 + 12.0) / 5.0;

for (int i = 0; i < 5; i++) {
filter_update(&filter, samples[i]);
}

TEST_ASSERT_FLOAT_WITHIN(0.01f, expected_avg, filter_get_value(&filter));
}

// 测试批量数据缓冲
void test_data_buffer_overflow(void)
{
data_buffer_t buf;
buffer_init(&buf, 10);

// 填满缓冲区
for (int i = 0; i < 10; i++) {
TEST_ASSERT_TRUE(buffer_push(&buf, (float)i));
}

// 第11个应该失败
TEST_ASSERT_FALSE(buffer_push(&buf, 99.0f));

// 弹出后可以继续写入
float val;
TEST_ASSERT_TRUE(buffer_pop(&buf, &val));
TEST_ASSERT_EQUAL_FLOAT(0.0f, val);
TEST_ASSERT_TRUE(buffer_push(&buf, 99.0f));
}

void app_main(void)
{
UNITY_BEGIN();

RUN_TEST(test_temperature_parsing);
RUN_TEST(test_crc_validation);
RUN_TEST(test_sensor_timeout_handling);
RUN_TEST(test_moving_average_filter);
RUN_TEST(test_data_buffer_overflow);

UNITY_END();
}

3.2 CI 中的测试执行

# .gitlab-ci.yml 测试阶段

unit-test:host:
stage: test
image: ${CI_REGISTRY_IMAGE}/esp–idf–builder:${IDF_VERSION}
before_script:
– source /opt/esp/idf/export.sh
script:
– echo "运行单元测试 (Host模式)…"
– idf.py create–component–test–target
– cd build && ctest ––output–on–failure –j4
artifacts:
reports:
junit: build/test_results/*.xml
paths:
– build/test_results/
expire_in: 7 days
rules:
– if: $CI_PIPELINE_SOURCE == "merge_request_event"
– if: $CI_COMMIT_BRANCH == "main"

# 硬件在环测试（需要连接真实ESP32设备）
hardware-test:
stage: test
image: ${CI_REGISTRY_IMAGE}/esp–idf–builder:${IDF_VERSION}
tags:
– esp32–hardware # 需要有硬件的Runner
before_script:
– source /opt/esp/idf/export.sh
script:
– echo "硬件在环测试…"
# 烧录测试固件
– idf.py –p /dev/ttyUSB0 flash
# 运行测试并捕获输出
– timeout 120 idf.py –p /dev/ttyUSB0 monitor | tee monitor_output.txt
# 检查测试结果
– python3 scripts/parse_test_results.py monitor_output.txt
# 必须看到 "All tests passed"
– grep –q "All tests passed" monitor_output.txt
after_script:
# 保存测试日志和串口输出
– python3 scripts/upload_test_logs.py monitor_output.txt
rules:
– if: $CI_COMMIT_BRANCH == "main"
when: manual # 量产固件需要手动触发硬件测试
artifacts:
paths:
– monitor_output.txt
expire_in: 30 days

3.3 测试覆盖率统计

# scripts/check_coverage.py
import subprocess
import sys

def run_coverage_analysis():
"""运行覆盖率分析并生成报告"""

# 使用 gcov 收集覆盖率数据
subprocess.run([
"gcov", "build/esp-idf/main/*.gcno",
"-o", "build/coverage/"
], check=True)

# 生成 HTML 报告
subprocess.run([
"genhtml", "build/coverage/*.gcov",
"-o", "build/coverage/html"
], check=True)

# 解析覆盖率
result = subprocess.run(
["lcov", "–summary", "build/coverage.info"],
capture_output=True, text=True
)

# 检查是否达标
if "lines" in result.stdout:
# 提取覆盖率百分比
for line in result.stdout.split('\\n'):
if 'lines' in line.lower():
pct = float(line.split('%')[0].split(':')[–1].strip())
if pct < 70.0:
print(f"覆盖率 {pct}% 低于阈值 70%")
sys.exit(1)
print(f"覆盖率达标: {pct}%")

return True

if __name__ == "__main__":
run_coverage_analysis()

四、固件签名与安全打包

4.1 固件签名机制

量产固件必须签名，防止固件被篡改或非法刷入非授权设备：

┌──────────────────────────────────────────────────┐
│ 固件签名与验签流程 │
│ │
│ CI/CD 构建端: │
│ ┌──────────┐ ┌──────────┐ ┌───────────┐ │
│ │ 编译固件 │──→ │ 计算hash │──→ │ 私钥签名 │ │
│ │ app.bin │ │ SHA-256 │ │ RSA-3072 │ │
│ └──────────┘ └──────────┘ └─────┬─────┘ │
│ │ │
│ ┌────────────────────┘ │
│ ▼ │
│ ┌──────────┐ │
│ │ 签名固件 │ app-signed.bin │
│ │ = app + │ + signature sector │
│ │ sig │ │
│ └────┬─────┘ │
│ │ │
│ ────────────── OTA 推送 ────────────────── │
│ │ │
│ 设备端: ▼ │
│ ┌──────────┐ │
│ OTA下载 ──→ │ 验证签名 │ │
│ │ (公钥) │ │
│ └────┬─────┘ │
│ 签名正确? │ │
│ ├── 是 → 写入OTA分区 │
│ └── 否 → 拒绝升级 │
└──────────────────────────────────────────────────┘

4.2 生成密钥对与配置

#!/bin/bash
# scripts/generate_keys.sh

# 生成RSA-3072密钥对（用于固件签名）
openssl genrsa -out firmware_signing_key.pem 3072
openssl rsa -in firmware_signing_key.pem -pubout -out firmware_signing_key_pub.pem

echo "密钥对已生成:"
echo " 私钥: firmware_signing_key.pem (保密，仅CI环境使用)"
echo " 公钥: firmware_signing_key_pub.pem (烧入设备)"
echo ""
echo "注意: 私钥必须安全存储在CI变量中，不要提交到代码仓库"

在 sdkconfig 中启用安全启动：

# sdkconfig.defaults – 安全配置
CONFIG_SECURE_SIGNED_APPS_NO_SECURE_BOOT=y
CONFIG_SECURE_BOOT_BUILD_SIGNED_BINARIES=y
CONFIG_SECURE_BOOT_ENCRYPTION=n

# OTA 相关配置
CONFIG_APP_ROLLBACK_ENABLE=y # 启用应用回滚
CONFIG_BOOTLOADER_APP_ROLLBACK_ENABLE=y
CONFIG_BOOTLOADER_APP_ANTI_ROLLBACK=y # 防止固件降级
CONFIG_BOOTLOADER_APP_SEC_VER_SIZE_EFUSE=32

4.3 签名打包脚本

# scripts/sign_firmware.py
import hashlib
import subprocess
import argparse
import os
import json
from pathlib import Path
from datetime import datetime, timezone

def sign_firmware(firmware_path, private_key_path, output_dir):
"""使用 espsecure 签名固件"""

signed_path = os.path.join(output_dir, "app-signed.bin")

# 使用 ESP-IDF 的 espsecure 工具签名
cmd = [
"espsecure.py", "sign_data",
"–version", "2",
"–keyfile", private_key_path,
"–output", signed_path,
firmware_path
]

result = subprocess.run(cmd, capture_output=True, text=True)
if result.returncode != 0:
print(f"签名失败: {result.stderr}")
return None

# 计算签名后固件的哈希
with open(signed_path, 'rb') as f:
firmware_hash = hashlib.sha256(f.read()).hexdigest()

# 生成固件清单
manifest = {
"firmware": os.path.basename(signed_path),
"sha256": firmware_hash,
"size": os.path.getsize(signed_path),
"signed": True,
"sign_time": datetime.now(timezone.utc).isoformat(),
"algorithm": "RSA-3072-PSS"
}

manifest_path = os.path.join(output_dir, "manifest.json")
with open(manifest_path, 'w') as f:
json.dump(manifest, f, indent=2)

print(f"固件签名完成:")
print(f" 签名固件: {signed_path}")
print(f" SHA-256: {firmware_hash}")
print(f" 大小: {manifest['size']} bytes")

return signed_path

if __name__ == "__main__":
parser = argparse.ArgumentParser(description="ESP32固件签名工具")
parser.add_argument("–firmware", required=True, help="待签名固件路径")
parser.add_argument("–key", required=True, help="私钥路径")
parser.add_argument("–output", default="build/firmware", help="输出目录")
args = parser.parse_args()

os.makedirs(args.output, exist_ok=True)
sign_firmware(args.firmware, args.key, args.output)

五、OTA 远程升级系统

5.1 OTA 分区表设计

ESP32 的 OTA 机制依赖双分区设计。一个合理的分区表是 OTA 稳定运行的基础：

# partitions.csv
# Name, Type, SubType, Offset, Size, Flags
nvs, data, nvs, 0x9000, 0x4000,
phy_init, data, phy, 0xf000, 0x1000,
factory, app, factory, 0x10000, 0x100000,
ota_0, app, ota_0, 0x110000, 0x100000,
ota_1, app, ota_1, 0x210000, 0x100000,
otadata, data, ota, 0x310000, 0x2000,
storage, data, nvs, 0x312000, 0x20000,

对应的分区布局图：

┌──────────────────────────────────────────────────┐
│ ESP32 Flash 分区布局 (4MB) │
├──────────┬──────────┬────────────────────────────┤
│ 偏移地址 │ 分区名 │ 大小 │ 用途 │
├──────────┼──────────┼────────┼───────────────────┤
│ 0x0000 │ bootloader│ 0x8000 │ 引导加载器 │
│ 0x8000 │ partition │ 0x1000 │ 分区表 │
│ 0x9000 │ nvs │ 0x4000 │ 非易失存储 │
│ 0xF000 │ phy_init │ 0x1000 │ PHY校准数据 │
│ 0x10000 │ factory │ 1MB │ 出厂固件(恢复用) │
│ 0x110000 │ ota_0 │ 1MB │ OTA分区A │
│ 0x210000 │ ota_1 │ 1MB │ OTA分区B │
│ 0x310000 │ otadata │ 8KB │ OTA状态记录 │
│ 0x312000 │ storage │ 128KB │ 文件存储区 │
└──────────┴──────────┴────────┴───────────────────┘

5.2 OTA 客户端实现

#include "esp_ota_ops.h"
#include "esp_http_client.h"
#include "esp_https_ota.h"
#include "esp_log.h"
#include "esp_app_format.h"
#include "mbedtls/sha256.h"

static const char *TAG = "OTA_MANAGER";

#define OTA_URL_SIZE 256
#define OTA_BUF_SIZE 4096
#define OTA_TIMEOUT_MS 30000
#define MAX_RETRIES 3

typedef struct {
char current_version[32];
char new_version[32];
int total_size;
int downloaded_size;
bool is_downloading;
bool verify_success;
} ota_status_t;

static ota_status_t s_ota_status;

// 进度回调
typedef void (*ota_progress_cb_t)(int downloaded, int total);
static ota_progress_cb_t s_progress_cb = NULL;

// OTA 下载与写入
static esp_err_t ota_download_and_write(const char *url)
{
esp_http_client_config_t config = {
.url = url,
.timeout_ms = OTA_TIMEOUT_MS,
.buffer_size = OTA_BUF_SIZE,
.buffer_size_tx = 1024,
.cert_pem = NULL, // 自签名证书或NULL
.skip_cert_common_name_check = false,
};

esp_https_ota_config_t ota_config = {
.http_config = &config,
.bulk_flash_erase = false, // 逐块擦除，减少阻塞时间
.partial_http_download = false,
};

s_ota_status.is_downloading = true;
s_ota_status.downloaded_size = 0;

esp_err_t ret = esp_https_ota(&ota_config);
s_ota_status.is_downloading = false;

return ret;
}

// 自定义OTA流程：支持进度追踪和版本校验
esp_err_t perform_ota_upgrade(const char *firmware_url, const char *expected_version)
{
ESP_LOGI(TAG, "开始OTA升级: %s", firmware_url);

// 获取当前运行分区信息
const esp_partition_t *running = esp_ota_get_running_partition();
esp_ota_img_states_t running_state;
esp_ota_get_partition_state(running, &running_state);

ESP_LOGI(TAG, "当前运行分区: %s, 状态: %d",
running->label, running_state);

// 获取下一个OTA分区
const esp_partition_t *update_partition =
esp_ota_get_next_update_partition(NULL);
if (update_partition == NULL) {
ESP_LOGE(TAG, "无法获取OTA分区");
return ESP_FAIL;
}
ESP_LOGI(TAG, "目标OTA分区: %s (地址: 0x%08x)",
update_partition->label, update_partition->address);

// 初始化OTA
esp_ota_handle_t update_handle = 0;
esp_err_t err = esp_ota_begin(update_partition,
OTA_WITH_SEQUENTIAL_WRITES,
&update_handle);
if (err != ESP_OK) {
ESP_LOGE(TAG, "esp_ota_begin 失败: %s", esp_err_to_name(err));
return err;
}

// HTTP下载固件
esp_http_client_config_t http_config = {
.url = firmware_url,
.timeout_ms = OTA_TIMEOUT_MS,
.buffer_size = OTA_BUF_SIZE,
};

esp_http_client_handle_t client = esp_http_client_init(&http_config);
err = esp_http_client_open(client, 0);
if (err != ESP_OK) {
ESP_LOGE(TAG, "HTTP连接失败: %s", esp_err_to_name(err));
esp_http_client_cleanup(client);
esp_ota_abort(update_handle);
return err;
}

int total_size = esp_http_client_fetch_headers(client);
ESP_LOGI(TAG, "固件大小: %d bytes", total_size);
s_ota_status.total_size = total_size;

// 下载并写入
char *buffer = malloc(OTA_BUF_SIZE);
if (!buffer) {
esp_http_client_close(client);
esp_http_client_cleanup(client);
esp_ota_abort(update_handle);
return ESP_ERR_NO_MEM;
}

int data_read;
int written = 0;

while ((data_read = esp_http_client_read(client, buffer, OTA_BUF_SIZE)) > 0) {
err = esp_ota_write(update_handle, buffer, data_read);
if (err != ESP_OK) {
ESP_LOGE(TAG, "OTA写入失败: %s", esp_err_to_name(err));
break;
}
written += data_read;
s_ota_status.downloaded_size = written;

// 进度回调
if (s_progress_cb && total_size > 0) {
s_progress_cb(written, total_size);
}

// 每写入10%打印一次进度
if (total_size > 0 && (written % (total_size / 10)) < OTA_BUF_SIZE) {
ESP_LOGI(TAG, "下载进度: %d%% (%d/%d)",
(written * 100) / total_size, written, total_size);
}
}

free(buffer);
esp_http_client_close(client);
esp_http_client_cleanup(client);

if (written != total_size) {
ESP_LOGE(TAG, "下载不完整: %d/%d", written, total_size);
esp_ota_abort(update_handle);
return ESP_FAIL;
}

// 完成写入
err = esp_ota_end(update_handle);
if (err != ESP_OK) {
ESP_LOGE(TAG, "OTA结束失败: %s", esp_err_to_name(err));
return err;
}

// 设置启动分区
err = esp_ota_set_boot_partition(update_partition);
if (err != ESP_OK) {
ESP_LOGE(TAG, "设置启动分区失败: %s", esp_err_to_name(err));
return err;
}

ESP_LOGI(TAG, "OTA升级完成，重启后将运行新固件");
return ESP_OK;
}

5.3 OTA 回滚机制

OTA 升级后如果新固件有问题，必须有自动回滚机制防止设备变砖：

// OTA成功后的状态管理
void ota_post_upgrade_task(void *arg)
{
ESP_LOGI(TAG, "新固件首次启动，验证中…");

// 标记为待验证状态
esp_ota_img_states_t state;
const esp_partition_t *running = esp_ota_get_running_partition();
esp_err_t err = esp_ota_get_partition_state(running, &state);

if (err != ESP_OK || state != ESP_OTA_IMG_PENDING_VERIFY) {
ESP_LOGE(TAG, "分区状态异常");
return;
}

// 执行自检
bool self_test_passed = run_self_test();

if (self_test_passed) {
ESP_LOGI(TAG, "自检通过，确认新固件");
esp_ota_mark_app_valid_cancel_rollback();
ESP_LOGI(TAG, "固件已标记为有效");
} else {
ESP_LOGE(TAG, "自检失败，回滚到上一版本");
esp_ota_mark_app_invalid_rollback_and_reboot();
// 不会执行到这里，设备已重启
}
}

// 自检函数
bool run_self_test(void)
{
int checks_passed = 0;
int total_checks = 5;

// 1. 检查Flash完整性
if (check_flash_integrity()) {
checks_passed++;
ESP_LOGI(TAG, "[1/%d] Flash完整性检查 ✓", total_checks);
} else {
ESP_LOGE(TAG, "[1/%d] Flash完整性检查 ✗", total_checks);
}

// 2. 检查WiFi模块
if (check_wifi_module()) {
checks_passed++;
ESP_LOGI(TAG, "[2/%d] WiFi模块检查 ✓", total_checks);
} else {
ESP_LOGE(TAG, "[2/%d] WiFi模块检查 ✗", total_checks);
}

// 3. 检查传感器
if (check_sensors()) {
checks_passed++;
ESP_LOGI(TAG, "[3/%d] 传感器检查 ✓", total_checks);
} else {
ESP_LOGE(TAG, "[3/%d] 传感器检查 ✗", total_checks);
}

// 4. 检查内存
if (check_memory_health()) {
checks_passed++;
ESP_LOGI(TAG, "[4/%d] 内存健康检查 ✓", total_checks);
} else {
ESP_LOGE(TAG, "[4/%d] 内存健康检查 ✗", total_checks);
}

// 5. 检查与云端的通信
if (check_cloud_connectivity()) {
checks_passed++;
ESP_LOGI(TAG, "[5/%d] 云端通信检查 ✓", total_checks);
} else {
ESP_LOGE(TAG, "[5/%d] 云端通信检查 ✗", total_checks);
}

// 至少通过4项检查才算成功
return checks_passed >= 4;
}

5.4 OTA 完整状态机

┌─────────────────────────────────────────────────────────┐
│ OTA 升级状态机 │
│ │
│ ┌──────────┐ │
│ ┌────────│ IDLE │←─────────────┐ │
│ │ │ (空闲) │ │ │
│ │ └────┬─────┘ │ │
│ │ │ 收到升级命令 │ │
│ │ ┌────▼─────┐ │ │
│ │ │CHECKING │ │ │
│ │ │ (检查版本)│ │ │
│ │ └────┬─────┘ │ │
│ │ ┌────────┴────────┐ │ │
│ │ │ │ │ │
│ │ 版本相同 版本不同 │ │
│ │ │ │ │ │
│ │ │ ┌────▼─────┐ │ │
│ │ │ │DOWNLOAD │ │ │
│ │ │ │ (下载中) │ │ │
│ │ │ └────┬─────┘ │ │
│ │ │ │ │ │
│ │ │ ┌─────┴──────┐ │ │
│ │ │ │ │ │ │
│ │ │ 下载成功 下载失败 │ │
│ │ │ │ │ │ │
│ │ │ ┌──────▼──┐ ┌──────▼───┐ │
│ │ │ │WRITING │ │RETRYING │ │
│ │ │ │(写入Flash)│ │ (重试中) │ │
│ │ │ └──────┬──┘ └──────┬───┘ │
│ │ │ │ │ │
│ │ │ ┌────▼────┐ 重试3次失败 │
│ │ │ │REBOOTING│ ┌─────▼──────┐ │
│ │ │ │ (重启) │ │FAILED │ │
│ │ │ └────┬────┘ │ (失败) │ │
│ │ │ │ └────────────┘ │
│ │ │ ┌──────▼──────┐ │
│ │ │ │VERIFYING │ │
│ │ │ │ (自检验证) │ │
│ │ │ └──┬─────┬───┘ │
│ │ │ │ │ │
│ │ │ 验证通过 验证失败 │
│ │ │ │ │ │
│ │ │ ┌────▼──┐ ┌────▼──────┐ │
│ │ │ │VALID │ │ROLLBACK │ │
│ │ │ │(确认有效)│ │(回滚旧固件)│ │
│ │ │ └───┬───┘ └───────────┘ │
│ │ │ │ │
│ └─────┴──────┴────────────────────────────┘ │
│ │
└─────────────────────────────────────────────────────────┘

六、OTA 发布服务与设备管理

6.1 OTA 发布服务端

# scripts/publish_ota.py
"""OTA 固件发布服务"""
import json
import os
import shutil
import hashlib
from datetime import datetime
from pathlib import Path

class OTAReleaseManager:
def __init__(self, release_dir="/var/www/ota/releases"):
self.release_dir = Path(release_dir)
self.release_dir.mkdir(parents=True, exist_ok=True)
self.manifest_file = self.release_dir / "manifest.json"

def publish_firmware(self, firmware_path, version, target_chip,
rollout_percentage=100):
"""发布新固件"""

# 创建版本目录
version_dir = self.release_dir / version
version_dir.mkdir(exist_ok=True)

# 复制固件
firmware_name = f"app-{target_chip}.bin"
dest_path = version_dir / firmware_name
shutil.copy2(firmware_path, dest_path)

# 计算SHA-256
sha256 = self._calculate_sha256(dest_path)

# 生成版本信息
release_info = {
"version": version,
"target_chip": target_chip,
"firmware_url": f"/releases/{version}/{firmware_name}",
"sha256": sha256,
"size": os.path.getsize(dest_path),
"release_time": datetime.utcnow().isoformat() + "Z",
"rollout_percentage": rollout_percentage,
"status": "active"
}

# 更新清单
manifest = self._load_manifest()
manifest["releases"].append(release_info)
manifest["latest_version"] = version
self._save_manifest(manifest)

print(f"固件发布成功: v{version}")
print(f" URL: {release_info['firmware_url']}")
print(f" SHA-256: {sha256}")

return release_info

def _calculate_sha256(self, file_path):
h = hashlib.sha256()
with open(file_path, 'rb') as f:
while chunk := f.read(8192):
h.update(chunk)
return h.hexdigest()

def _load_manifest(self):
if self.manifest_file.exists():
with open(self.manifest_file, 'r') as f:
return json.load(f)
return {"releases": [], "latest_version": None}

def _save_manifest(self, manifest):
with open(self.manifest_file, 'w') as f:
json.dump(manifest, f, indent=2)

if __name__ == "__main__":
import sys

if len(sys.argv) < 4:
print("用法: publish_ota.py <firmware_path> <version> <target_chip>")
sys.exit(1)

manager = OTAReleaseManager()
manager.publish_firmware(sys.argv[1], sys.argv[2], sys.argv[3])

6.2 灰度发布机制

# scripts/rollout.py
"""灰度发布控制器"""
import json
import random
from pathlib import Path

class RolloutController:
def __init__(self, manifest_path):
self.manifest_path = Path(manifest_path)

def check_update_for_device(self, device_id, current_version):
"""检查设备是否应该收到更新"""
manifest = self._load_manifest()
latest = manifest.get("latest_version")

if not latest or latest == current_version:
return None # 无更新

# 找到最新版本的发布信息
release = None
for r in manifest["releases"]:
if r["version"] == latest:
release = r
break

if not release:
return None

# 灰度发布：基于设备ID哈希决定是否更新
rollout_pct = release.get("rollout_percentage", 100)
device_hash = int(hashlib.md5(device_id.encode()).hexdigest(), 16) % 100

if device_hash < rollout_pct:
return release # 该设备在灰度范围内
else:
return None # 不在灰度范围内

def increase_rollout(self, version, percentage):
"""增加灰度比例"""
manifest = self._load_manifest()
for r in manifest["releases"]:
if r["version"] == version:
r["rollout_percentage"] = percentage
self._save_manifest(manifest)
print(f"版本 {version} 灰度比例调整为 {percentage}%")
return
print(f"版本 {version} 未找到")

# CI/CD 中的灰度发布流程:
# 1. 初次发布: rollout_percentage = 5 (仅5%设备)
# 2. 观察24h无异常: rollout_percentage = 20
# 3. 继续观察: rollout_percentage = 50
# 4. 最终全量: rollout_percentage = 100

七、完整 CI/CD Pipeline 可视化

7.1 完整流水线概览

┌──────────────────────────────────────────────────────────────┐
│ ESP32 固件 CI/CD 完整流水线 │
├──────────────────────────────────────────────────────────────┤
│ │
│ 开发者 CI/CD 系统 设备端 │
│ ┌──────┐ ┌──────────────┐ ┌──────────┐ │
│ │ git │──push────→│ 1.Lint │ │ │ │
│ │ push │ │ 静态分析 │ │ ESP32 │ │
│ └──────┘ ├──────────────┤ │ 设备群 │ │
│ │ 2.Build │ │ │ │
│ │ 多目标编译 │ │ │ │
│ │ ESP32/S3/C3 │ │ │ │
│ ├──────────────┤ │ │ │
│ │ 3.Test │ │ │ │
│ │ 单元测试 │ │ │ │
│ │ 覆盖率检查 │ │ │ │
│ ├──────────────┤ │ │ │
│ │ 4.Sign │ │ │ │
│ │ 固件签名 │ │ │ │
│ │ 版本打包 │ │ │ │
│ ├──────────────┤ │ │ │
│ │ 5.Publish │ │ │ │
│ │ 上传OTA服务 │ │ │ │
│ │ 灰度5% │──OTA─→│ 5%设备 │ │
│ ├──────────────┤ │ 升级 │ │
│ │ 6.Monitor │←report│ │ │
│ │ 监控设备状态 │ │ │ │
│ │ 回滚检测 │ │ │ │
│ ├──────────────┤ │ │ │
│ │ 7.Scale │ │ │ │
│ │ 扩大灰度 │──OTA─→│ 50%设备 │ │
│ │ 50% → 100% │ │ 升级 │ │
│ └──────────────┘ └──────────┘ │
│ │
│ 监控面板: │
│ ┌──────────────────────────────────────────────────────┐ │
│ │ 物联网平台 OTA 管理界面 │ │
│ │ • 在线设备: 1,234 台 │ │
│ │ • 已升级: 1,180 台 (95.6%) │ │
│ │ • 升级失败: 12 台 (已自动回滚) │ │
│ │ • 待升级: 42 台 │ │
│ │ • 平均升级耗时: 45s │ │
│ │ • 固件版本: release-a1b2c3d4 │ │
│ └──────────────────────────────────────────────────────┘ │
└──────────────────────────────────────────────────────────────┘

7.2 设备端 OTA Agent

// components/ota_manager/ota_agent.c
#include "esp_http_client.h"
#include "esp_crt_bundle.h"
#include "freertos/task.h"
#include "freertos/event_groups.h"

#define OTA_CHECK_INTERVAL_S 3600 // 每小时检查一次更新
#define OTA_CHECK_URL "https://ota.example.com/api/check"
#define OTA_DOWNLOAD_URL_FMT "https://ota.example.com/releases/%s/app-esp32.bin"

static EventGroupHandle_t s_ota_events;
#define OTA_BIT_CHECKING (1 << 0)
#define OTA_BIT_DOWNLOADING (1 << 1)
#define OTA_BIT_REBOOTING (1 << 2)

// OTA Agent 主任务
void ota_agent_task(void *arg)
{
const char *current_version = get_firmware_version_string();
char device_id[32];
get_device_id(device_id, sizeof(device_id));

ESP_LOGI(TAG, "OTA Agent 启动, 当前版本: %s", current_version);

while (1) {
xEventGroupSetBits(s_ota_events, OTA_BIT_CHECKING);

// 检查更新
ota_update_info_t update_info;
esp_err_t ret = check_for_update(device_id, current_version, &update_info);

xEventGroupClearBits(s_ota_events, OTA_BIT_CHECKING);

if (ret == ESP_OK && update_info.update_available) {
ESP_LOGI(TAG, "发现新版本: %s (大小: %d bytes)",
update_info.version, update_info.size);

// 验证SHA-256
if (strlen(update_info.sha256) > 0) {
ESP_LOGI(TAG, "期望SHA-256: %s", update_info.sha256);
}

// 确保有足够电量（电池设备）
if (is_battery_powered() && get_battery_level() < 30) {
ESP_LOGW(TAG, "电量不足，跳过本次升级");
} else {
// 执行升级
xEventGroupSetBits(s_ota_events, OTA_BIT_DOWNLOADING);

char url[256];
snprintf(url, sizeof(url), OTA_DOWNLOAD_URL_FMT,
update_info.version);

ret = perform_ota_upgrade(url, update_info.version);

xEventGroupClearBits(s_ota_events, OTA_BIT_DOWNLOADING);

if (ret == ESP_OK) {
ESP_LOGI(TAG, "升级成功，3秒后重启…");
xEventGroupSetBits(s_ota_events, OTA_BIT_REBOOTING);
vTaskDelay(pdMS_TO_TICKS(3000));
esp_restart();
} else {
ESP_LOGE(TAG, "升级失败: %s, 将在下次检查时重试",
esp_err_to_name(ret));
report_ota_failure(device_id, ret);
}
}
}

// 等待下次检查
vTaskDelay(pdMS_TO_TICKS(OTA_CHECK_INTERVAL_S * 1000));
}
}

// 检查更新
esp_err_t check_for_update(const char *device_id,
const char *current_version,
ota_update_info_t *info)
{
char post_data[256];
snprintf(post_data, sizeof(post_data),
"{\\"device_id\\":\\"%s\\",\\"version\\":\\"%s\\"}",
device_id, current_version);

esp_http_client_config_t config = {
.url = OTA_CHECK_URL,
.method = HTTP_METHOD_POST,
.timeout_ms = 10000,
.crt_bundle_attach = esp_crt_bundle_attach,
};

esp_http_client_handle_t client = esp_http_client_init(&config);
esp_http_client_set_post_field(client, post_data, strlen(post_data));
esp_http_client_set_header(client, "Content-Type", "application/json");

esp_err_t err = esp_http_client_perform(client);
if (err == ESP_OK) {
int status = esp_http_client_get_status_code(client);
int length = esp_http_client_get_content_length(client);

if (status == 200 && length > 0) {
char *response = malloc(length + 1);
int read_len = esp_http_client_read(client, response, length);
response[read_len] = '\\0';

// 解析JSON响应
cJSON *json = cJSON_Parse(response);
cJSON *available = cJSON_GetObjectItem(json, "update_available");
cJSON *version = cJSON_GetObjectItem(json, "version");
cJSON *size = cJSON_GetObjectItem(json, "size");
cJSON *sha256 = cJSON_GetObjectItem(json, "sha256");

if (cJSON_IsTrue(available)) {
info->update_available = true;
strncpy(info->version, version->valuestring,
sizeof(info->version) – 1);
info->size = size->valueint;
strncpy(info->sha256, sha256->valuestring,
sizeof(info->sha256) – 1);
} else {
info->update_available = false;
}

cJSON_Delete(json);
free(response);
}
}

esp_http_client_cleanup(client);
return err;
}

7.3 GitLab CI 发布阶段完整配置

# .gitlab-ci.yml – 发布阶段

package:firmware:
stage: package
image: ${CI_REGISTRY_IMAGE}/esp–idf–builder:${IDF_VERSION}
needs:
– job: build:esp32
artifacts: true
– job: build:esp32s3
artifacts: true
– job: build:esp32c3
artifacts: true
script:
– echo "打包固件…"
– mkdir –p release_package

# 收集所有目标的固件
– for target in esp32 esp32s3 esp32c3; do
cp build/firmware/app_${target}.bin release_package/ || true;
cp build/firmware/flash_all_${target}.bin release_package/ || true;
done

# 生成合并固件包
– tar –czf firmware–${FIRMWARE_VERSION}.tar.gz release_package/

# 计算包的SHA-256
– sha256sum firmware–${FIRMWARE_VERSION}.tar.gz > checksum.txt

# 生成发布说明
– |
cat > RELEASE_NOTES.md << EOF
# 固件版本 ${FIRMWARE_VERSION}

**构建时间**: $(date –u)
**Git Commit**: ${CI_COMMIT_SHA}
**触发分支**: ${CI_COMMIT_BRANCH}

## 包含变更
${CI_COMMIT_MESSAGE}

## 支持芯片
– ESP32
– ESP32–S3
– ESP32–C3

## 校验和
$(cat checksum.txt)
EOF

artifacts:
paths:
– firmware–*.tar.gz
– RELEASE_NOTES.md
– checksum.txt
expire_in: 90 days

publish:ota:
stage: publish
image: python:3.9–slim
needs:
– job: package:firmware
artifacts: true
– job: unit–test:host
rules:
– if: $CI_COMMIT_BRANCH == "main"
script:
– pip3 install requests

# 解压固件包
– tar –xzf firmware–*.tar.gz –C /tmp/

# 签名固件
– python3 scripts/sign_firmware.py
––firmware /tmp/release_package/app_esp32.bin
––key ${FIRMWARE_SIGNING_KEY}
––output /tmp/signed/

# 发布到OTA服务器
– python3 scripts/publish_ota.py
––firmware /tmp/signed/app–signed.bin
––version ${FIRMWARE_VERSION}
––target esp32
––rollout 5

# 通知物联网平台
– |
curl -X POST "${IOT_PLATFORM_URL}/api/ota/notify" \\
-H "Authorization: Bearer ${IOT_PLATFORM_TOKEN}" \\
-H "Content-Type: application/json" \\
-d "{\\"version\\":\\"${FIRMWARE_VERSION}\\",\\"rollout\\":5}"

– echo "OTA发布完成，灰度比例: 5%"

# 自动扩大灰度（需手动确认）
scale:rollout:
stage: publish
rules:
– if: $CI_COMMIT_BRANCH == "main"
when: manual
script:
– echo "扩大灰度发布比例到 ${ROLLOUT_PERCENT}%"
– python3 scripts/rollout.py increase ${FIRMWARE_VERSION} ${ROLLOUT_PERCENT}
variables:
ROLLOUT_PERCENT: "50"

八、监控与告警

8.1 OTA 升级监控指标

# scripts/ota_monitor.py
"""OTA 升级监控服务"""
import time
import json
from collections import defaultdict

class OTAMonitor:
def __init__(self):
self.stats = defaultdict(lambda: {
'total': 0,
'success': 0,
'failed': 0,
'rolled_back': 0,
'in_progress': 0
})
self.alerts = []

def record_event(self, device_id, version, event_type):
"""记录OTA事件"""
stat = self.stats[version]
stat['total'] += 1

if event_type == 'download_started':
stat['in_progress'] += 1
elif event_type == 'upgrade_success':
stat['success'] += 1
stat['in_progress'] -= 1
elif event_type == 'upgrade_failed':
stat['failed'] += 1
stat['in_progress'] -= 1
self._check_failure_rate(version)
elif event_type == 'rollback':
stat['rolled_back'] += 1
self._trigger_alert(version, device_id, "固件回滚")

def _check_failure_rate(self, version):
"""检查失败率是否超过阈值"""
stat = self.stats[version]
if stat['total'] > 10:
failure_rate = (stat['failed'] + stat['rolled_back']) / stat['total']
if failure_rate > 0.05: # 超过5%失败率
self._trigger_alert(version, None,
f"失败率 {failure_rate:.1%} 超过阈值 5%")

def _trigger_alert(self, version, device_id, message):
"""触发告警"""
alert = {
'version': version,
'device_id': device_id,
'message': message,
'timestamp': time.time()
}
self.alerts.append(alert)
print(f"[ALERT] {message} (version={version}, device={device_id})")

def get_dashboard_data(self):
"""获取仪表盘数据"""
return {
'versions': dict(self.stats),
'alerts': self.alerts[–10:] # 最近10条告警
}

8.2 设备端健康上报

// 设备端定期上报状态到物联网平台
void report_device_health(void)
{
const firmware_info_t *fw = get_firmware_info();

cJSON *report = cJSON_CreateObject();
cJSON_AddStringToObject(report, "device_id", get_device_id_str());
cJSON_AddStringToObject(report, "firmware_version", fw->version);
cJSON_AddStringToObject(report, "commit", fw->commit_hash);

// 系统状态
cJSON *sys = cJSON_CreateObject();
cJSON_AddNumberToObject(sys, "uptime_s",
esp_timer_get_time() / 1000000);
cJSON_AddNumberToObject(sys, "free_heap",
esp_get_free_heap_size());
cJSON_AddNumberToObject(sys, "min_heap",
esp_get_minimum_free_heap_size());
cJSON_AddNumberToObject(sys, "reset_reason",
esp_reset_reason());
cJSON_AddItemToObject(report, "system", sys);

// OTA状态
esp_ota_img_states_t state;
const esp_partition_t *running = esp_ota_get_running_partition();
esp_ota_get_partition_state(running, &state);
cJSON_AddNumberToObject(report, "ota_partition_state", state);

// 发送
char *json_str = cJSON_Print(report);
send_to_iot_platform("/api/device/health", json_str);
free(json_str);
cJSON_Delete(report);
}

九、产品推荐与工具链

沧州虎王科技在 ESP32 固件 CI/CD 建设过程中，深度使用以下工具链：

• ESP32工具箱V2.0：集成固件烧录、串口监控、Flash 分区可视化、OTA 测试模拟等功能。在 CI/CD 流水线的硬件在环测试阶段，可以用它快速批量烧录和验证设备。支持自动检测设备串口、一键回滚固件和 OTA 分区状态可视化。

• 随身WiFi硬件调试工具（hardware.czkree.com）：在无有线网络环境下进行 OTA 测试的利器。可以搭建临时 WiFi 热点连接 ESP32 设备，同时通过手机端界面实时监控串口输出和 OTA 进度，非常适合现场部署调试。

• 物联网平台：提供完整的 OTA 升级管理后台，支持固件版本管理、灰度发布策略、设备分组推送、升级状态实时监控和失败告警。CI/CD 流水线通过 REST API 与平台对接，实现从代码提交到设备升级的全自动化。

总结

ESP32 固件 CI/CD 流水线的建设，是将嵌入式开发从手工模式升级为工业化生产的关键一步。本文覆盖了完整链路：

关键经验：先建立测试体系，再实施 OTA，最后才做灰度发布。没有充分的自动化测试，OTA 就是赌博；没有灰度发布，全量推送就是冒险。沧州虎王科技团队按此路径在多个量产项目中实现了零变砖升级，值得推广。

作者：沧州虎王科技技术团队
标签：物联网、嵌入式、ESP32
产品推荐：ESP32工具箱V2.0 | 随身WiFi硬件调试工具(hardware.czkree.com) | 物联网平台