Horldsence
a89d7ae5fb
把桌面端从 Slint 迁到 Tauri v2，后端先落地，界面待接。拆成三层： - desktop/core（cam-core，67 测试）：从旧 client/ 抽出的非界面部分。UDP 接收按 发送方重组、装甲板发现与血量、帧率窗口、槽位分配、配置与密码。 - desktop/src-tauri（27 测试）：Tauri 后端。回环 WebSocket 推二进制 JPEG、 每 200ms 一份状态快照、密码闸门、截图、ffmpeg 录制 mp4、演示数据源。 - protocol：新增 ArmorHealth::decode_from，把帧头里的 sender_id 交出来—— 装甲板一直带着 MAC 派生 id 在广播，旧解码路径把它丢了，所以"装甲板下拉可选" 不需要改固件。 几个改对了的旧问题： - 槽位分配：旧实现只排除"槽位自己原来那个 id"，于是队伍一的板子一离线，它的槽位 会去拿队伍二那块，两格显示同一块板子。现在候选要排除另一个槽位占着的 id。 - 帧率窗口的基线：掉线又回来的板子带着一个很大的累计帧数回来，从 0 起算会报出 几百 fps。改成 seeded + 用"已结算窗口数"判断有没有读数。 - 密码哈希：password-hash 能解析 `$argon2id$broken` 这种截断串，于是"配置坏了" 会被读成"密码错了"。现在要求盐与摘要都在。 - 录制文件名：同一秒停一次再开一次会用同一个名字，ffmpeg 的 -y 静默覆盖上一段。 现在先收尾上一段再开新的，并且重名往后排号。 - 两道闸门：帧的限制与发现列表的限制分开。设置界面需要列候选摄像头，但**画面** 不需要——旧版把两件事绑在一道闸门上，只能在"能列候选"和"挡住对手"之间二选一。 测试：protocol 15、cam-core 67、desktop 27，其中录制那条真的起 ffmpeg 并用 ffprobe 读回帧数与分辨率（只断言"文件存在"会漏掉空容器）。
1016 lines
35 KiB
Rust
1016 lines
35 KiB
Rust
//! 线路格式：图传视频 + 设备发现 + 装甲板预留接口。
|
||
//!
|
||
//! 三端共用这一个 crate：两块 ESP32 图传板、队伍客户端、裁判终端。
|
||
//!
|
||
//! # 拓扑
|
||
//!
|
||
//! ```text
|
||
//! ┌──────────────────────────┐
|
||
//! │ 裁判终端（PC，仅显示） │
|
||
//! │ 队伍一 + 队伍二 + 血量 │
|
||
//! └───────────┬──────────────┘
|
||
//! │ 同一个局域网，广播可用
|
||
//! ┌──────────────────┼──────────────────┐
|
||
//! │ │ │
|
||
//! ┌───────┴───────┐ ┌───────┴───────┐ ┌───────┴───────┐
|
||
//! │ 图传板 A │ │ 图传板 B │ │ 装甲板 ×4 │
|
||
//! │ (队伍一) │ │ (队伍二) │ │ (预留，未实现) │
|
||
//! └───────────────┘ └───────────────┘ └───────────────┘
|
||
//! ```
|
||
//!
|
||
//! # 为什么视频走 UDP 而不是 TCP
|
||
//!
|
||
//! 实测：传感器产出 22 fps，TCP MJPEG 只送出 1.9 fps，且只占用了链路能力的 1%。
|
||
//! 链路是空的，帧却堵在 TCP 的队头阻塞后面等一个实时画面等不起的重传计时器。
|
||
//! UDP 没有重传与确认，缺了分片的帧整个丢弃、下一帧顶上。
|
||
//!
|
||
//! # 分片
|
||
//!
|
||
//! 每个视频数据报 = [`HEADER_LEN`] 字节头 + 至多 [`MAX_FRAGMENT_PAYLOAD`]
|
||
//! 字节 JPEG 数据。头里的 `sender_id` 让接收端能在同一端口上区分多块板子，
|
||
//! 这是裁判终端能同时看两路的前提。
|
||
//!
|
||
//! ```text
|
||
//! 0 1 2 3
|
||
//! 0 1 2 3 4 5 6 7 8 9 0 1 2 3 4 5 6 7 8 9 0 1 2 3 4 5 6 7 8 9 0 1
|
||
//! +-------------------------------+-------+-----------------------+
|
||
//! | magic (u32) | type | version (u16) |
|
||
//! +-------------------------------+-------+-----------------------+
|
||
//! | sender_id (u32) | frame_id (u32) |
|
||
//! +-------------------------------+-------+-----------------------+
|
||
//! | frag_index (u16) | frag_count (u16) |
|
||
//! +-------------------------------+-------+-----------------------+
|
||
//! | total_len (u32) | width (u16) | height (u16) |
|
||
//! +-------------------------------+---------------+---------------+
|
||
//! | payload (0..=1400) |
|
||
//! +---------------------------------------------------------------+
|
||
//! ```
|
||
|
||
use std::net::{Ipv4Addr, SocketAddrV4};
|
||
|
||
// ---------------------------------------------------------------------------
|
||
// 端口
|
||
// ---------------------------------------------------------------------------
|
||
|
||
/// 视频与设备发现共用端口。
|
||
///
|
||
/// 发现和控制消息体积小、频率低，与视频混在一个端口上可以让板子只开一个 socket，
|
||
/// 简化固件侧。视频与控制的数据报头不同（`msg_type`），不会混淆。
|
||
pub const DEFAULT_PORT: u16 = 5600;
|
||
|
||
/// 装甲板预留端口。
|
||
///
|
||
/// 现在没有实现，裁判终端只是监听它并把血量显示为占位符。板子接上以后往这个端口
|
||
/// 发消息即可，不需要改动线路格式或界面结构。
|
||
pub const ARMOR_PORT: u16 = 5601;
|
||
|
||
// ---------------------------------------------------------------------------
|
||
// 帧头
|
||
// ---------------------------------------------------------------------------
|
||
|
||
/// 线路格式版本。
|
||
///
|
||
/// 字段增删时递增。不匹配的对端会拒绝处理，而不是按错误布局解析出垃圾数据——
|
||
/// 那会表现为随机崩溃或画面错乱，比拒绝连接难查得多。
|
||
pub const PROTOCOL_VERSION: u16 = 1;
|
||
|
||
/// 标识本协议。
|
||
///
|
||
/// 取值使得它不可能出现在 JPEG 数据开头，所以一个同端口上的无关 UDP 包会被挡掉。
|
||
pub const MAGIC: u32 = 0x4D56_4331; // "MVC1"
|
||
|
||
/// 帧头长度（字节）。
|
||
///
|
||
/// `magic`(4) + `msg_type`(1) + `version`(2) + `sender_id`(4) + `frame_id`(4) +
|
||
/// `frag_index`(2) + `frag_count`(2) + `total_len`(4) + `width`(2) + `height`(2)
|
||
/// = **27**。
|
||
///
|
||
/// 27 不如 4 字节对齐，但这里刻意不去凑：头只占每个 1427 字节数据报的 1.9%，而
|
||
/// 为对齐在固件和客户端两侧各写一段手写的偏移常量，是引入静默错位的更划算的代价。
|
||
pub const HEADER_LEN: usize = 27;
|
||
|
||
/// 单个数据报的最大 JPEG 载荷。
|
||
///
|
||
/// 1400 + 27 = 1427 字节，低于典型以太网的 1500 MTU。这一点比看起来重要：超 MTU
|
||
/// 的数据报要么被 AP 丢弃，要么被 IP 分片，两种情况的丢包都记在应用账上而不是网络账
|
||
/// 上——后者本来更容易排查。
|
||
pub const MAX_FRAGMENT_PAYLOAD: usize = 1400;
|
||
|
||
/// 数据报类型。
|
||
///
|
||
/// 决定同一端口上的字节如何解释。把控制和视频放在一个端口的前提就是这个字段
|
||
/// 一定在最前面，且值互不相同。
|
||
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
|
||
#[repr(u8)]
|
||
pub enum MsgType {
|
||
/// 一帧 JPEG 的第 N 个分片。
|
||
Video = 1,
|
||
/// 客户端广播："我是这个角色，请给我画面。"
|
||
Subscribe = 2,
|
||
/// 板子回复："我在线，我是这个 id。"
|
||
Hello = 3,
|
||
/// 心跳，用于发现与保活。
|
||
Heartbeat = 4,
|
||
/// 装甲板上报血量（预留，当前无实现）。
|
||
ArmorHealth = 5,
|
||
}
|
||
|
||
impl MsgType {
|
||
/// 从线上的字节还原。
|
||
pub fn from_u8(v: u8) -> Option<Self> {
|
||
Some(match v {
|
||
1 => MsgType::Video,
|
||
2 => MsgType::Subscribe,
|
||
3 => MsgType::Hello,
|
||
4 => MsgType::Heartbeat,
|
||
5 => MsgType::ArmorHealth,
|
||
_ => return None,
|
||
})
|
||
}
|
||
}
|
||
|
||
/// 客户端在订阅里声明的角色。
|
||
///
|
||
/// 板子据此决定要不要给这个客户端推流。裁判终端用
|
||
/// [`SubscribeRole::Referee`](SubscribeRole::Referee) 表示"我要两路都看"，但它仍然
|
||
/// 是靠收到两块板子的 Hello 后分别订阅来完成——角色只是省掉了手工筛选。
|
||
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
|
||
#[repr(u8)]
|
||
pub enum SubscribeRole {
|
||
/// 队伍客户端，只要一台板子的画面。
|
||
Team = 1,
|
||
/// 裁判终端，要全部板子的画面。
|
||
Referee = 2,
|
||
}
|
||
|
||
impl SubscribeRole {
|
||
pub fn from_u8(v: u8) -> Option<Self> {
|
||
match v {
|
||
1 => Some(SubscribeRole::Team),
|
||
2 => Some(SubscribeRole::Referee),
|
||
_ => None,
|
||
}
|
||
}
|
||
}
|
||
|
||
/// 数据报头，20 字节，小端。
|
||
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
|
||
#[repr(C)]
|
||
pub struct Header {
|
||
pub msg_type: MsgType,
|
||
pub version: u16,
|
||
/// 发送方标识。
|
||
///
|
||
/// 板子用 MAC 派生（见 [`id_from_mac`]），这样重启后 id 不变，客户端保存的绑定
|
||
/// 不会失效。裁判终端用固定的 [`REFEREE_ID`]。
|
||
pub sender_id: u32,
|
||
/// 帧序号，按发送方独立计数，回绕无害。
|
||
pub frame_id: u32,
|
||
pub frag_index: u16,
|
||
pub frag_count: u16,
|
||
/// 整张 JPEG 的长度。
|
||
pub total_len: u32,
|
||
pub width: u16,
|
||
pub height: u16,
|
||
}
|
||
|
||
impl Default for Header {
|
||
fn default() -> Self {
|
||
Self {
|
||
msg_type: MsgType::Video,
|
||
version: PROTOCOL_VERSION,
|
||
sender_id: 0,
|
||
frame_id: 0,
|
||
frag_index: 0,
|
||
frag_count: 0,
|
||
total_len: 0,
|
||
width: 0,
|
||
height: 0,
|
||
}
|
||
}
|
||
}
|
||
|
||
/// 解析帧头并校验魔数与版本。
|
||
///
|
||
/// 任何不满足的输入都返回 `None`，调用方据此当作"不是给我的"跳过，而不是当作
|
||
/// 错误——同一端口上可能出现板子的存活探测包，也可能日后出现其他协议。
|
||
pub fn decode_header(bytes: &[u8]) -> Option<Header> {
|
||
if bytes.len() < HEADER_LEN {
|
||
return None;
|
||
}
|
||
|
||
let magic = u32::from_le_bytes(bytes[0..4].try_into().ok()?);
|
||
if magic != MAGIC {
|
||
return None;
|
||
}
|
||
|
||
let msg_type = MsgType::from_u8(bytes[4])?;
|
||
|
||
let version = u16::from_le_bytes(bytes[5..7].try_into().ok()?);
|
||
if version != PROTOCOL_VERSION {
|
||
return None;
|
||
}
|
||
|
||
Some(Header {
|
||
msg_type,
|
||
version,
|
||
sender_id: u32::from_le_bytes(bytes[7..11].try_into().ok()?),
|
||
frame_id: u32::from_le_bytes(bytes[11..15].try_into().ok()?),
|
||
frag_index: u16::from_le_bytes(bytes[15..17].try_into().ok()?),
|
||
frag_count: u16::from_le_bytes(bytes[17..19].try_into().ok()?),
|
||
total_len: u32::from_le_bytes(bytes[19..23].try_into().ok()?),
|
||
width: u16::from_le_bytes(bytes[23..25].try_into().ok()?),
|
||
height: u16::from_le_bytes(bytes[25..27].try_into().ok()?),
|
||
})
|
||
}
|
||
|
||
/// 把帧头写进 `out`（追加，不清空）。
|
||
pub fn encode_header_into(header: &Header, out: &mut Vec<u8>) {
|
||
out.extend_from_slice(&MAGIC.to_le_bytes());
|
||
out.push(header.msg_type as u8);
|
||
out.extend_from_slice(&header.version.to_le_bytes());
|
||
out.extend_from_slice(&header.sender_id.to_le_bytes());
|
||
out.extend_from_slice(&header.frame_id.to_le_bytes());
|
||
out.extend_from_slice(&header.frag_index.to_le_bytes());
|
||
out.extend_from_slice(&header.frag_count.to_le_bytes());
|
||
out.extend_from_slice(&header.total_len.to_le_bytes());
|
||
out.extend_from_slice(&header.width.to_le_bytes());
|
||
out.extend_from_slice(&header.height.to_le_bytes());
|
||
|
||
debug_assert_eq!(out.len(), HEADER_LEN);
|
||
}
|
||
|
||
/// 裁判终端的固定标识。
|
||
///
|
||
/// 固定值而非 MAC 派生：裁判终端不是网络意义上的"设备集合"的一员，它只有一个，
|
||
/// 用常量比让它参与 MAC 派生更简单也更清楚。
|
||
pub const REFEREE_ID: u32 = 0;
|
||
|
||
/// 从 MAC 地址派生发送方标识。
|
||
///
|
||
/// 用固定长度输入的 FNV-1a：ESP32 没有随机数源可依赖，而 id 必须在重启后保持不变，
|
||
/// 否则客户端保存的绑定每次重启都会失效。
|
||
///
|
||
/// 返回 0 的输入（组播地址）被避开，因为 0 留给裁判终端。
|
||
pub fn id_from_mac(mac: &[u8; 6]) -> u32 {
|
||
const FNV_OFFSET: u32 = 0x811c9dc5;
|
||
const FNV_PRIME: u32 = 0x01000193;
|
||
|
||
let mut hash = FNV_OFFSET;
|
||
|
||
for byte in mac {
|
||
hash ^= *byte as u32;
|
||
hash = hash.wrapping_mul(FNV_PRIME);
|
||
}
|
||
|
||
// 避开 0，它是裁判终端的保留值。
|
||
if hash == 0 {
|
||
1
|
||
} else {
|
||
hash
|
||
}
|
||
}
|
||
|
||
/// 视频分片：一个数据报。
|
||
#[derive(Debug, Clone, PartialEq, Eq)]
|
||
pub struct Fragment {
|
||
pub header: Header,
|
||
pub payload: Vec<u8>,
|
||
}
|
||
|
||
impl Fragment {
|
||
/// 追加写入 `out`，不分配。
|
||
///
|
||
/// 发送端对每帧的每个分片复用同一个缓冲：在 22fps、每帧 8–10 个分片下，这是
|
||
/// 每秒约 250 次分配与不分配的区别——在这块同时要抢内部 RAM 的芯片上很实在。
|
||
pub fn encode_into(&self, out: &mut Vec<u8>) {
|
||
encode_header_into(&self.header, out);
|
||
out.extend_from_slice(&self.payload);
|
||
}
|
||
|
||
pub fn encode(&self) -> Vec<u8> {
|
||
let mut out = Vec::with_capacity(HEADER_LEN + self.payload.len());
|
||
self.encode_into(&mut out);
|
||
out
|
||
}
|
||
|
||
/// 解析一个数据报。
|
||
///
|
||
/// 头之后的所有字节都算载荷。校验不一致的分片在这里就被挡住，因此重组器不必
|
||
/// 再操心。
|
||
pub fn decode(bytes: &[u8]) -> Option<Self> {
|
||
let header = decode_header(bytes)?;
|
||
|
||
if header.msg_type != MsgType::Video {
|
||
return None;
|
||
}
|
||
|
||
// 一帧至少一个分片，且分片下标必须落在范围内。
|
||
if header.frag_count == 0 || header.frag_index >= header.frag_count {
|
||
return None;
|
||
}
|
||
|
||
let payload = &bytes[HEADER_LEN..];
|
||
|
||
// JPEG 不可能比承载它的分片还短。检查很便宜，却能在造成一次大额分配之前
|
||
// 挡掉损坏的头。
|
||
if payload.len() as u64 > header.total_len as u64 {
|
||
return None;
|
||
}
|
||
|
||
Some(Fragment {
|
||
header,
|
||
payload: payload.to_vec(),
|
||
})
|
||
}
|
||
}
|
||
|
||
/// 一张完整重组好的 JPEG。
|
||
#[derive(Debug, Clone, PartialEq, Eq)]
|
||
pub struct Frame {
|
||
/// 来源板子的标识。
|
||
pub sender_id: u32,
|
||
pub frame_id: u32,
|
||
pub width: u16,
|
||
pub height: u16,
|
||
pub data: Vec<u8>,
|
||
}
|
||
|
||
/// 把一张 JPEG 切成数据报。
|
||
///
|
||
/// 分片数由载荷大小推出，因此调用方不可能产出与自身矛盾的集合。
|
||
pub fn fragment_frame(
|
||
sender_id: u32,
|
||
frame_id: u32,
|
||
jpeg: &[u8],
|
||
width: u16,
|
||
height: u16,
|
||
) -> Vec<Fragment> {
|
||
let count = jpeg.len().div_ceil(MAX_FRAGMENT_PAYLOAD).max(1) as u16;
|
||
let total_len = jpeg.len() as u32;
|
||
|
||
(0..count)
|
||
.map(|i| {
|
||
let start = i as usize * MAX_FRAGMENT_PAYLOAD;
|
||
let end = (start + MAX_FRAGMENT_PAYLOAD).min(jpeg.len());
|
||
|
||
Fragment {
|
||
header: Header {
|
||
msg_type: MsgType::Video,
|
||
version: PROTOCOL_VERSION,
|
||
sender_id,
|
||
frame_id,
|
||
frag_index: i,
|
||
frag_count: count,
|
||
total_len,
|
||
width,
|
||
height,
|
||
},
|
||
payload: jpeg[start..end].to_vec(),
|
||
}
|
||
})
|
||
.collect()
|
||
}
|
||
|
||
/// 客户端的订阅请求。
|
||
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
|
||
pub struct Subscribe {
|
||
pub role: SubscribeRole,
|
||
pub client_id: u32,
|
||
}
|
||
|
||
impl Subscribe {
|
||
pub fn encode(&self) -> Vec<u8> {
|
||
let header = Header {
|
||
msg_type: MsgType::Subscribe,
|
||
version: PROTOCOL_VERSION,
|
||
sender_id: self.client_id,
|
||
..Default::default()
|
||
};
|
||
|
||
let mut out = Vec::with_capacity(HEADER_LEN + 1);
|
||
encode_header_into(&header, &mut out);
|
||
out.push(self.role as u8);
|
||
out
|
||
}
|
||
|
||
pub fn decode(bytes: &[u8]) -> Option<Self> {
|
||
let header = decode_header(bytes)?;
|
||
|
||
if header.msg_type != MsgType::Subscribe || bytes.len() < HEADER_LEN + 1 {
|
||
return None;
|
||
}
|
||
|
||
Some(Subscribe {
|
||
role: SubscribeRole::from_u8(bytes[HEADER_LEN])?,
|
||
client_id: header.sender_id,
|
||
})
|
||
}
|
||
}
|
||
|
||
/// 板子的在线通告与心跳。
|
||
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
|
||
pub struct Hello {
|
||
pub sender_id: u32,
|
||
/// 固件版本，便于现场排查烧录了哪一版。
|
||
pub firmware: u16,
|
||
}
|
||
|
||
impl Hello {
|
||
pub fn encode(&self, msg_type: MsgType) -> Vec<u8> {
|
||
let header = Header {
|
||
msg_type,
|
||
version: PROTOCOL_VERSION,
|
||
sender_id: self.sender_id,
|
||
..Default::default()
|
||
};
|
||
|
||
let mut out = Vec::with_capacity(HEADER_LEN + 2);
|
||
encode_header_into(&header, &mut out);
|
||
out.extend_from_slice(&self.firmware.to_le_bytes());
|
||
out
|
||
}
|
||
|
||
pub fn decode(bytes: &[u8]) -> Option<Self> {
|
||
let header = decode_header(bytes)?;
|
||
|
||
if !matches!(header.msg_type, MsgType::Hello | MsgType::Heartbeat) {
|
||
return None;
|
||
}
|
||
|
||
// 固件版本是可选的：早期或精简的通告可能不带这个字段。
|
||
let firmware = if bytes.len() >= HEADER_LEN + 2 {
|
||
u16::from_le_bytes(bytes[HEADER_LEN..HEADER_LEN + 2].try_into().ok()?)
|
||
} else {
|
||
0
|
||
};
|
||
|
||
Some(Hello {
|
||
sender_id: header.sender_id,
|
||
firmware,
|
||
})
|
||
}
|
||
}
|
||
|
||
// ---------------------------------------------------------------------------
|
||
// 装甲板
|
||
// ---------------------------------------------------------------------------
|
||
|
||
/// 一块装甲板的上报。
|
||
///
|
||
/// 由 `armor/armor-firmware` 里的主板（ESP32-C3）广播到 [`ARMOR_PORT`]，裁判终端监听
|
||
/// 该端口并把它显示成四格血量。整车一条血量，所以四条上报的 `hp` 相同、`index`
|
||
/// 分别是 0..3。
|
||
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
|
||
pub struct ArmorHealth {
|
||
/// 哪一边：0 = 队伍一，1 = 队伍二。
|
||
pub team: u8,
|
||
/// 机器人在该队伍的装甲板序号，0..3。
|
||
pub index: u8,
|
||
/// 当前血量。
|
||
pub hp: u16,
|
||
}
|
||
|
||
impl ArmorHealth {
|
||
/// 编码成线上字节，帧头的 `sender_id` 留 0。
|
||
///
|
||
/// 0 是 [`REFEREE_ID`]，所以板子发的时候应该用 [`encode_from`](Self::encode_from)
|
||
/// 填上自己的 id（由 MAC 派生），抓包时才能分辨是哪块板子。
|
||
pub fn encode(&self) -> Vec<u8> {
|
||
self.encode_from(0)
|
||
}
|
||
|
||
/// 与 [`encode`](Self::encode) 相同，但帧头带上发送方自己的 id。
|
||
pub fn encode_from(&self, sender_id: u32) -> Vec<u8> {
|
||
let header = Header {
|
||
msg_type: MsgType::ArmorHealth,
|
||
version: PROTOCOL_VERSION,
|
||
sender_id,
|
||
..Default::default()
|
||
};
|
||
|
||
let mut out = Vec::with_capacity(HEADER_LEN + 4);
|
||
encode_header_into(&header, &mut out);
|
||
out.push(self.team);
|
||
out.push(self.index);
|
||
out.extend_from_slice(&self.hp.to_le_bytes());
|
||
out
|
||
}
|
||
|
||
pub fn decode(bytes: &[u8]) -> Option<Self> {
|
||
Self::decode_from(bytes).map(|(_, health)| health)
|
||
}
|
||
|
||
/// 与 [`decode`](Self::decode) 相同，但把帧头里的 `sender_id` 一起返回。
|
||
///
|
||
/// 板子用 [`encode_from`](Self::encode_from) 把自己的 MAC 派生 id 填进去，所以
|
||
/// 这一个 u32 就是**哪块装甲板主板在说话**。裁判端与队伍端要列出"场上有哪几块
|
||
/// 装甲板"并让用户选中其中一块，靠的就是它——没有它就只剩 `team` 字段，
|
||
/// 同队的两块板子无从区分。
|
||
///
|
||
/// 保持 [`decode`](Self::decode) 不变是有意的：`team`/`index`/`hp` 是上报的语义，
|
||
/// 而 `sender_id` 是链路层的信封，不需要它的人不该被迫处理它。
|
||
pub fn decode_from(bytes: &[u8]) -> Option<(u32, Self)> {
|
||
let header = decode_header(bytes)?;
|
||
|
||
if header.msg_type != MsgType::ArmorHealth || bytes.len() < HEADER_LEN + 4 {
|
||
return None;
|
||
}
|
||
|
||
let health = ArmorHealth {
|
||
team: bytes[HEADER_LEN],
|
||
index: bytes[HEADER_LEN + 1],
|
||
hp: u16::from_le_bytes(bytes[HEADER_LEN + 2..HEADER_LEN + 4].try_into().ok()?),
|
||
};
|
||
|
||
Some((header.sender_id, health))
|
||
}
|
||
}
|
||
|
||
/// 视频端口地址。
|
||
pub fn video_endpoint(ip: Ipv4Addr) -> SocketAddrV4 {
|
||
SocketAddrV4::new(ip, DEFAULT_PORT)
|
||
}
|
||
|
||
/// 255.255.255.255，用于广播发现。
|
||
pub const BROADCAST_ADDR: Ipv4Addr = Ipv4Addr::new(255, 255, 255, 255);
|
||
|
||
// ---------------------------------------------------------------------------
|
||
// 重组
|
||
// ---------------------------------------------------------------------------
|
||
|
||
/// 帧被丢弃的原因。
|
||
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
|
||
pub enum DropReason {
|
||
/// 新的帧开始了，而这一张还没收齐。
|
||
Superseded,
|
||
/// 发送方沉默过久。
|
||
TimedOut,
|
||
}
|
||
|
||
/// 从乱序、有丢失的 UDP 数据报里重组帧。
|
||
///
|
||
/// 策略就是全部要点：**一张帧只有所有分片都到齐才交付，且最新的帧永远优先。**
|
||
///
|
||
/// UDP 不保证跨数据报的顺序与到达，所以试图"补洞"的接收端做的功比"等下一张完整
|
||
/// 帧"更多——而且补的过程中会显示撕裂的画面。
|
||
///
|
||
/// 超时堵住另一种失败：发送方在帧中途死掉时，半截缓冲会永远留在这里。
|
||
pub struct FrameAssembler {
|
||
sender_id: u32,
|
||
current_id: Option<u32>,
|
||
expected: usize,
|
||
buffer: Vec<u8>,
|
||
width: u16,
|
||
height: u16,
|
||
since: Option<std::time::Instant>,
|
||
timeout: std::time::Duration,
|
||
/// 已收到的分片数（不含重复）。用来统计网络丢了多少。
|
||
seen: usize,
|
||
/// 每个分片是否已收到。
|
||
///
|
||
/// 存在的原因是 `seen` 会被重复包灌水：UDP 会重复投递，一个丢失的分片可以
|
||
/// 被另一个包的副本凑数，于是"收够了"却留着空洞，交出去一张尾部是 0 的坏图。
|
||
have: Vec<bool>,
|
||
}
|
||
|
||
impl FrameAssembler {
|
||
/// 新建一个重组器，`timeout` 后放弃未收齐的帧。
|
||
pub fn new(timeout: std::time::Duration) -> Self {
|
||
Self {
|
||
sender_id: 0,
|
||
current_id: None,
|
||
expected: 0,
|
||
buffer: Vec::new(),
|
||
width: 0,
|
||
height: 0,
|
||
since: None,
|
||
timeout,
|
||
seen: 0,
|
||
have: Vec::new(),
|
||
}
|
||
}
|
||
|
||
/// 这个重组器正在处理哪个发送方。
|
||
pub fn sender_id(&self) -> u32 {
|
||
self.sender_id
|
||
}
|
||
|
||
/// 收到一个分片。
|
||
///
|
||
/// 这张分片补齐一张帧时返回 [`Frame`]。
|
||
///
|
||
/// `Err(Superseded)` 表示这个分片属于一张**已被放弃**的帧。多路复用时（裁判
|
||
/// 终端同时收两块板子）必须每个发送方一个重组器——混在一起会让两块板子的分片
|
||
/// 互相顶掉。这里不做检查，是调用方的责任。
|
||
pub fn push(&mut self, frag: Fragment) -> Result<Frame, DropReason> {
|
||
let sender_id = frag.header.sender_id;
|
||
let frame_id = frag.header.frame_id;
|
||
|
||
// 换了发送方：上一张属于别人的帧就此作废。
|
||
if self.sender_id != 0 && self.sender_id != sender_id {
|
||
self.reset();
|
||
}
|
||
self.sender_id = sender_id;
|
||
|
||
let superseded = match self.current_id {
|
||
None => false,
|
||
Some(id) => id != frame_id,
|
||
};
|
||
|
||
if superseded {
|
||
self.reset();
|
||
}
|
||
|
||
match self.current_id {
|
||
None => {
|
||
self.current_id = Some(frame_id);
|
||
self.expected = frag.header.frag_count as usize;
|
||
self.buffer.clear();
|
||
// 按 total_len 预留，最后拼接时不再重新分配。
|
||
self.buffer.reserve(frag.header.total_len as usize);
|
||
self.width = frag.header.width;
|
||
self.height = frag.header.height;
|
||
self.since = Some(std::time::Instant::now());
|
||
self.seen = 0;
|
||
self.have.clear();
|
||
self.have.resize(self.expected, false);
|
||
}
|
||
Some(id) if id == frame_id => {}
|
||
_ => unreachable!("current_id was just reconciled with frame_id"),
|
||
}
|
||
|
||
let index = frag.header.frag_index as usize;
|
||
|
||
// 重复分片直接跳过，且**不**增加计数。
|
||
//
|
||
// 用 `seen` 一个计数判断收齐是不够健壮的：一个丢失的分片若能被某个分片的
|
||
// 副本凑数，计数就达标了，而缓冲区里留着空洞，交出去一张尾部是 0 的 JPEG。
|
||
// 位图让"收到过"这件事是显式的，不用靠计数的算术去推断。
|
||
//
|
||
// 要说明的是：在本项目实测的链路上（2.4GHz AP，一到两个订阅者，20 秒 2000+
|
||
// 个分片）**一次重复投递都没出现过**，所以这层防护是为了正确性而不是为了
|
||
// 修一个正在发生的故障。之前观察到的 34 张坏 JPEG 另有原因：探测脚本自己
|
||
// 的重组状态只按 sender_id 存、不按 frame_id 存，于是丢片之后下一帧的分片
|
||
// 会接着凑上一帧的计数。那个脚本已修，见 scripts/udp_probe.py。
|
||
if self.have[index] {
|
||
return Err(DropReason::Superseded);
|
||
}
|
||
self.have[index] = true;
|
||
|
||
let start = index * MAX_FRAGMENT_PAYLOAD;
|
||
let end = start + frag.payload.len();
|
||
|
||
if end > self.buffer.len() {
|
||
self.buffer.resize(end, 0);
|
||
}
|
||
self.buffer[start..end].copy_from_slice(&frag.payload);
|
||
self.seen += 1;
|
||
|
||
// 计数达标还不够：最后一个分片必须真的到过，否则中间可能缺一块。
|
||
// 显式检查尾块，比依赖计数的算术更直接，也更容易看出意图。
|
||
let tail_index =
|
||
(frag.header.total_len as usize).saturating_sub(1) / MAX_FRAGMENT_PAYLOAD;
|
||
|
||
if self.seen >= self.expected && self.have[tail_index] {
|
||
let total = frag.header.total_len as usize;
|
||
self.buffer.truncate(total.min(self.buffer.len()));
|
||
|
||
let frame = Frame {
|
||
sender_id,
|
||
frame_id,
|
||
width: self.width,
|
||
height: self.height,
|
||
data: std::mem::take(&mut self.buffer),
|
||
};
|
||
|
||
self.reset();
|
||
return Ok(frame);
|
||
}
|
||
|
||
Err(DropReason::Superseded)
|
||
}
|
||
|
||
/// 若未收齐的帧等待过久则丢弃它。
|
||
///
|
||
/// 需要由定时器调用。否则发送方在首末分片之间死掉时，重组器会一直留着那张帧。
|
||
pub fn expire(&mut self) -> Option<DropReason> {
|
||
match self.since {
|
||
Some(started) if started.elapsed() > self.timeout => {
|
||
self.reset();
|
||
Some(DropReason::TimedOut)
|
||
}
|
||
_ => None,
|
||
}
|
||
}
|
||
|
||
/// 是否有一张帧正在收。
|
||
pub fn is_in_progress(&self) -> bool {
|
||
self.current_id.is_some()
|
||
}
|
||
|
||
fn reset(&mut self) {
|
||
self.current_id = None;
|
||
self.expected = 0;
|
||
self.buffer = Vec::new();
|
||
self.buffer.shrink_to_fit();
|
||
self.since = None;
|
||
self.seen = 0;
|
||
self.have = Vec::new();
|
||
}
|
||
}
|
||
|
||
#[cfg(test)]
|
||
mod tests {
|
||
use super::*;
|
||
|
||
#[test]
|
||
fn header_roundtrips() {
|
||
let header = Header {
|
||
msg_type: MsgType::Video,
|
||
version: PROTOCOL_VERSION,
|
||
sender_id: 0xDEAD_BEEF,
|
||
frame_id: 12345,
|
||
frag_index: 3,
|
||
frag_count: 9,
|
||
total_len: 20_000,
|
||
width: 640,
|
||
height: 480,
|
||
};
|
||
|
||
let mut buf = Vec::new();
|
||
encode_header_into(&header, &mut buf);
|
||
|
||
assert_eq!(buf.len(), HEADER_LEN);
|
||
assert_eq!(decode_header(&buf), Some(header));
|
||
}
|
||
|
||
#[test]
|
||
fn fragment_roundtrips_and_is_multiple_sources() {
|
||
let jpeg: Vec<u8> = (0..5000u32).map(|i| (i % 251) as u8).collect();
|
||
let frags = fragment_frame(0xAAAA_BBBB, 7, &jpeg, 640, 480);
|
||
|
||
assert_eq!(frags.len(), 4);
|
||
|
||
for f in &frags {
|
||
assert_eq!(f.header.sender_id, 0xAAAA_BBBB);
|
||
let decoded = Fragment::decode(&f.encode()).expect("decodes");
|
||
assert_eq!(&decoded, f);
|
||
}
|
||
}
|
||
|
||
#[test]
|
||
fn reassembles_and_separates_senders() {
|
||
let jpeg: Vec<u8> = (0..5000u32).map(|i| (i % 251) as u8).collect();
|
||
|
||
// 模拟裁判终端：两个发送方各自的重组器。
|
||
let mut a = FrameAssembler::new(std::time::Duration::from_secs(2));
|
||
let mut b = FrameAssembler::new(std::time::Duration::from_secs(2));
|
||
|
||
let mut frame_a = None;
|
||
let mut frame_b = None;
|
||
|
||
for f in fragment_frame(1, 1, &jpeg, 640, 480) {
|
||
if let Ok(frame) = a.push(f) {
|
||
frame_a = Some(frame);
|
||
}
|
||
}
|
||
for f in fragment_frame(2, 1, &jpeg, 320, 240) {
|
||
if let Ok(frame) = b.push(f) {
|
||
frame_b = Some(frame);
|
||
}
|
||
}
|
||
|
||
let fa = frame_a.expect("sender 1 completed");
|
||
let fb = frame_b.expect("sender 2 completed");
|
||
|
||
assert_eq!(fa.sender_id, 1);
|
||
assert_eq!(fb.sender_id, 2);
|
||
assert_eq!(fa.data, jpeg);
|
||
assert_eq!(fb.data, jpeg);
|
||
assert_eq!((fa.width, fa.height), (640, 480));
|
||
assert_eq!((fb.width, fb.height), (320, 240));
|
||
}
|
||
|
||
#[test]
|
||
fn a_gap_means_no_frame_until_the_next_one() {
|
||
let jpeg = vec![0xAAu8; 5000];
|
||
let frags = fragment_frame(1, 1, &jpeg, 640, 480);
|
||
let mut asm = FrameAssembler::new(std::time::Duration::from_secs(2));
|
||
|
||
// 丢掉第 2 个分片。
|
||
for f in frags.iter().filter(|f| f.header.frag_index != 2) {
|
||
assert!(asm.push(f.clone()).is_err());
|
||
}
|
||
assert!(asm.is_in_progress(), "still waiting on the missing fragment");
|
||
|
||
// 新的帧取代没收齐的那张，而不是去补洞。
|
||
let next = fragment_frame(1, 2, &jpeg, 640, 480);
|
||
assert!(asm.push(next[0].clone()).is_err());
|
||
|
||
let mut got = None;
|
||
for f in next.iter().skip(1) {
|
||
if let Ok(frame) = asm.push(f.clone()) {
|
||
got = Some(frame);
|
||
}
|
||
}
|
||
assert_eq!(got.expect("second frame completes").frame_id, 2);
|
||
}
|
||
|
||
#[test]
|
||
fn expires_a_stalled_frame() {
|
||
let jpeg = vec![0x11u8; 3000];
|
||
let frags = fragment_frame(9, 1, &jpeg, 640, 480);
|
||
let mut asm = FrameAssembler::new(std::time::Duration::from_millis(1));
|
||
|
||
let _ = asm.push(frags[0].clone());
|
||
assert!(asm.is_in_progress());
|
||
std::thread::sleep(std::time::Duration::from_millis(5));
|
||
|
||
assert_eq!(asm.expire(), Some(DropReason::TimedOut));
|
||
assert!(!asm.is_in_progress());
|
||
}
|
||
|
||
#[test]
|
||
fn rejects_foreign_and_malformed_datagrams() {
|
||
assert!(decode_header(b"").is_none());
|
||
assert!(decode_header(&[0u8; 8]).is_none());
|
||
// 长度够但魔数不对。
|
||
assert!(decode_header(&[0u8; HEADER_LEN]).is_none());
|
||
|
||
// 正确的魔数但版本不对：宁可拒绝，也不能按错误布局解析。
|
||
let header = Header {
|
||
sender_id: 1,
|
||
frag_count: 1,
|
||
total_len: 10,
|
||
..Default::default()
|
||
};
|
||
let mut buf = Vec::new();
|
||
encode_header_into(&header, &mut buf);
|
||
buf[5] = 99; // version 低字节
|
||
assert!(decode_header(&buf).is_none(), "wrong version must be refused");
|
||
|
||
let jpeg = vec![0u8; 100];
|
||
|
||
// frag_index >= frag_count 自相矛盾。
|
||
let mut bad = fragment_frame(1, 1, &jpeg, 64, 48).remove(0).encode();
|
||
bad[15..17].copy_from_slice(&9u16.to_le_bytes());
|
||
assert!(Fragment::decode(&bad).is_none());
|
||
|
||
// 载荷比声明的总长还大：损坏。
|
||
let mut over = fragment_frame(1, 1, &jpeg, 64, 48).remove(0).encode();
|
||
over[19..23].copy_from_slice(&4u32.to_le_bytes());
|
||
assert!(Fragment::decode(&over).is_none());
|
||
}
|
||
|
||
#[test]
|
||
fn subscribe_and_hello_roundtrip() {
|
||
let sub = Subscribe {
|
||
role: SubscribeRole::Referee,
|
||
client_id: 0x1234_5678,
|
||
};
|
||
assert_eq!(Subscribe::decode(&sub.encode()), Some(sub));
|
||
|
||
let hello = Hello {
|
||
sender_id: 0x8765_4321,
|
||
firmware: 1,
|
||
};
|
||
assert_eq!(Hello::decode(&hello.encode(MsgType::Hello)), Some(hello));
|
||
assert_eq!(
|
||
Hello::decode(&hello.encode(MsgType::Heartbeat)),
|
||
Some(hello)
|
||
);
|
||
}
|
||
|
||
#[test]
|
||
fn armor_health_roundtrips() {
|
||
// 现在有实现了（armor/ 里的 ESP32-C3 固件在发它），布局必须保持可往返。
|
||
let hp = ArmorHealth {
|
||
team: 1,
|
||
index: 2,
|
||
hp: 420,
|
||
};
|
||
|
||
let encoded = hp.encode();
|
||
assert_eq!(ArmorHealth::decode(&encoded), Some(hp));
|
||
}
|
||
|
||
#[test]
|
||
fn armor_health_can_carry_the_sending_board_id() {
|
||
let hp = ArmorHealth {
|
||
team: 0,
|
||
index: 3,
|
||
hp: 97,
|
||
};
|
||
|
||
let encoded = hp.encode_from(0x1234_5678);
|
||
assert_eq!(
|
||
decode_header(&encoded).map(|h| h.sender_id),
|
||
Some(0x1234_5678),
|
||
"抓包时要能分辨是哪块装甲板发的"
|
||
);
|
||
assert_eq!(ArmorHealth::decode(&encoded), Some(hp));
|
||
|
||
// 帧头 id 不影响帧体，也不该影响旧调用方的行为。
|
||
assert_eq!(ArmorHealth::decode(&hp.encode()), Some(hp));
|
||
assert_eq!(decode_header(&hp.encode()).map(|h| h.sender_id), Some(0));
|
||
}
|
||
|
||
#[test]
|
||
fn armor_health_decode_from_returns_the_board_id() {
|
||
// 客户端要列出"场上有哪几块装甲板"，只能靠帧头里这个 id：team 只有 0/1，
|
||
// 同队的两块板子在 team 上无法区分。
|
||
let hp = ArmorHealth {
|
||
team: 1,
|
||
index: 1,
|
||
hp: 100,
|
||
};
|
||
|
||
assert_eq!(
|
||
ArmorHealth::decode_from(&hp.encode_from(0xABCD_1234)),
|
||
Some((0xABCD_1234, hp))
|
||
);
|
||
|
||
// decode 是 decode_from 丢掉信封的那一半，两者必须一致。
|
||
assert_eq!(
|
||
ArmorHealth::decode(&hp.encode_from(0xABCD_1234)),
|
||
ArmorHealth::decode_from(&hp.encode_from(0xABCD_1234)).map(|(_, h)| h)
|
||
);
|
||
|
||
// 不是装甲板上报就不该解出来。
|
||
assert!(ArmorHealth::decode_from(&[0u8; 8]).is_none());
|
||
}
|
||
|
||
#[test]
|
||
fn ids_are_stable_and_never_zero() {
|
||
let mac = [0x90, 0x70, 0x69, 0xF8, 0xF2, 0x14];
|
||
let a = id_from_mac(&mac);
|
||
let b = id_from_mac(&mac);
|
||
|
||
assert_eq!(a, b, "the same MAC must always give the same id");
|
||
assert_ne!(a, 0, "0 is reserved for the referee terminal");
|
||
|
||
// 不同 MAC 不应撞车。
|
||
assert_ne!(a, id_from_mac(&[1, 2, 3, 4, 5, 6]));
|
||
}
|
||
|
||
#[test]
|
||
fn a_duplicate_cannot_complete_a_frame_with_a_hole() {
|
||
// UDP 会重复投递。一个丢失的分片若能被另一片的副本凑数，接收端就会以为
|
||
// 收齐了，交出一张中间是 0 的坏 JPEG——而且帧率正常、没有任何报错线索。
|
||
let jpeg: Vec<u8> = (0..5000u32).map(|i| (i % 251) as u8).collect();
|
||
let frags = fragment_frame(1, 1, &jpeg, 640, 480);
|
||
let mut asm = FrameAssembler::new(std::time::Duration::from_secs(2));
|
||
|
||
// 丢掉第 2 片，然后把第 0 片连送三次来灌 seen。
|
||
for f in frags.iter().filter(|f| f.header.frag_index != 2) {
|
||
let _ = asm.push(f.clone());
|
||
}
|
||
for _ in 0..3 {
|
||
assert!(
|
||
asm.push(frags[0].clone()).is_err(),
|
||
"a duplicate must not be allowed to complete the frame"
|
||
);
|
||
}
|
||
|
||
assert!(asm.is_in_progress(), "the hole must keep it incomplete");
|
||
|
||
// 补上真正的第 2 片，才在这一片上收齐——所以结果要从这一次的返回值拿。
|
||
let completed = asm.push(frags[2].clone());
|
||
|
||
assert_eq!(
|
||
completed.expect("completes once the hole is filled").data,
|
||
jpeg,
|
||
);
|
||
}
|
||
|
||
#[test]
|
||
fn a_gap_that_is_never_filled_never_yields_a_frame() {
|
||
let jpeg = vec![0xAAu8; 5000];
|
||
let frags = fragment_frame(1, 1, &jpeg, 640, 480);
|
||
let mut asm = FrameAssembler::new(std::time::Duration::from_secs(2));
|
||
|
||
// 反复送同一片，永远凑不齐——接收端必须一直等着，而不是交出半张图。
|
||
for _ in 0..50 {
|
||
assert!(asm.push(frags[0].clone()).is_err());
|
||
}
|
||
assert!(asm.is_in_progress(), "50 duplicates must not complete a frame");
|
||
assert_eq!(asm.sender_id(), 1, "and it must still be the same camera");
|
||
}
|
||
|
||
#[test]
|
||
fn empty_jpeg_still_produces_one_fragment() {
|
||
let frags = fragment_frame(1, 1, &[], 64, 48);
|
||
assert_eq!(frags.len(), 1);
|
||
assert!(frags[0].payload.is_empty());
|
||
}
|
||
|
||
#[test]
|
||
fn datagrams_stay_under_a_typical_mtu() {
|
||
let jpeg = vec![0u8; 100_000];
|
||
for f in fragment_frame(1, 1, &jpeg, 800, 600) {
|
||
assert!(f.encode().len() <= HEADER_LEN + MAX_FRAGMENT_PAYLOAD);
|
||
assert!(f.encode().len() < 1500, "must fit an Ethernet MTU");
|
||
}
|
||
}
|
||
} |