zhou yang和Claude Opus 5.5
271c649f66
- firmware.py：固件库。上传时核 ESP 镜像头（0xE9 + app 描述 + project_name == phonerail + ≤3 MB），
存成 /root/phonerail/fw/<sha 前 16 位>.bin（nginx 7765 放出去），库里记版本 / 编译时间 / 上传人
- api：GET/POST/DELETE /api/firmware（请求体即 .bin，不引 multipart）、POST /api/boards/{id}/ota（进审计）；
通用 /cmd 不放行 ota —— URL 与哈希只能由固件库生成；OTA 事件按 arg 显示中文、失败 / 回滚标红
- 前端：Firmware.vue + OtaState.vue；推送对话框默认不重启，勾「立刻重启」时列出每块板会断电的路并二次确认；
单板页显示 OTA 状态
- simboard 模拟 ota（下载进度 / 就绪 / 重启 / 自证）；test_firmware.py 13 例，后端共 39 例（本机 + 服务器 3.8）
- service 加 --fw-dir；已部署。板 B 实测：上传 → 推送 → 就绪 → 同镜像带重启 → 不重下直接重启 → 自证健康
Co-Authored-By: Claude Opus 5.5 <[email protected]>
127 行
5.6 KiB
Python
127 行
5.6 KiB
Python
"""固件库 + OTA 命令（板侧 C2，firmware/main/ota.h；需求 §10.8「C2 定稿的 OTA」）。
|
||
|
||
流程：网页上传 .bin → 这里解析 ESP 镜像头、算 sha256、存成 `<fw_dir>/<sha 前 16 位>.bin` →
|
||
nginx 7765（rack/deploy/nginx-phonerail-fw.conf，**不过门户**）把它按这个名字放出去 →
|
||
`ota` 命令经 MQTT/TLS 带 url + sha256 + size，板子自己下载、哈希对上才切分区。
|
||
|
||
服务器这一侧先把能拦的拦掉，免得一次推送在板子上下完 1 MB 才报错：
|
||
· 不是 ESP 镜像（首字节 0xE9）/ 没有 app 描述（0xABCD5432）→ 拒；
|
||
· project_name 不是 `phonerail`（传错了别的项目的 .bin）→ 拒（板子那头也会核一遍）；
|
||
· 大于一个槽（3 MB）→ 拒。
|
||
**板子的校验仍是唯一权威**，这里只是早点告诉操作者。
|
||
|
||
⚠️ `reboot=True` = 板子校验通过后立刻复位 = **五台手机硬断电一次**（EN 在复位时被下拉）。
|
||
界面必须二次确认；这里不替人做这个决定。
|
||
"""
|
||
import hashlib
|
||
import os
|
||
import re
|
||
import struct
|
||
from typing import List, Optional
|
||
|
||
from .db import Db, now_ms
|
||
|
||
SLOT_BYTES = 0x300000 # firmware/main/ota.h OTA_SLOT_BYTES
|
||
URL_MAX = 64 # OTA_URL_MAX（命令 JSON 字符串上限）
|
||
PROJECT = "phonerail"
|
||
SHA_RE = re.compile(r"^[0-9a-f]{64}$")
|
||
|
||
# esp_app_desc_t 紧跟在镜像头（24 B）+ 第一个段头（8 B）之后
|
||
_DESC = 32
|
||
_DESC_MAGIC = 0xABCD5432
|
||
|
||
class FwError(ValueError):
|
||
"""镜像不能收 / 推送请求不对（中文说明）。"""
|
||
|
||
|
||
def _cstr(b: bytes) -> str:
|
||
return b.split(b"\0", 1)[0].decode("ascii", "replace")
|
||
|
||
|
||
def parse_image(data: bytes) -> dict:
|
||
"""ESP32 app 镜像 → {version, project, built, idf}。不是本工程的 app 镜像抛 FwError。"""
|
||
if not 0 < len(data) <= SLOT_BYTES:
|
||
raise FwError("镜像大小 %d 字节，必须在 1~%d 之间（一个 OTA 槽）" % (len(data), SLOT_BYTES))
|
||
if len(data) < _DESC + 256 or data[0] != 0xE9:
|
||
raise FwError("不是 ESP 固件镜像（首字节应为 0xE9）：要传 build/phonerail.bin，不是 .elf / 合并镜像")
|
||
magic = struct.unpack_from("<I", data, _DESC)[0]
|
||
if magic != _DESC_MAGIC:
|
||
raise FwError("镜像里没有 app 描述（可能是 bootloader 或分区表）")
|
||
info = {
|
||
"version": _cstr(data[_DESC + 16:_DESC + 48]),
|
||
"project": _cstr(data[_DESC + 48:_DESC + 80]),
|
||
"built": (_cstr(data[_DESC + 96:_DESC + 112]) + " " + _cstr(data[_DESC + 80:_DESC + 96])).strip(),
|
||
"idf": _cstr(data[_DESC + 112:_DESC + 144]),
|
||
}
|
||
if info["project"] != PROJECT:
|
||
raise FwError("这是「%s」的固件，不是 %s" % (info["project"], PROJECT))
|
||
return info
|
||
|
||
|
||
class FwStore:
|
||
def __init__(self, db: Db, fw_dir: str, base_url: str):
|
||
self.db = db
|
||
self.dir = fw_dir
|
||
self.base = base_url.rstrip("/")
|
||
if len(self.url_for("0" * 64)) > URL_MAX:
|
||
raise ValueError("--fw-base 太长：URL 会超过板子的 %d 字节上限" % URL_MAX)
|
||
os.makedirs(self.dir, exist_ok=True)
|
||
|
||
@staticmethod
|
||
def name_for(sha: str) -> str:
|
||
return sha[:16] + ".bin" # nginx 7765 只放行这种名字
|
||
|
||
def url_for(self, sha: str) -> str:
|
||
return "%s/%s" % (self.base, self.name_for(sha))
|
||
|
||
def add(self, data: bytes, filename: str, user: str) -> dict:
|
||
info = parse_image(data)
|
||
sha = hashlib.sha256(data).hexdigest()
|
||
path = os.path.join(self.dir, self.name_for(sha))
|
||
tmp = os.path.join(self.dir, "." + self.name_for(sha) + ".tmp")
|
||
with open(tmp, "wb") as f:
|
||
f.write(data)
|
||
os.chmod(tmp, 0o644) # nginx（www-data）要读得到
|
||
os.replace(tmp, path)
|
||
self.db.x("INSERT OR IGNORE INTO firmware (sha256, version, project, built, idf, size, filename, "
|
||
"uploaded_by, uploaded) VALUES (?,?,?,?,?,?,?,?,?)",
|
||
(sha, info["version"], info["project"], info["built"], info["idf"], len(data),
|
||
re.split(r"[\\/]", filename or "")[-1][:128], user, now_ms())) # 两种分隔符都切：服务器是 Linux
|
||
return self.get(sha)
|
||
|
||
def get(self, sha: str) -> Optional[dict]:
|
||
rows = self.db.q("SELECT * FROM firmware WHERE sha256=?", (sha,))
|
||
if not rows:
|
||
return None
|
||
d = dict(rows[0])
|
||
d["url"] = self.url_for(sha)
|
||
d["present"] = os.path.isfile(os.path.join(self.dir, self.name_for(sha)))
|
||
return d
|
||
|
||
def list(self) -> List[dict]:
|
||
return [self.get(r["sha256"]) for r in
|
||
self.db.q("SELECT sha256 FROM firmware ORDER BY uploaded DESC")]
|
||
|
||
def delete(self, sha: str) -> bool:
|
||
if self.get(sha) is None:
|
||
return False
|
||
try:
|
||
os.remove(os.path.join(self.dir, self.name_for(sha)))
|
||
except FileNotFoundError:
|
||
pass
|
||
self.db.x("DELETE FROM firmware WHERE sha256=?", (sha,))
|
||
return True
|
||
|
||
def command(self, sha: str, reboot) -> dict:
|
||
"""推送请求 → 发给板子的 `ota` 命令（不含 seq）。"""
|
||
if not isinstance(sha, str) or not SHA_RE.match(sha):
|
||
raise FwError("sha256 必须是 64 位小写十六进制")
|
||
if not isinstance(reboot, bool):
|
||
raise FwError("reboot 必须是 true/false")
|
||
fw = self.get(sha)
|
||
if fw is None:
|
||
raise FwError("固件库里没有这个镜像")
|
||
if not fw["present"]:
|
||
raise FwError("镜像文件不在下载目录里了（被手动删过？）：重新上传")
|
||
return {"cmd": "ota", "url": fw["url"], "sha256": sha, "size": fw["size"], "reboot": reboot}
|