armor: ESP32-C3 装甲板固件（一主三从）+ 烧录脚本，之前一直没提交
这一块代码本来就写好了、根 README 里也按"已实现"描述，但一直没有进仓库， 只存在于工作区里。同一个提交里带上 `.env.example` 的两个新变量（ARMOR_TEAM、 ARMOR_THRESHOLD）——它们的说明文字就是为这套固件写的。 内容： - armor-protocol：规则层（血量、每板独立 50ms 受击冷却、受击后只有那一路灯带全黑、 按键 FSM、上报编码）。纯计算，主机上 22 个测试，已跑通。 - armor-firmware：四路压电片 ADC 连续采样、四路 WS2812（C3 只有一个 RMT 通道可用， 靠 rmt_tx_switch_gpio 轮流驱动）、NVS、BOOT 键、WiFi + UDP 广播。 - scripts：build_armor.sh / flash_armor.sh / armor_probe.py（不开界面单看广播）。 armor/ 是独立 workspace（目标 RISC-V 的 ESP32-C3），与外层 Xtensa 的图传固件互不相干， 各自 .cargo/config.toml 就是这条边界。构建产物由根 .gitignore 的 `/armor/target/` 覆盖， 本次没有把任何 target/ 或 build/ 下的东西加进来。 顺带把 `.DS_Store` 加进 .gitignore。
This commit is contained in:
Executable
+167
@@ -0,0 +1,167 @@
|
||||
#!/usr/bin/env python3
|
||||
"""独立验证装甲板血量上报的诊断脚本。
|
||||
|
||||
不依赖 Rust，也不依赖 protocol / armor-protocol crate——常量是手写的。正因如此它
|
||||
是线路格式的**独立**检查：如果 Rust 侧的编码写错了，这个脚本仍按文档解析，会立刻
|
||||
报出不一致。
|
||||
|
||||
用法：
|
||||
|
||||
python3 scripts/armor_probe.py [秒数]
|
||||
|
||||
它监听 UDP 5601（装甲板主板的广播目标），把收到 `ArmorHealth` 数据报按
|
||||
`(team, index)` 汇总，打印血量变化与心跳间隔。板子每 200 ms 广播四条，所以
|
||||
正常情况下每秒每个 `(team, index)` 大约 5 条。
|
||||
|
||||
用它可以确认三件事，都不需要打开裁判端界面：
|
||||
|
||||
1. 板子在广播、IP 与 MAC 派生 id 是多少
|
||||
2. 四条 index 的血量是否一致（协议约定：整车一条血，四格同值）
|
||||
3. 血量有没有随受击下降（每击 −3，100 起）
|
||||
|
||||
常量与 protocol/src/lib.rs 保持一致。改了协议记得改这里——这正是它作为交叉检查
|
||||
的价值所在。
|
||||
"""
|
||||
import socket
|
||||
import sys
|
||||
import time
|
||||
from collections import defaultdict
|
||||
|
||||
# ---- 与 protocol/src/lib.rs 一致 ----
|
||||
HEADER = 27 # magic4 + type1 + ver2 + sender4 + frame4 + idx2 + cnt2 + len4 + w2 + h2
|
||||
MAGIC = 0x4D564331 # "MVC1"
|
||||
ARMOR_PORT = 5601 # ARMOR_PORT
|
||||
MSG_ARMOR_HEALTH = 5 # MsgType::ArmorHealth
|
||||
|
||||
TEAM_NAMES = {0: "队伍一", 1: "队伍二"}
|
||||
|
||||
|
||||
def decode(data):
|
||||
"""解析一个数据报，返回 (team, index, hp, sender_id, version) 或 None。"""
|
||||
if len(data) < HEADER + 4:
|
||||
return None
|
||||
|
||||
if int.from_bytes(data[0:4], "little") != MAGIC:
|
||||
return None
|
||||
|
||||
if data[4] != MSG_ARMOR_HEALTH:
|
||||
return None
|
||||
|
||||
version = int.from_bytes(data[5:7], "little")
|
||||
sender_id = int.from_bytes(data[7:11], "little")
|
||||
team = data[HEADER]
|
||||
index = data[HEADER + 1]
|
||||
hp = int.from_bytes(data[HEADER + 2 : HEADER + 4], "little")
|
||||
|
||||
return team, index, hp, sender_id, version
|
||||
|
||||
|
||||
def main():
|
||||
seconds = float(sys.argv[1]) if len(sys.argv) > 1 else 10.0
|
||||
|
||||
sock = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
|
||||
sock.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
|
||||
sock.bind(("", ARMOR_PORT))
|
||||
sock.settimeout(0.5)
|
||||
|
||||
print(f"listening on UDP {ARMOR_PORT} for {seconds:g}s ...")
|
||||
|
||||
# (team, index) -> 最后一次 (hp, 收到时刻)
|
||||
latest = {}
|
||||
# (team, index) -> 收到过的 hp 序列（去重后按时间）
|
||||
history = defaultdict(list)
|
||||
# (team, index) -> 包数
|
||||
counts = defaultdict(int)
|
||||
senders = set()
|
||||
versions = set()
|
||||
|
||||
deadline = time.time() + seconds
|
||||
while time.time() < deadline:
|
||||
try:
|
||||
data, addr = sock.recvfrom(2048)
|
||||
except socket.timeout:
|
||||
continue
|
||||
|
||||
parsed = decode(data)
|
||||
if parsed is None:
|
||||
continue
|
||||
|
||||
team, index, hp, sender_id, version = parsed
|
||||
now = time.time()
|
||||
key = (team, index)
|
||||
counts[key] += 1
|
||||
senders.add((sender_id, f"{addr[0]}:{addr[1]}"))
|
||||
versions.add(version)
|
||||
|
||||
previous = latest.get(key)
|
||||
if previous is None or previous[0] != hp:
|
||||
history[key].append((round(now, 3), hp))
|
||||
print(
|
||||
f" {TEAM_NAMES.get(team, f'team{team}')} plate {index}: "
|
||||
f"hp {hp}"
|
||||
+ (f" (was {previous[0]})" if previous else " (first seen)")
|
||||
)
|
||||
latest[key] = (hp, now)
|
||||
|
||||
print()
|
||||
|
||||
if not counts:
|
||||
print("没有收到装甲板上报。检查：")
|
||||
print(" - 板子上电了吗（UART0 上应有 'reporting to 255.255.255.255:5601'）")
|
||||
print(" - 板子和这台机器在同一个局域网、同一个子网吗")
|
||||
print(" - 防火墙放行 UDP 5601 了吗")
|
||||
return 1
|
||||
|
||||
print(f"发送方 ({len(senders)}):")
|
||||
for sender_id, addr in sorted(senders):
|
||||
print(f" 0x{sender_id:08X} {addr}")
|
||||
|
||||
print(f"协议版本: {sorted(versions)}")
|
||||
print()
|
||||
|
||||
by_team = defaultdict(dict)
|
||||
for (team, index), count in sorted(counts.items()):
|
||||
by_team[team][index] = count
|
||||
|
||||
elapsed = max(seconds, 0.001)
|
||||
|
||||
print("每个 (队伍, index) 的包数 / 心率:")
|
||||
for team, plates in sorted(by_team.items()):
|
||||
cells = []
|
||||
for index in range(4):
|
||||
n = plates.get(index, 0)
|
||||
cells.append(f"index{index}={n} ({n / elapsed:.1f}/s)")
|
||||
print(f" {TEAM_NAMES.get(team, f'team{team}')}: " + " ".join(cells))
|
||||
|
||||
# 协议约定：整车一条血 → 四格同值；四条都在新鲜期内才等于"这条队伍在线"。
|
||||
print()
|
||||
print("四条 index 的血量:")
|
||||
for team, plates in sorted(by_team.items()):
|
||||
hps = []
|
||||
for index in range(4):
|
||||
reading = latest.get((team, index))
|
||||
hps.append(str(reading[0]) if reading else "-")
|
||||
same = len(set(hps)) == 1 and "-" not in hps
|
||||
print(
|
||||
f" {TEAM_NAMES.get(team, f'team{team}')}: [{', '.join(hps)}]"
|
||||
+ (" <- 一致（整车一条血量）" if same else " <- 不一致：检查 index 或丢包")
|
||||
)
|
||||
|
||||
if history:
|
||||
print()
|
||||
print("血量变化:")
|
||||
for key in sorted(history):
|
||||
team, index = key
|
||||
events = history[key]
|
||||
if len(events) == 1:
|
||||
continue
|
||||
line = " ".join(f"t={t:+.1f}s hp={hp}" for t, hp in events[:12])
|
||||
if len(events) > 12:
|
||||
line += f" ... (+{len(events) - 12} more)"
|
||||
print(f" {TEAM_NAMES.get(team, f'team{team}')} plate {index}: {line}")
|
||||
|
||||
return 0
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
sys.exit(main())
|
||||
Reference in New Issue
Block a user