zhou yang和Claude Opus 5.5
1cd4fd4082
- 固件库里登记过的（界面上传）不自动删；没登记的（ota_push.py 直传、台架镜像）放满 7 天删； 上传中断的 .tmp 1 小时删；非 <16 hex>.bin / .tmp 的文件名一律不碰 - 板子下载一次几秒就完、回滚用的是板上另一个槽，删旧文件不影响任何板子 - main.py：有 --fw-dir 时起清理循环，出错只记日志；test_firmware.py 加 2 例，共 41 例（本机 + 服务器） - 已部署；本机模拟板实测放一个 8 天前的孤儿镜像，启动即被清掉 Co-Authored-By: Claude Opus 5.5 <[email protected]>
183 行
8.3 KiB
Python
183 行
8.3 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
|
||
import time
|
||
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}$")
|
||
|
||
# 自动清理（prune）。下载目录里只有两种文件归本模块管：`<16 hex>.bin` 与上传中途的 `.<16 hex>.bin.tmp`；
|
||
# 别的名字一律不碰（目录是给 nginx 的，谁往里放了别的东西，不是这里该删的）。
|
||
FILE_RE = re.compile(r"^[0-9a-f]{16}\.bin$")
|
||
TMP_RE = re.compile(r"^\.[0-9a-f]{16}\.bin\.tmp$")
|
||
ORPHAN_KEEP_S = 7 * 86400 # 库里没登记的镜像（ota_push.py 直接传的、台架镜像）留 7 天
|
||
TMP_KEEP_S = 3600 # 上传中断留下的半截文件
|
||
|
||
# esp_app_desc_t 紧跟在镜像头（24 B）+ 第一个段头（8 B）之后
|
||
_DESC = 32
|
||
_DESC_MAGIC = 0xABCD5432
|
||
# 能力标记（firmware/main/warm_esp.c 的 g_phonerail_caps，.rodata_custom_desc）紧跟 esp_app_desc_t（256 B）
|
||
_CAPS = _DESC + 256
|
||
_CAPS_TAG = b"PRCAPS1\0"
|
||
CAP_WARM = 1 # 认「重启不断电」的接管：推它并立刻重启，手机不断电
|
||
|
||
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))
|
||
caps = struct.unpack_from("<I", data, _CAPS + 8)[0] if data[_CAPS:_CAPS + 8] == _CAPS_TAG else 0
|
||
info["warm"] = bool(caps & CAP_WARM)
|
||
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)
|
||
path = os.path.join(self.dir, self.name_for(sha))
|
||
d["present"] = os.path.isfile(path)
|
||
d["warm"] = False # 从文件现读，不进库（不用改表）
|
||
if d["present"]:
|
||
try:
|
||
with open(path, "rb") as f:
|
||
d["warm"] = parse_image(f.read(_CAPS + 12) + b"\0" * 256)["warm"]
|
||
except (OSError, FwError):
|
||
pass
|
||
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 prune(self, now_s: Optional[float] = None) -> List[str]:
|
||
"""清掉**库里没登记**、而且放了 ORPHAN_KEEP_S 以上的镜像，以及 TMP_KEEP_S 以上的半截上传。
|
||
返回删掉的文件名。
|
||
|
||
⚠️ 固件库里登记过的（界面上传的）**不自动删** —— 那是人挑出来的版本，界面上有删除按钮。
|
||
没登记的删了也不影响板子：板子下载一次几秒就完，之后不再需要服务器上的文件
|
||
（回滚用的是板子自己另一个槽里的旧版，不是这里的）。7 天是给「推了一半、第二天接着推」留的余地。
|
||
"""
|
||
now = time.time() if now_s is None else now_s
|
||
known = {self.name_for(r["sha256"]) for r in self.db.q("SELECT sha256 FROM firmware")}
|
||
gone = []
|
||
try:
|
||
names = os.listdir(self.dir)
|
||
except FileNotFoundError:
|
||
return gone
|
||
for name in names:
|
||
if FILE_RE.match(name):
|
||
if name in known:
|
||
continue
|
||
keep = ORPHAN_KEEP_S
|
||
elif TMP_RE.match(name):
|
||
keep = TMP_KEEP_S
|
||
else:
|
||
continue
|
||
path = os.path.join(self.dir, name)
|
||
try:
|
||
if now - os.path.getmtime(path) < keep:
|
||
continue
|
||
os.remove(path)
|
||
gone.append(name)
|
||
except OSError:
|
||
pass # 刚被别人删了 / 权限：下一轮再说
|
||
return gone
|
||
|
||
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}
|