zhou yang和Claude Opus 5.5
7bf30e3c38
- warm_esp.{h,c}：warm_restart() 记下开着的路（RTC_NOINIT，魔数 + 掩码 + 反码）→ gpio_hold 锁五根 EN → 重启；
目标镜像没有能力标记（第 288 字节 PRCAPS1，.rodata_custom_desc）就不锁 —— 老固件写不动锁住的 EN。
warm_take()：软件复位 + 标记对得上才接管，一次性；再读 EN 真电平核对
- bringup：adopt 掩码；rail_en_init_keep() 接管的路写高、五根一律放开 hold（崩溃时旧锁可能还在）；
刀片不在 / mux 失联的接管路立刻关（SLOT_EMPTY arg 1）；第二段已开着的路不重开、不等 PG
- pca9555_init_keep()：方向寄存器是本固件写的才留 VBUS（掉过电的片子锁存器是 0xFF），键 / EDL 清零
- HAL：hal_gpio_hold()；ESP 输出脚配 INPUT_OUTPUT（读得到真电平）；sim 给 hold 建模
- 事件 LOG_EV_OTA arg 7 = 接管；兜底定时器回滚故意不走接管
- 顶层 CMake：phonerail.h 进 CMAKE_CONFIGURE_DEPENDS（否则镜像头版本停在旧值）；-u g_phonerail_caps 防 gc
- test_bringup 加 6 例（接管 / 非计划复位全关 / 空槽 / mux 失联 / PCA 掉过电 / hold 冻结），全工程 448 例全绿
- rack：固件库识别能力标记，推送对话框按「支持 / 不支持重启不断电」给不同提示（已部署）
- 需求文档：SR-6 改为「非计划复位全关」+ 计划内重启例外；§10.9 新增「重启不断电」
- 板 B：台架镜像（空槽强制 CH1 在位，未提交）开 CH1 → OTA 带重启 → 事件 18/7 mv=1，CH1 一直开着；
现跑 0.1.0-c3。未测：示波器看重启期间 EN / 输出毛刺（要插刀片）
Co-Authored-By: Claude Opus 5.5 <[email protected]>
141 行
6.3 KiB
Python
141 行
6.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
|
||
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
|
||
# 能力标记（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 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}
|