1. OTA基础架构搭建
任务目标
搭建一个可独立运行的OTA客户端基础框架，包含网络模块、固件下载模块、升级模块与状态管理模块。框架以命令行程序形式演示一次完整的OTA升级流程：检查更新、下载固件、校验固件、执行升级、更新状态。
环境依赖
- Python 3.8+（仅标准库，无第三方依赖）
- 操作系统：Windows / Linux / macOS 均可
核心实现
以下代码实现了一个最小但完整的OTA客户端框架。四个模块职责清晰：NetworkModule模拟从远端获取版本信息与固件数据；DownloadModule负责下载并校验固件完整性；UpgradeModule执行固件替换；StateManager维护升级状态机。主流程串联各模块完成一次升级闭环。
import hashlib
import json
import os
import shutil
from enum import Enum, auto
class UpgradeState(Enum):
IDLE = auto()
CHECKING = auto()
DOWNLOADING = auto()
VERIFYING = auto()
UPGRADING = auto()
COMPLETED = auto()
FAILED = auto()
class StateManager:
"""状态管理模块：维护升级状态机并支持持久化"""
def __init__(self, state_file="ota_state.json"):
self.state_file = state_file
self.state = UpgradeState.IDLE
self.current_version = "1.0.0"
self._load()
def _load(self):
if os.path.exists(self.state_file):
with open(self.state_file, "r") as f:
data = json.load(f)
self.state = UpgradeState[data["state"]]
self.current_version = data["current_version"]
def _save(self):
with open(self.state_file, "w") as f:
json.dump({
"state": self.state.name,
"current_version": self.current_version
}, f)
def set_state(self, new_state):
self.state = new_state
self._save()
def update_version(self, version):
self.current_version = version
self._save()
class NetworkModule:
"""网络模块：模拟OTA服务器接口"""
def __init__(self, server_dir="server_data"):
self.server_dir = server_dir
os.makedirs(server_dir, exist_ok=True)
def check_for_update(self, current_version):
"""检查远端是否有新版本，返回版本号或None"""
version_file = os.path.join(self.server_dir, "version.json")
if not os.path.exists(version_file):
return None
with open(version_file, "r") as f:
remote_version = json.load(f)["version"]
return remote_version if remote_version != current_version else None
def download_firmware(self, local_path):
"""模拟从服务器下载固件到本地"""
remote_fw = os.path.join(self.server_dir, "firmware.bin")
if not os.path.exists(remote_fw):
raise FileNotFoundError("服务器上不存在固件文件")
shutil.copy(remote_fw, local_path)
return local_path
class DownloadModule:
"""固件下载模块：负责下载与完整性校验"""
def __init__(self, network, download_dir="downloads"):
self.network = network
self.download_dir = download_dir
os.makedirs(download_dir, exist_ok=True)
def download_and_verify(self, expected_sha256):
"""下载固件并校验SHA256，返回本地路径或抛出异常"""
local_path = os.path.join(self.download_dir, "firmware.bin")
self.network.download_firmware(local_path)
sha256 = hashlib.sha256()
with open(local_path, "rb") as f:
for chunk in iter(lambda: f.read(4096), b""):
sha256.update(chunk)
actual_hash = sha256.hexdigest()
if actual_hash != expected_sha256:
raise ValueError(f"SHA256校验失败: 期望{expected_sha256}, 实际{actual_hash}")
return local_path
class UpgradeModule:
"""升级模块：执行固件替换"""
def __init__(self, firmware_dir="firmware"):
self.firmware_dir = firmware_dir
os.makedirs(firmware_dir, exist_ok=True)
def apply_upgrade(self, downloaded_fw, new_version):
"""将下载的固件安装到正式位置"""
target_path = os.path.join(self.firmware_dir, "current.bin")
shutil.copy(downloaded_fw, target_path)
return target_path
class OTAClient:
"""OTA客户端主控：编排各模块完成升级流程"""
def __init__(self):
self.state_mgr = StateManager()
self.network = NetworkModule()
self.downloader = DownloadModule(self.network)
self.upgrader = UpgradeModule()
def run_upgrade_check(self):
"""执行一次完整的OTA升级流程"""
try:
# 1. 检查更新
self.state_mgr.set_state(UpgradeState.CHECKING)
new_version = self.network.check_for_update(self.state_mgr.current_version)
if new_version is None:
print("已是最新版本，无需升级")
self.state_mgr.set_state(UpgradeState.IDLE)
return False
print(f"发现新版本: {new_version}")
# 2. 获取固件校验值（模拟从服务器获取）
hash_file = os.path.join(self.network.server_dir, "firmware.sha256")
with open(hash_file, "r") as f:
expected_sha256 = f.read().strip()
# 3. 下载并校验
self.state_mgr.set_state(UpgradeState.DOWNLOADING)
local_fw = self.downloader.download_and_verify(expected_sha256)
self.state_mgr.set_state(UpgradeState.VERIFYING)
print("固件下载并校验通过")
# 4. 执行升级
self.state_mgr.set_state(UpgradeState.UPGRADING)
self.upgrader.apply_upgrade(local_fw, new_version)
# 5. 更新状态
self.state_mgr.update_version(new_version)
self.state_mgr.set_state(UpgradeState.COMPLETED)
print(f"升级完成，当前版本: {new_version}")
return True
except Exception as e:
self.state_mgr.set_state(UpgradeState.FAILED)
print(f"升级失败: {e}")
return False
def setup_demo_server():
"""构造模拟OTA服务器数据（仅用于演示）"""
server_dir = "server_data"
os.makedirs(server_dir, exist_ok=True)
# 版本信息
with open(os.path.join(server_dir, "version.json"), "w") as f:
json.dump({"version": "2.0.0"}, f)
# 固件内容（模拟新固件）
firmware_content = b"FAKE_FIRMWARE_V2.0.0_CONTENT"
with open(os.path.join(server_dir, "firmware.bin"), "wb") as f:
f.write(firmware_content)
# SHA256校验值
sha256 = hashlib.sha256(firmware_content).hexdigest()
with open(os.path.join(server_dir, "firmware.sha256"), "w") as f:
f.write(sha256)
if __name__ == "__main__":
# 首次运行先构造演示数据
if not os.path.exists("server_data"):
setup_demo_server()
client = OTAClient()
client.run_upgrade_check()
运行与验证
- 将上述代码保存为
ota_client.py。 - 在终端执行：
python ota_client.py - 预期输出（首次运行）：
发现新版本: 2.0.0
固件下载并校验通过
升级完成，当前版本: 2.0.0
- 再次执行
python ota_client.py，预期输出：已是最新版本，无需升级。这是因为状态文件ota_state.json已记录当前版本为 2.0.0。
验证通过条件：两次运行输出与上述预期一致，且目录下生成 ota_state.json（状态持久化）、downloads/firmware.bin（下载的固件）、firmware/current.bin（安装后的固件）三个文件。
关键说明
- 网络模块使用本地文件系统模拟服务器，实际项目中替换为HTTP/HTTPS请求即可，接口签名不变。
- 状态管理模块通过JSON文件持久化，重启后仍能恢复上次升级状态。
- SHA256校验确保固件在传输过程中未被篡改，这是OTA升级的安全基线。