第一章：为什么需要边缘智能？
1.1 云端 AI 的局限
| 问题 | 说明 |
|---|---|
| 高延迟 | 视频分析往返 >1s，无法用于实时控制 |
| 带宽瓶颈 | 1000 台摄像头 × 2Mbps = 2Gbps 上行带宽 |
| 隐私泄露 | 家庭监控视频上传第三方服务器 |
| 单点故障 | 云服务宕机 → 全系统瘫痪 |
1.2 边缘智能的优势
- 实时性：本地推理 <50ms
- 带宽节省：仅上传元数据（如'异常概率=0.92'）
- 离线可用：网络中断仍可工作
- 合规友好：原始数据不出设备（GDPR/CCPA）
典型架构演进： Cloud-only → Cloud + Edge → Edge-first, Cloud-assisted
第二章：边缘硬件选型与开发环境
2.1 主流边缘芯片对比
| 芯片 | RAM | Flash | AI 加速 | 适用场景 |
|---|---|---|---|---|
| ESP32 | 520KB | 4MB+ | 无 | 低功耗传感器节点 |
| Raspberry Pi Pico | 264KB | 2MB | 无 | 教育/原型 |
| Coral Edge TPU | 1GB | 8GB | Google Edge TPU | 视频分析 |
| NVIDIA Jetson Nano | 4GB | 16GB | CUDA GPU | 复杂 CV/NLP |
本篇聚焦：ESP32（成本 <$5）+ Raspberry Pi（作为网关）
2.2 开发环境搭建
# 安装 MicroPython 固件（ESP32）
esptool.py --port /dev/ttyUSB0 erase_flash
esptool.py --port /dev/ttyUSB0 --baud 460800 write_flash -z 0 esp32-idf4-20230426-v1.20.0.bin
# 安装 uPyLoader（文件传输工具）
pip install upyloader
IDE 推荐：Thonny（内置 MicroPython 支持）
第三章：TinyML 模型开发与部署
3.1 模型压缩四步法
- 剪枝：移除冗余权重（TensorFlow Model Optimization Toolkit）
- 量化：FP32 → INT8（减少 75% 体积）
- 知识蒸馏：大模型 → 小模型
- 算子裁剪：仅保留必要 OP（如去掉
tf.image.resize）
3.2 转换为 TFLite Micro
# convert_to_tflite.py
import tensorflow as tf
# 加载 Keras 模型
model = tf.keras.models.load_model('vibration_cnn.h5')
# 量化转换
converter = tf.lite.TFLiteConverter.from_keras_model(model)
converter.optimizations = [tf.lite.Optimize.DEFAULT]
converter.target_spec.supported_types = [tf.int8]
converter.inference_input_type = tf.int8
converter.inference_output_type = tf.int8
tflite_model = converter.convert()
# 保存为 C 数组（供 MicroPython 加载）
with open('model.cc', 'w') as f:
f.write('const unsigned char model_data[] = {\n')
f.write(','.join([str(b) for b in tflite_model]))
f.write('\n};\n')
注意：MicroPython 无法直接加载
.tflite，需通过 C 扩展或预编译固件。
3.3 替代方案：ONNX Runtime for Micro
- 优势：支持更多算子，Python API 更友好
- 步骤：
# 导出 ONNX
torch.onnx.export(model, dummy_input, "model.onnx")
# 转换为 ORT 格式（micro 版本）
from onnxruntime import InferenceSession
# （需交叉编译 C++ 库嵌入固件）
简化策略：初期使用 纯 Python 微模型（如决策树、小型 MLP）
第四章：边缘端实现（MicroPython）
4.1 传感器数据采集（I2C）
# sensors/vibration.py
from machine import I2C, Pin
import struct
class ADXL345:
def __init__(self, i2c):
self.i2c = i2c
self.addr = 0x53
# 初始化寄存器
self.i2c.writeto_mem(self.addr, 0x2D, b'\x08') # 测量模式
def read_xyz(self):
data = self.i2c.readfrom_mem(self.addr, 0x32, 6)
x, y, z = struct.unpack('<hhh', data)
return x/256, y/256, z/256 # 转换为 g 单位
4.2 本地推理（简化版 MLP）
# models/tiny_mlp.py
import math
def inference(features: list) -> float:
"""3 层 MLP，参数硬编码（从训练导出）"""
w1 = [[0.12, -0.34], [-0.56, 0.78]] # 权重矩阵（INT8 量化后）
b1 = [0.1, -0.2]
w2 = [0.9, -0.8]
b2 = 0.05
# 隐藏层
h = [max(0, sum(w*f for w,f in zip(w_row, features)) + b) for w_row, b in zip(w1, b1)]
# 输出层
output = sum(w*h_val for w, h_val in zip(w2, h)) + b2
return 1 / (1 + math.exp(-output)) # Sigmoid
4.3 事件驱动上传
# main.py
from umqtt.simple import MQTTClient
import network
import time
# 连接 WiFi
sta = network.WLAN(network.STA_IF)
sta.active(True)
sta.connect('SSID', 'PASSWORD')
# 初始化 sensor
sensor = ADXL345(I2C(scl=Pin(22), sda=Pin(21)))
mqtt = MQTTClient('esp32_01', 'mqtt.your-cloud.com', ssl=True)
while True:
x, y, z = sensor.read_xyz()
anomaly_score = inference([x, y, z])
if anomaly_score > 0.8: # 仅异常时上传
mqtt.publish('factory/device_01/alert', '{"score":%.2f}' % anomaly_score)
time.sleep(1) # 低功耗：可改为 deep sleep
内存优化：禁用 GC，预分配缓冲区。
第五章：边缘 - 云通信安全
5.1 MQTT over TLS
- 证书管理：
# 生成客户端证书（OpenSSL）
openssl req -x509 -newkey rsa:2048 -keyout client.key -out client.crt -days 365 -nodes
# MicroPython 加载
with open('client.crt', 'rb') as f:
cert = f.read()
mqtt = MQTTClient(..., ssl=True, ssl_params={'cert': cert})
5.2 CoAP（低功耗替代）
- 适用场景：电池供电设备（如农业传感器）
- MicroPython 库：
aiocoap（需 asyncio 支持）
# coap_client.py
import aiocoap
async def send_coap(data):
protocol = await aiocoap.Context.create_client_context()
request = aiocoap.Message(code=aiocoap.POST, payload=data.encode(), uri='coap://cloud:5683/sensor')
response = await protocol.request(request).response
await protocol.shutdown()
协议选择：实时性要求高 → MQTT；超低功耗 → CoAP + DTLS
第六章：云端管理平台（Flask）
6.1 设备注册与状态监控
# models/edge_device.py
from mongoengine import Document, StringField, DateTimeField
class EdgeDevice(Document):
device_id = StringField(unique=True) # 如 "esp32_01"
last_seen = DateTimeField()
firmware_version = StringField()
model_version = StringField() # 当前运行的 AI 模型版本
status = StringField(choices=['online', 'offline', 'error'])
6.2 MQTT 消息处理
# services/mqtt_handler.py
import paho.mqtt.client as mqtt
import json
from datetime import datetime
def on_message(client, userdata, msg):
payload = json.loads(msg.payload)
device_id = msg.topic.split('/')[1] # 更新设备状态
EdgeDevice.objects(device_id=device_id).update(
set__last_seen=datetime.utcnow(),
set__status='online'
)
# 处理告警
if 'alert' in msg.topic:
create_alert(device_id, payload['score'])
# 启动后台 MQTT 客户端
background_mqtt = mqtt.Client()
background_mqtt.on_message = on_message
background_mqtt.connect('localhost', 1883)
background_mqtt.subscribe('factory/+/alert')
background_mqtt.loop_start()
6.3 OTA 模型更新
# routes/ota.py
from flask import Blueprint, request, jsonify
import bsdiff4
app = Blueprint('ota', __name__)
@app.route('/ota/model/<device_id>', methods=['POST'])
def push_model_update(device_id: str):
# 1. 生成差分更新包（bsdiff）
old_model = get_current_model(device_id)
new_model = request.files['model']
diff = bsdiff4.file_diff(old_model, new_model)
# 2. 通过 MQTT 发送
mqtt.publish(f'ota/{device_id}/model_diff', diff)
# 3. 记录任务
OTAJob(device_id=device_id, type='model', status='pending').save()
return jsonify({"status": "update_queued"})
边缘端 OTA 流程：接收差分包合并到 Flash 分区 B → 重启后从 B 启动 → 验证成功 → 标记 B 为主分区
第七章：场景实战
7.1 工业预测性维护
- 硬件：ESP32 + ADXL345 振动传感器
- 模型：1D-CNN（输入：512 点振动频谱）
效果：
本地检测轴承故障（准确率 92%） 仅上传异常事件，带宽减少 99%
7.2 智慧农业
- 传感器：土壤湿度 + 温度 + 光照
- 决策逻辑：
if soil_moisture < 30 and sunlight > 500:
activate_pump(duration=10) # 本地控制继电器
mqtt.publish('farm/pump', 'activated')
- 优势：断网时仍可自动灌溉
7.3 零售客流统计
- 硬件：ESP32-CAM（200 万像素）
- 合规性：符合 GDPR '数据最小化'原则
隐私保护流程：
拍摄图像运行人脸检测（Tiny YOLO） 立即丢弃原始图，仅保留人脸框坐标 上传'客流数=5'
第八章：性能与功耗优化
8.1 内存管理
- 禁用动态分配：预分配所有缓冲区
- 使用 memoryview：避免复制大数组
# 预分配振动数据缓冲区
VIB_BUFFER = bytearray(1024)
mv = memoryview(VIB_BUFFER)
i2c.readfrom_into(addr, mv[0:6]) # 直接写入
8.2 低功耗模式
- 深度睡眠：ESP32 电流 <10μA
- 唤醒源：定时器 or 外部中断（如振动超阈值）
# 每 10 分钟唤醒一次
import machine
rtc = machine.RTC()
rtc.irq(trigger=rtc.ALARM0, wake=machine.DEEPSLEEP)
rtc.alarm(rtc.ALARM0, 600000) # 10 分钟
machine.deepsleep()
第九章：安全加固
9.1 安全启动
- 原理：验证固件签名后再执行
- ESP32 支持：eFuses 存储公钥哈希
9.2 安全存储
- 敏感数据（如 MQTT 密码）加密存储：
# 使用设备唯一 ID 作为密钥
import esp32
key = esp32.unique_id()
encrypted = aes_encrypt(password, key)
第十章：规模化管理
10.1 千级设备分组
批量操作：
向
factory_A所有设备推送新模型
标签系统：
location: factory_Atype: vibration_sensor
10.2 边缘集群（Raspberry Pi 网关）
优势：
减少云连接数（1000 ESP32 → 10 Pi） 本地协同推理（如多传感器融合）
架构：
ESP32 → 本地 Pi 网关（汇总数据） → 云
总结：智能下沉，价值上升
边缘智能不是把云缩小，而是重新思考智能的分布。
