1. OTA升级基础架构设计

任务目标

设计一个面向IoT终端的OTA升级基础架构，覆盖设备注册、版本管理、升级任务下发与状态上报四个核心环节。你需要明确每个模块的职责边界、模块间的交互协议，并落地一个可运行的最小闭环验证系统。

架构总览

系统采用服务端-设备端两级结构，服务端由三个独立模块组成，设备端通过MQTT协议与服务端通信。整体交互流程如下：

交互流程简述：设备首次连接时调用注册接口登记自身信息（步骤1）；管理员通过版本管理模块上传新固件（步骤2）；任务下发模块根据设备当前版本与目标版本生成升级任务（步骤3）；设备执行升级过程中持续上报状态，状态上报模块负责记录并驱动后续动作（步骤4）。

环境依赖

Python 3.8+，安装 flask、paho-mqtt、requests
MQTT Broker（推荐 mosquitto，默认 localhost:1883）

完整实现

以下代码包含服务端API（Flask）与设备端模拟逻辑，保存为 ota_system.py 后可直接运行。服务端使用SQLite存储数据，通过MQTT向设备下发升级指令。

import sqlite3, json, threading, time, hashlib
from flask import Flask, request, jsonify
import paho.mqtt.client as mqtt

# ---------- 服务端 ----------
app = Flask(__name__)
DB_PATH = "ota.db"
MQTT_BROKER = "localhost"
MQTT_PORT = 1883
mqtt_client = mqtt.Client()

def init_db():
    conn = sqlite3.connect(DB_PATH)
    c = conn.cursor()
    c.execute("CREATE TABLE IF NOT EXISTS devices (device_id TEXT PRIMARY KEY, current_version TEXT)")
    c.execute("CREATE TABLE IF NOT EXISTS versions (version TEXT PRIMARY KEY, file_hash TEXT, size INTEGER)")
    c.execute("CREATE TABLE IF NOT EXISTS tasks (task_id TEXT PRIMARY KEY, device_id TEXT, target_version TEXT, status TEXT)")
    conn.commit(); conn.close()

def db_exec(sql, args=()):
    conn = sqlite3.connect(DB_PATH)
    c = conn.cursor()
    c.execute(sql, args)
    conn.commit()
    result = c.fetchall()
    conn.close()
    return result

# 1. 设备注册
@app.route("/register", methods=["POST"])
def register():
    data = request.get_json()
    device_id = data["device_id"]
    current_version = data.get("current_version", "v1.0.0")
    db_exec("INSERT OR REPLACE INTO devices VALUES (?, ?)", (device_id, current_version))
    return jsonify({"status": "registered", "device_id": device_id})

# 2. 版本登记
@app.route("/version", methods=["POST"])
def add_version():
    data = request.get_json()
    version = data["version"]
    file_hash = hashlib.sha256(version.encode()).hexdigest()[:16]
    size = len(version) * 10  # 模拟固件大小
    db_exec("INSERT OR REPLACE INTO versions VALUES (?, ?, ?)", (version, file_hash, size))
    return jsonify({"status": "version_added", "version": version, "hash": file_hash})

# 3. 创建升级任务并下发
@app.route("/upgrade", methods=["POST"])
def create_task():
    data = request.get_json()
    device_id = data["device_id"]
    target_version = data["target_version"]
    task_id = f"task_{int(time.time())}"
    db_exec("INSERT INTO tasks VALUES (?, ?, ?, ?)", (task_id, device_id, target_version, "pending"))
    
    # 查询设备当前版本
    dev = db_exec("SELECT current_version FROM devices WHERE device_id=?", (device_id,))
    if not dev:
        return jsonify({"error": "device not found"}), 404
    current_version = dev[0][0]
    
    # 通过MQTT下发升级指令
    payload = json.dumps({"task_id": task_id, "target_version": target_version})
    mqtt_client.publish(f"ota/{device_id}/command", payload)
    return jsonify({"task_id": task_id, "current_version": current_version, "target_version": target_version})

# 4. 状态上报接收
@app.route("/status", methods=["POST"])
def receive_status():
    data = request.get_json()
    task_id = data["task_id"]
    status = data["status"]  # downloading / verifying / activating / success / failed
    db_exec("UPDATE tasks SET status=? WHERE task_id=?", (status, task_id))
    return jsonify({"received": True})

# ---------- 设备端模拟 ----------
class OtaDevice:
    def __init__(self, device_id, current_version):
        self.device_id = device_id
        self.current_version = current_version
        self.mqtt = mqtt.Client()
        self.mqtt.on_message = self.on_message
        self.mqtt.connect(MQTT_BROKER, MQTT_PORT)
        self.mqtt.subscribe(f"ota/{device_id}/command")
        self.mqtt.loop_start()
        # 注册
        requests.post("http://localhost:5000/register", json={"device_id": device_id, "current_version": current_version})

    def on_message(self, client, userdata, msg):
        cmd = json.loads(msg.payload)
        task_id = cmd["task_id"]
        target = cmd["target_version"]
        # 模拟升级流程
        self.report(task_id, "downloading")
        time.sleep(0.5)
        self.report(task_id, "verifying")
        time.sleep(0.3)
        self.current_version = target
        self.report(task_id, "success")
        print(f"[设备 {self.device_id}] 升级完成 → {target}")

    def report(self, task_id, status):
        requests.post("http://localhost:5000/status", json={"task_id": task_id, "status": status})

# ---------- 启动 ----------
if __name__ == "__main__":
    init_db()
    mqtt_client.connect(MQTT_BROKER, MQTT_PORT)
    mqtt_client.loop_start()
    # 启动服务端线程
    threading.Thread(target=lambda: app.run(port=5000, debug=False, use_reloader=False), daemon=True).start()
    time.sleep(1)
    
    # 模拟设备
    dev = OtaDevice("dev_001", "v1.0.0")
    
    # 模拟管理员操作
    requests.post("http://localhost:5000/version", json={"version": "v2.0.0"})
    resp = requests.post("http://localhost:5000/upgrade", json={"device_id": "dev_001", "target_version": "v2.0.0"})
    print("下发任务:", resp.json())
    
    time.sleep(3)  # 等待设备完成升级
    # 查询最终状态
    tasks = db_exec("SELECT * FROM tasks")
    print("任务记录:", tasks)
    dev.mqtx.loop_stop()

运行与验证

启动MQTT Broker：mosquitto -v（确保监听1883端口）
运行系统：python ota_system.py
预期输出：程序自动完成设备注册→版本登记→任务下发→设备模拟升级→状态上报全流程，控制台打印设备升级完成信息，最终查询任务记录显示状态为 success

通过条件：终端出现 [设备 dev_001] 升级完成 → v2.0.0，且任务记录中对应task状态为 success。若需验证多设备场景，可创建多个 OtaDevice 实例并分别下发任务。

常见问题

MQTT连接失败：确认mosquitto已启动且端口未被占用
端口5000被占用：修改 app.run 中的端口参数
设备收不到指令：检查topic是否匹配 ota/{device_id}/command

📚 课程目录