ESP32-S3 摄像头图传:固件 + Slint 客户端 + 共享协议

两块 ESP32-S3-WROOM-1-N16R8 图传板(DVP 接口接 OV2640/OV3660/OV5640),
向队伍客户端与裁判终端推 UDP 图传。固件是跑在 ESP-IDF 上的 Rust,用官方
esp32-camera 组件采集;客户端是 Rust + Slint 桌面程序。

三个 crate 各自独立构建:
  .            firmware,交叉编译到 Xtensa
  protocol/    线路格式与帧重组,两端共用,无 ESP-IDF 依赖,有单元测试
  client/      Slint 桌面客户端

视频走 UDP 而非 TCP:实测传感器产出 22 fps,TCP MJPEG 只送出 1.9 fps,
且只占用链路能力的 1%——链路是空的,帧堵在队头阻塞后面等一个实时画面
等不起的重传计时器。改为独立数据报、不重传、缺片丢整帧后达到 20.8 fps。

发现是双向的:板子每 2 秒广播 Hello,客户端广播 Subscribe,板子据此定向
推流。板子身份由 MAC 派生,重启不变,所以客户端可以把绑定存进配置。

裁判端同时收两块板子(每个发送方一个重组器),队伍端绑定其中一块并只显示
自己那一边的血量。装甲板接口已预留(ArmorSource trait + ARMOR_PORT),
当前无数据源,界面显示占位符。

vendor 了 esp32-camera 2.1.7:组件注册表的版本求解器找不到它,且需要打一处
补丁(见 components/esp32-camera/target/esp32s3/ll_cam.c)。

.gitignore 三条 target/ 都锚定在根目录,因为 protocol/ 与 client/ 各自也是
独立 crate;而 components/esp32-camera/target/ 是上游的按芯片分目录,里面是
源码,必须保留。
This commit is contained in:
Horldsence
2026-10-05 18:22:19 +08:00
commit 4f893ce140
134 changed files with 41128 additions and 0 deletions
+310
View File
@@ -0,0 +1,310 @@
//! 绑定配置:这个客户端是什么身份,以及看哪块板子。
//!
//! 队伍的"选定后不可更改"落在 [`Config::camera_id`] 上:首次运行时从发现列表里选,
//! 结果写进配置文件,之后直接读,不再询问。
//!
//! 用 TOML 而不是 JSON,因为这个文件是**给人看的**——现场排查时要能一眼看懂改了什么。
//! 而它只有十来个字段,不值得为它引入整个序列化框架。
use std::path::{Path, PathBuf};
use anyhow::{Context, Result};
/// 客户端身份。
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum Role {
/// 队伍终端:一块画面 + 自己那一边的血量。
Team,
/// 裁判终端:两块画面 + 双方血量。
Referee,
}
impl Role {
pub fn as_str(self) -> &'static str {
match self {
Role::Team => "team",
Role::Referee => "referee",
}
}
pub fn parse(s: &str) -> Option<Self> {
match s {
"team" => Some(Role::Team),
"referee" => Some(Role::Referee),
_ => None,
}
}
pub fn to_protocol(self) -> protocol::SubscribeRole {
match self {
Role::Team => protocol::SubscribeRole::Team,
Role::Referee => protocol::SubscribeRole::Referee,
}
}
}
/// 磁盘上的配置。
#[derive(Debug, Clone)]
pub struct Config {
/// 客户端身份。
pub role: Role,
/// 队伍端:自己属于哪一边,0 = 队伍一,1 = 队伍二。
///
/// 决定血量栏显示哪一边。裁判端两个都要,所以忽略这个值。
pub team: u8,
/// 绑定的板子 id(MAC 派生,重启不变)。
///
/// `None` 表示还没选过,首次运行会在界面上询问。
pub camera_id: Option<u32>,
/// 板子的 IP。DHCP 可能变,所以这只是发现阶段的一个提示。
pub camera_addr: Option<std::net::Ipv4Addr>,
}
impl Default for Config {
/// 默认当裁判机:它不需要任何配置就能用,是现场最省事的起点。
fn default() -> Self {
Self {
role: Role::Referee,
team: 0,
camera_id: None,
camera_addr: None,
}
}
}
impl Config {
/// 从磁盘读取。
///
/// 文件不存在时返回默认配置,这是正常路径——首次运行本来就没有配置。
pub fn load(path: &Path) -> Result<Self> {
if !path.exists() {
log::info!("no config at {}; starting with defaults", path.display());
return Ok(Self::default());
}
let text = std::fs::read_to_string(path)
.with_context(|| format!("reading {}", path.display()))?;
Self::parse(&text).with_context(|| format!("parsing {}", path.display()))
}
/// 写入磁盘。
///
/// 手写序列化而不是引入 toml crate:这个文件只有十来个字段,且格式必须与
/// [`Config::parse`] 严格对称——两份代码比一个依赖更容易在 review 时看出不匹配。
pub fn save(&self, path: &Path) -> Result<()> {
if let Some(dir) = path.parent() {
std::fs::create_dir_all(dir)
.with_context(|| format!("creating {}", dir.display()))?;
}
let mut out = String::new();
out.push_str("# cam-client 配置。\n");
out.push_str("#\n");
out.push_str("# role : team | referee\n");
out.push_str("# team = 一路画面 + 自己那一边的血量,需要绑定一块板子\n");
out.push_str("# referee = 两路画面 + 双方血量,不需要绑定\n");
out.push_str("# team : 0 = 队伍一, 1 = 队伍二(仅 team 端用)\n");
out.push_str("#\n");
out.push_str("# 绑定之后不要再改 camera_id:队伍端按它选画面,改了就换了视角。\n");
out.push_str("# camera_id 由板子的 MAC 派生,重启后不变,所以这个绑定是稳的。\n\n");
out.push_str(&format!("role = \"{}\"\n", self.role.as_str()));
out.push_str(&format!("team = {}\n", self.team));
if let Some(id) = self.camera_id {
out.push_str(&format!("camera_id = \"0x{id:08X}\"\n"));
}
if let Some(addr) = self.camera_addr {
out.push_str(&format!("camera_addr = \"{addr}\"\n"));
}
std::fs::write(path, out).with_context(|| format!("writing {}", path.display()))?;
log::info!("saved config to {}", path.display());
Ok(())
}
/// 解析 TOML 的一个子集。
///
/// 只支持 `key = value`、字符串、`整数`、十六进制字符串和注释——恰好是这个文件
/// 需要的全部,遇到不认识的形式会明确报错而不是静默忽略,因为静默忽略会让一个
/// 打错的配置看起来像是生效了。
fn parse(text: &str) -> Result<Self> {
let mut config = Config::default();
let mut role_seen = false;
for (lineno, raw) in text.lines().enumerate() {
let line = strip_comment(raw).trim();
if line.is_empty() {
continue;
}
let Some((key, value)) = line.split_once('=') else {
anyhow::bail!("line {}: expected `key = value`, got {line:?}", lineno + 1);
};
let key = key.trim();
let value = value.trim().trim_matches('"');
match key {
"role" => {
config.role = Role::parse(value).ok_or_else(|| {
anyhow::anyhow!(
"line {}: role must be `team` or `referee`, got {value:?}",
lineno + 1
)
})?;
role_seen = true;
}
"team" => {
config.team = value
.parse::<u8>()
.with_context(|| format!("line {}: team must be 0 or 1", lineno + 1))?;
}
"camera_id" => {
config.camera_id = Some(parse_id(value).with_context(|| {
format!("line {}: camera_id must look like 0x1A2B3C4D", lineno + 1)
})?);
}
"camera_addr" => {
config.camera_addr = Some(value.parse().with_context(|| {
format!("line {}: camera_addr must be an IPv4 address", lineno + 1)
})?);
}
_ => {
anyhow::bail!(
"line {}: unknown key {key:?}; refusing to guess",
lineno + 1
);
}
}
}
if !role_seen {
// 缺 role 时按裁判端处理:它是唯一无需配置就能工作的模式。
config.role = Role::Referee;
}
Ok(config)
}
}
/// 去掉 `#` 之后的内容。
fn strip_comment(line: &str) -> &str {
match line.find('#') {
Some(at) => &line[..at],
None => line,
}
}
/// 解析 `0x1A2B3C4D` 或十进制形式的板子 id。
fn parse_id(text: &str) -> Result<u32> {
let text = text.trim();
if let Some(hex) = text.strip_prefix("0x").or_else(|| text.strip_prefix("0X")) {
u32::from_str_radix(hex, 16).with_context(|| format!("{text:?} is not hex"))
} else {
text.parse::<u32>()
.with_context(|| format!("{text:?} is not a number"))
}
}
/// 默认配置文件位置。
///
/// 优先 `$XDG_CONFIG_HOME` 或 `~/.config`,与 Linux 上其它工具一致;macOS 上用同一个
/// 路径而不是 `~/Library/Application Support`,是为了让两台机器(裁判机和队伍机可能
/// 是不同系统)的配置结构一致。
pub fn default_config_path() -> PathBuf {
let base = std::env::var_os("XDG_CONFIG_HOME")
.map(PathBuf::from)
.filter(|p| !p.as_os_str().is_empty())
.unwrap_or_else(|| {
let home = std::env::var_os("HOME").map(PathBuf::from).unwrap_or_else(|| {
PathBuf::from(".")
});
home.join(".config")
});
base.join("cam-client").join("config.toml")
}
#[cfg(test)]
mod tests {
use super::*;
#[test]
fn parses_a_team_config() {
let config = Config::parse(
r#"
# 队伍一的机器
role = "team"
team = 1
camera_id = "0x3D4D8A9A"
"#,
)
.expect("parses");
assert_eq!(config.role, Role::Team);
assert_eq!(config.team, 1);
assert_eq!(config.camera_id, Some(0x3D4D_8A9A));
assert!(config.is_bound());
}
#[test]
fn referee_needs_no_binding() {
let config = Config::parse("role = \"referee\"").expect("parses");
assert_eq!(config.role, Role::Referee);
assert!(config.is_bound(), "the referee watches everything");
}
#[test]
fn a_team_without_a_camera_is_unbound() {
let config = Config::parse("role = \"team\"\nteam = 0").expect("parses");
assert_eq!(config.role, Role::Team);
assert!(!config.is_bound(), "must ask which camera to watch");
}
#[test]
fn roundtrips_through_save_and_load() {
let mut dir = std::env::temp_dir();
dir.push(format!("cam-client-test-{}", std::process::id()));
let path = dir.join("nested").join("config.toml");
let config = Config {
role: Role::Team,
team: 1,
camera_id: Some(0x1234_5678),
camera_addr: Some("192.168.1.50".parse().unwrap()),
};
config.save(&path).expect("saves");
let loaded = Config::load(&path).expect("loads");
assert_eq!(loaded.role, config.role);
assert_eq!(loaded.team, config.team);
assert_eq!(loaded.camera_id, config.camera_id);
assert_eq!(loaded.camera_addr, config.camera_addr);
let _ = std::fs::remove_dir_all(&dir);
}
#[test]
fn a_missing_file_is_the_default_not_an_error() {
let config = Config::load(Path::new("/nonexistent/cam-client/config.toml"))
.expect("a missing config is a normal first run");
assert_eq!(config.role, Role::Referee);
}
#[test]
fn refuses_garbage_instead_of_ignoring_it() {
// 静默忽略一个打错的 key,会让配置看起来生效了。
assert!(Config::parse("rol = \"team\"").is_err(), "unknown key");
assert!(Config::parse("role = \"camera\"").is_err(), "bad role value");
assert!(Config::parse("role = \"team\"\nteam = nine").is_err(), "bad team");
assert!(Config::parse("just some text").is_err(), "not key = value");
}
}
+628
View File
@@ -0,0 +1,628 @@
//! ESP32 图传查看器。
//!
//! 两种身份,由 [`Role`] 决定:
//!
//! - **队伍端** —— 一块画面 + 自己那一边的血量。需要绑定一块板子(选定后不可更改)。
//! - **裁判端** —— 两块画面左右并排 + 双方血量。不需要绑定。
//!
//! 用法:
//!
//! ```text
//! cam-client 默认裁判端
//! cam-client --role team --team 1 队伍一
//! cam-client --camera-id 0x3D4D8A9A 绑定到某块板子
//! ```
//!
//! 绑定与身份存在配置文件里(默认 `~/.config/cam-client/config.toml`)。首次运行
//! 队伍端会列出发现的摄像头,选定后写入,之后不再询问。
//!
//! # 为什么血量是占位符
//!
//! 装甲板接口已经预留([`receiver::ArmorSource`] 与 `protocol::ArmorHealth`),但
//! 现在没有实现也没有数据源。界面显示占位符而不是 0——0 看起来像真值,现场看到会
//! 以为对方已经没血了。
mod config;
mod receiver;
use std::error::Error;
use std::net::Ipv4Addr;
use std::path::PathBuf;
use std::sync::{Arc, Mutex};
use std::time::{Duration, Instant};
use config::{Config, Role};
use receiver::{
decode_frame, ArmorSource, CameraInfo, DecodedFrame, NoArmorYet, Receiver,
SourceStats, ARMOR_BOARDS,
};
slint::include_modules!();
const USAGE: &str = "\
用法: cam-client [选项]
--role <team|referee> 身份。默认 referee(无需配置即可看两路)
--team <0|1> 自己属于哪一边,仅 team 端用
--camera-id <0xID> 绑定到某块板子(MAC 派生,重启不变)
--camera <IP> 向哪块板子发订阅(默认 255.255.255.255 广播)
--config <路径> 配置文件位置
--list 列出发现的摄像头后退出
-h, --help 显示这段帮助
";
/// 事件循环轮询新数据的间隔。
///
/// 不是每帧一次:显示不需要每秒重绘 22 次,更快只是让事件循环和接收路径抢 CPU。
const UI_POLL: Duration = Duration::from_millis(33);
/// 帧率统计窗口。
///
/// 按计数除以墙钟时间,而不是对帧间隔取平均。一次轮询之间可能完成多帧,所以平均间隔
/// 量的是**显示**的刷新快慢而不是**相机**的出帧快慢——那会把上限 22fps 的传感器
/// 报成 36fps。一个笃定的错误数字比没有数字更糟。
const FPS_WINDOW: Duration = Duration::from_secs(2);
/// 某一路的显示状态。
struct View {
/// 显示哪块板子(`None` = 还没有画面)。
sender_id: Option<u32>,
fps: f32,
window_started: Instant,
window_frames: u64,
reported: bool,
}
impl View {
fn new() -> Self {
Self {
sender_id: None,
fps: 0.0,
window_started: Instant::now(),
window_frames: 0,
reported: false,
}
}
/// 累计窗口内的帧数,算出帧率。
fn update_fps(&mut self, stats: &SourceStats) {
let window = self.window_started.elapsed();
if window < FPS_WINDOW {
return;
}
self.fps = (stats.frames_shown - self.window_frames) as f32 / window.as_secs_f32();
self.window_frames = stats.frames_shown;
self.window_started = Instant::now();
}
}
/// 解码后的画面缓存,供事件循环取用。
///
/// 解码线程只处理纯数据(`Vec<u8>` + 尺寸),`slint::Image` 的构造留在事件循环,
/// 因为 `slint::Image` 不是 `Send`。
type DecodedSlots = Arc<Mutex<Vec<(u32, DecodedFrame)>>>;
fn main() -> Result<(), Box<dyn Error>> {
env_logger::Builder::from_env(env_logger::Env::default().default_filter_or("info")).init();
let args = Args::parse(std::env::args().skip(1))?;
if args.help {
print!("{USAGE}");
return Ok(());
}
let config_path = args.config.clone().unwrap_or_else(config::default_config_path);
// 命令行覆盖配置文件,这样现场可以临时当另一台机器用而不用改文件。
let mut config = Config::load(&config_path)?;
if let Some(role) = args.role {
config.role = role;
}
if let Some(team) = args.team {
config.team = team;
}
if let Some(id) = args.camera_id {
config.camera_id = Some(id);
}
log::info!(
"role={} team={} camera_id={:?} config={}",
config.role.as_str(),
config.team,
config.camera_id,
config_path.display()
);
let rx = Arc::new(Receiver::start(
args.camera,
config.role.to_protocol(),
)?);
let discovery = rx.discovery();
let frames = rx.frames();
if args.list {
// 等几个广播周期,让列表有内容。
let socket = bind_socket()?;
let for_list = Arc::clone(&rx);
let handle = std::thread::spawn(move || for_list.run(socket));
std::thread::sleep(Duration::from_secs(5));
let cameras = discovery.online_cameras();
if cameras.is_empty() {
println!("没有发现摄像头。板子每 2 秒广播一次,再等等或检查网络。");
} else {
println!("发现 {} 块板子:", cameras.len());
for c in cameras {
println!(" 0x{:08X} {} (id 与 MAC 派生,重启不变)", c.id, c.addr);
}
}
rx.shutdown();
let _ = handle.join();
return Ok(());
}
// 血量目前没有数据源。接口已经就位,装甲板接上后换掉这一行即可。
let armor: Arc<dyn ArmorSource> = Arc::new(NoArmorYet);
let decoded: DecodedSlots = Arc::new(Mutex::new(Vec::new()));
// 接收线程。
{
let rx = Arc::clone(&rx);
std::thread::Builder::new()
.name("udp-reader".into())
.spawn(move || {
let Ok(socket) = bind_socket() else {
log::error!("could not bind the socket for the receive loop");
return;
};
rx.run(socket);
})?;
}
// 解码线程:一张 VGA 的 JPEG 要几毫秒,放在收包线程里会挤进数据报之间,造成
// 链路上本不存在的丢帧。
{
let frames = Arc::clone(&frames);
let decoded = Arc::clone(&decoded);
std::thread::Builder::new()
.name("jpeg-decode".into())
.spawn(move || {
loop {
if frames.pending_senders().is_empty() {
std::thread::sleep(Duration::from_millis(5));
continue;
}
for sender_id in frames.pending_senders() {
let Some(frame) = frames.take(sender_id) else {
continue;
};
match decode_frame(&frame.data) {
Some(pixels) => {
let mut slot = decoded.lock().expect("decoded mutex poisoned");
// 同一路只保留最新一张,避免解码已经过时的画面。
slot.retain(|(id, _)| *id != sender_id);
slot.push((sender_id, pixels));
}
None => log::warn!(
"frame {} from 0x{:08X} is not a decodable JPEG ({} bytes)",
frame.frame_id,
frame.sender_id,
frame.data.len()
),
}
}
}
})?;
}
let timer = slint::Timer::default();
// 每个身份一个闭包,而不是一个闭包里再 match 身份。
//
// 两种身份是两个不同的组件类型(TeamApp / RefereeApp),没有共同的超类可以
// 在一个闭包里统一调用 set_frame/set_frame_a。写成两个分支后每边都拿到具体
// 类型,属性赋值直接调用,不再需要 match 身份——那种 match 会让编译器在每个
// 赋值点都检查一遍身份,而它本来就是编译期就定死的。
match config.role {
Role::Team => {
let ui = TeamApp::new()?;
ui.show()?;
let weak = ui.as_weak();
let discovery = Arc::clone(&discovery);
let decoded = Arc::clone(&decoded);
let armor = Arc::clone(&armor);
let config_path = config_path.clone();
let team = config.team;
let bound = config.camera_id;
let mut view = View::new();
// 绑定回调与刷新定时器都要读这份列表,所以用锁共享,而不是让其中一个
// 把 `choices` 移走。
let choices: Arc<Mutex<Vec<CameraChoice>>> = Arc::new(Mutex::new(Vec::new()));
// 绑定完成时写配置。"选定不可更改"就落在这里:此后配置里已有
// camera_id,界面不再出现选择页。
let choices_for_pick = Arc::clone(&choices);
let pick_handle = ui.on_pick(move |index| {
let Ok(index) = usize::try_from(index) else {
return;
};
let picked = choices_for_pick
.lock()
.expect("choices mutex poisoned")
.get(index)
.map(|c| c.id_text.to_string());
let Some(id_text) = picked else {
return;
};
let Some(id) = parse_id(&id_text).ok() else {
log::error!("could not read back the selected id {id_text:?}");
return;
};
let saved = Config {
role: Role::Team,
team,
camera_id: Some(id),
camera_addr: None,
};
if let Err(err) = saved.save(&config_path) {
log::error!("could not save the binding: {err}");
log::error!("画面仍会显示,但下次启动要重新选择");
} else {
log::info!("bound to camera 0x{id:08X}; saved in the config");
}
});
let choices_for_poll = Arc::clone(&choices);
timer.start(slint::TimerMode::Repeated, UI_POLL, move || {
let Some(ui) = weak.upgrade() else {
return;
};
let cameras = discovery.online_cameras();
// 未绑定:只显示选择界面。
let Some(bound) = bound else {
let updated: Vec<CameraChoice> = cameras
.iter()
.map(|c| CameraChoice {
id_text: format!("0x{:08X}", c.id).into(),
addr_text: format!("{} ({:?})", c.addr, c.silence()).into(),
online: c.is_online(),
})
.collect();
*choices_for_poll.lock().expect("choices mutex poisoned") = updated.clone();
ui.set_choices(std::rc::Rc::new(slint::VecModel::from(updated)).into());
ui.set_choosing(true);
return;
};
ui.set_choosing(false);
if view.sender_id != Some(bound) {
view = View::new();
view.sender_id = Some(bound);
}
if let Some(image) = take_decoded(&decoded, bound) {
ui.set_frame(image);
}
let stats = discovery.stats(bound);
view.update_fps(&stats);
if !view.reported && stats.frames_shown > 0 {
log_diagnostics(&stats, bound);
view.reported = true;
}
ui.set_stats(render_camera_stats(&cameras, bound, &stats, view.fps));
ui.set_health(render_health(team, armor.as_ref()));
});
// 绑定回调必须活到窗口关闭。CallbackHandle 是 Copy,但 Drop 会注销
// 回调,所以必须真的把它泄漏掉——之前用 `mem::forget` 编译器提示它对
// Copy 类型无效,那确实无效:第一次点击后回调就被注销了,第二次点击
// 什么也不会发生。
std::boxed::Box::leak(Box::new(pick_handle));
}
Role::Referee => {
let ui = RefereeApp::new()?;
ui.show()?;
let weak = ui.as_weak();
let discovery = Arc::clone(&discovery);
let decoded = Arc::clone(&decoded);
let armor = Arc::clone(&armor);
let mut views: Vec<View> = (0..2).map(|_| View::new()).collect();
timer.start(slint::TimerMode::Repeated, UI_POLL, move || {
let Some(ui) = weak.upgrade() else {
return;
};
let cameras = discovery.online_cameras();
// 两路按 id 顺序填槽。槽位保持稳定(不会因为某块板子临时离线就重新
// 洗牌),否则画面会左右跳动。
let ids: Vec<u32> = cameras.iter().map(|c| c.id).take(2).collect();
for (i, view) in views.iter_mut().enumerate() {
let wanted = ids.get(i).copied();
if view.sender_id != wanted {
*view = View::new();
view.sender_id = wanted;
}
}
for (i, view) in views.iter_mut().enumerate() {
let Some(sender_id) = view.sender_id else {
continue;
};
if let Some(image) = take_decoded(&decoded, sender_id) {
if i == 0 {
ui.set_frame_a(image);
} else {
ui.set_frame_b(image);
}
}
let stats = discovery.stats(sender_id);
view.update_fps(&stats);
if !view.reported && stats.frames_shown > 0 {
log_diagnostics(&stats, sender_id);
view.reported = true;
}
let view_stats = render_camera_stats(&cameras, sender_id, &stats, view.fps);
if i == 0 {
ui.set_stats_a(view_stats);
} else {
ui.set_stats_b(view_stats);
}
}
ui.set_health_a(render_health(0, armor.as_ref()));
ui.set_health_b(render_health(1, armor.as_ref()));
});
}
}
// Timer 一旦被 drop 就停止,所以要活到进程结束。
let _timer = Box::leak(Box::new(timer));
slint::run_event_loop()?;
rx.shutdown();
Ok(())
}
/// 从解码缓存里取出某一路的最新画面。
///
/// 解码线程只处理纯数据,`slint::Image` 的构造留在这里——`slint::Image` 不是
/// `Send`,跨不了线程。
fn take_decoded(
decoded: &DecodedSlots,
sender_id: u32,
) -> Option<slint::Image> {
let mut slot = decoded.lock().expect("decoded mutex poisoned");
let pos = slot.iter().position(|(id, _)| *id == sender_id)?;
let (_, pixels) = slot.remove(pos);
let buffer = slint::SharedPixelBuffer::<slint::Rgba8Pixel>::clone_from_slice(
&pixels.rgba,
pixels.width,
pixels.height,
);
Some(slint::Image::from_rgba8(buffer))
}
/// 命令行参数。
struct Args {
role: Option<Role>,
team: Option<u8>,
camera_id: Option<u32>,
camera: Ipv4Addr,
config: Option<PathBuf>,
list: bool,
help: bool,
}
impl Args {
fn parse(argv: impl Iterator<Item = String>) -> Result<Self, Box<dyn Error>> {
let mut args = Args {
role: None,
team: None,
camera_id: None,
camera: protocol::BROADCAST_ADDR,
config: None,
list: false,
help: false,
};
let mut it = argv.peekable();
while let Some(arg) = it.next() {
match arg.as_str() {
"-h" | "--help" => args.help = true,
"--list" => args.list = true,
"--role" => {
let v = it.next().ok_or("--role needs a value")?;
args.role =
Some(Role::parse(&v).ok_or("role must be `team` or `referee`")?);
}
"--team" => {
let v = it.next().ok_or("--team needs a value")?;
args.team = Some(v.parse().map_err(|_| "--team must be 0 or 1")?);
}
"--camera-id" => {
let v = it.next().ok_or("--camera-id needs a value")?;
args.camera_id = Some(parse_id(&v)?);
}
"--camera" => {
let v = it.next().ok_or("--camera needs a value")?;
args.camera = v
.parse()
.map_err(|_| "--camera must be an IPv4 address")?;
}
"--config" => {
args.config = Some(PathBuf::from(it.next().ok_or("--config needs a path")?));
}
other => {
return Err(format!("unknown argument {other:?}\n\n{USAGE}").into());
}
}
}
Ok(args)
}
}
/// 绑定 UDP 接收 socket。
///
/// 固定端口(而非临时端口)是必须的:板子从订阅的**源端口**学到推流目标,而这个
/// socket 同时承担收包。
fn bind_socket() -> std::io::Result<std::net::UdpSocket> {
let socket = std::net::UdpSocket::bind(std::net::SocketAddrV4::new(
Ipv4Addr::UNSPECIFIED,
protocol::DEFAULT_PORT,
))?;
socket.set_read_timeout(Some(Duration::from_millis(100)))?;
// 广播是发现的前提:板子靠它宣告存在,客户端靠它发起订阅。
let _ = socket.set_broadcast(true);
Ok(socket)
}
/// 解析 `0x1A2B3C4D` 或十进制形式的板子 id。
fn parse_id(text: &str) -> Result<u32, Box<dyn Error>> {
let text = text.trim();
if let Some(hex) = text.strip_prefix("0x").or_else(|| text.strip_prefix("0X")) {
Ok(u32::from_str_radix(hex, 16)?)
} else {
Ok(text.parse()?)
}
}
/// 一次性诊断,在某一路第一次出画面时记录。
///
/// 不再单列"无法解码"计数:那是链路正常但内容坏了的情形,会在解码线程里逐帧告警,
/// 汇总在这里反而会把它和普通的丢帧混在一起。
fn log_diagnostics(stats: &SourceStats, sender_id: u32) {
log::info!(
"camera 0x{sender_id:08X}: {} frames shown | {} dropped ({:.1}%) | \
{} datagrams | {}x{} {} bytes",
stats.frames_shown,
stats.frames_dropped,
stats.loss_percent(),
stats.datagrams,
stats.width,
stats.height,
stats.last_frame_len,
);
}
/// 把某一路的统计渲染成界面结构。
fn render_camera_stats(
cameras: &[CameraInfo],
sender_id: u32,
stats: &SourceStats,
fps: f32,
) -> CameraStats {
let online = cameras
.iter()
.find(|c| c.id == sender_id)
.map(CameraInfo::is_online)
.unwrap_or(false);
// 从没出过画面时区分"在线但还没收到"和"离线":前者只是刚连上,显示"等待画面"
// 比让人以为板子挂了诚实。
let status = if stats.frames_shown == 0 {
if online {
"已连接,等待画面…".to_string()
} else {
"离线".to_string()
}
} else {
format!("0x{:08X}", sender_id)
};
CameraStats {
online,
status_text: status.into(),
fps: format!("{fps:.1}").into(),
resolution: format!("{}x{}", stats.width, stats.height).into(),
frame_bytes: format!("{:.1} kB", stats.last_frame_len as f32 / 1024.0).into(),
frames_shown: stats.frames_shown.to_string().into(),
frames_dropped: stats.frames_dropped.to_string().into(),
loss_percent: format!("{:.1}%", stats.loss_percent()).into(),
datagrams: stats.datagrams.to_string().into(),
gap_ms: stats
.last_interval
.map(|d| format!("{:.0} ms", d.as_secs_f32() * 1000.0))
.unwrap_or_else(|| "-".to_string())
.into(),
}
}
/// 渲染一边的血量。
///
/// 现在永远走 `available == false` 的分支,因为还没有数据源。装甲板接上后
/// [`ArmorSource::health`] 返回 `Some`,界面自动显示真实数值,这里不用改。
///
/// 没有数据时用占位符而不是 0:0 看起来像真值,现场看到会以为对方已经没血了。
fn render_health(team: u8, armor: &dyn ArmorSource) -> TeamHealth {
let name = match team {
0 => "队伍一",
_ => "队伍二",
};
match armor.health(team) {
Some(hps) => TeamHealth {
team_name: format!("{name} · 装甲板").into(),
available: true,
placeholder: String::new().into(),
hps: std::rc::Rc::new(slint::VecModel::from(
hps.iter().map(|hp| *hp as i32).collect::<Vec<i32>>(),
))
.into(),
},
None => TeamHealth {
team_name: name.into(),
available: false,
placeholder:
"装甲板未接入:血量接口已预留(protocol::ArmorHealth),当前无数据源".into(),
// 依然给出 4 个格子:界面用 `available` 决定颜色,所以空数组会让占位
// 框整个消失,看起来像布局出错,而不是像"还没有数据"。
hps: std::rc::Rc::new(slint::VecModel::from(vec![0i32; ARMOR_BOARDS])).into(),
},
}
}
+458
View File
@@ -0,0 +1,458 @@
//! UDP 接收、发现、多路帧重组。
//!
//! # 多路
//!
//! 裁判终端同时收两块图传板,队伍终端也会收到两块(只是界面上只显示一块)。
//! 所以重组状态**必须按发送方分开**——把两块板的帧塞进同一个重组器,会让它们
//! 的 `frame_id` 互相顶掉,两路都出不来。
//!
//! # 发现与订阅
//!
//! 两个方向各司其职:
//!
//! - 板子每 2 秒广播一次 [`protocol::MsgType::Hello`],客户端被动收到就知道有谁在线。
//! - 客户端广播 [`protocol::MsgType::Subscribe`],板子收到后开始定向推流。
//!
//! 板子靠**源地址**记住客户端,所以客户端只要周期性续订就能维持推流,不存在
//! 任何超时淘汰的竞态。
//!
//! # 丢帧策略
//!
//! 缺了分片的帧整个丢弃、下一帧顶上,不重传也不尝试补洞。对实时画面这是对的:
//! 等重传补齐时,那一帧早就不新鲜了。这个取舍的代价是丢帧率必须摆在界面上,否则
//! "画面卡"没法区分是链路问题还是传感器问题。
use std::collections::HashMap;
use std::io::ErrorKind;
use std::net::{Ipv4Addr, SocketAddr, SocketAddrV4, UdpSocket};
use std::sync::atomic::{AtomicBool, Ordering};
use std::sync::{Arc, Mutex};
use std::thread;
use std::time::{Duration, Instant};
use protocol::{Fragment, FrameAssembler, Subscribe, SubscribeRole, DEFAULT_PORT, HEADER_LEN};
/// 接受的最大数据报。高于协议的理论上限(1427),留出余量,这样一个被截断的读只可能
/// 来自网络,而不是来自过小的缓冲。
const RECV_BUFFER: usize = 2048;
/// 重新广播订阅的间隔。
///
/// 比板子判定"客户端掉线"的超时(6 秒)短得多,所以偶发丢一两个包不会中断推流。
const SUBSCRIBE_INTERVAL: Duration = Duration::from_millis(800);
/// 一张帧没收齐就放弃的时限。
///
/// 一帧在 22fps 下约 45ms 到,所以几百毫秒还没齐的永远不会齐了。
const FRAME_TIMEOUT: Duration = Duration::from_millis(500);
/// 一块已知的图传板。
#[derive(Debug, Clone)]
pub struct CameraInfo {
pub id: u32,
pub addr: SocketAddrV4,
pub firmware: u16,
/// 最近一次收到它广播的时刻。
pub last_seen: Instant,
}
impl CameraInfo {
/// 多久没有广播了。
pub fn silence(&self) -> Duration {
self.last_seen.elapsed()
}
/// 板子在线吗?
///
/// 阈值是广播间隔的数倍,所以偶发丢包不会让界面上的在线状态闪��。
pub fn is_online(&self) -> bool {
self.silence() < Duration::from_secs(7)
}
}
/// 一块板的运行时统计。
#[derive(Debug, Clone, Copy, Default)]
pub struct SourceStats {
pub frames_shown: u64,
/// 因分片缺失被放弃的帧。
pub frames_dropped: u64,
pub datagrams: u64,
pub last_frame_len: usize,
pub width: u16,
pub height: u16,
/// 上一帧到这一帧之间的间隔。
pub last_interval: Option<Duration>,
last_shown_at: Option<Instant>,
}
impl SourceStats {
/// 丢弃的帧占已收到帧的比例。
pub fn loss_percent(&self) -> f32 {
let total = self.frames_shown + self.frames_dropped;
if total == 0 {
0.0
} else {
self.frames_dropped as f32 * 100.0 / total as f32
}
}
fn on_frame(&mut self, width: u16, height: u16, len: usize) {
self.last_interval = self.last_shown_at.map(|t| t.elapsed());
self.last_shown_at = Some(Instant::now());
self.frames_shown += 1;
self.width = width;
self.height = height;
self.last_frame_len = len;
}
}
/// 每一路的最新帧,等着被解码。
///
/// 每路只保留最新的一张:在 22fps 下把每一帧都排队意味着内存里压着几十张 JPEG,
/// 然后解码那些几毫秒后就被下一帧顶掉的画面——而这正是这条链路要消除的延迟。
#[derive(Default)]
pub struct LatestFrames {
inner: Mutex<HashMap<u32, protocol::Frame>>,
}
impl LatestFrames {
pub fn put(&self, frame: protocol::Frame) {
self.inner
.lock()
.expect("latest-frames mutex poisoned")
.insert(frame.sender_id, frame);
}
/// 取走某一路的待处理帧。
pub fn take(&self, sender_id: u32) -> Option<protocol::Frame> {
self.inner
.lock()
.expect("latest-frames mutex poisoned")
.remove(&sender_id)
}
/// 列出所有有待处理帧的板子 id。
pub fn pending_senders(&self) -> Vec<u32> {
self.inner
.lock()
.expect("latest-frames mutex poisoned")
.keys()
.copied()
.collect()
}
}
/// 所有板子的共享状态,跨线程只读。
pub struct Discovery {
cameras: Mutex<HashMap<u32, CameraInfo>>,
stats: Mutex<HashMap<u32, SourceStats>>,
}
impl Discovery {
fn new() -> Self {
Self {
cameras: Mutex::new(HashMap::new()),
stats: Mutex::new(HashMap::new()),
}
}
/// 当前已知的板子,按 id 排序,让界面上���列表顺序稳定。
pub fn cameras(&self) -> Vec<CameraInfo> {
let mut list: Vec<CameraInfo> = self
.cameras
.lock()
.expect("cameras mutex poisoned")
.values()
.cloned()
.collect();
list.sort_by_key(|c| c.id);
list
}
/// 在线且刚刚广播过的板子。
pub fn online_cameras(&self) -> Vec<CameraInfo> {
self.cameras().into_iter().filter(CameraInfo::is_online).collect()
}
/// 某一路的统计。
pub fn stats(&self, sender_id: u32) -> SourceStats {
self.stats
.lock()
.expect("stats mutex poisoned")
.get(&sender_id)
.copied()
.unwrap_or_default()
}
fn note_hello(&self, id: u32, addr: SocketAddrV4, firmware: u16) {
let mut cameras = self.cameras.lock().expect("cameras mutex poisoned");
cameras
.entry(id)
.and_modify(|c| {
c.addr = addr;
c.last_seen = Instant::now();
c.firmware = firmware;
})
.or_insert(CameraInfo {
id,
addr,
firmware,
last_seen: Instant::now(),
});
}
fn note_frame(&self, id: u32, width: u16, height: u16, len: usize) {
let mut stats = self.stats.lock().expect("stats mutex poisoned");
stats.entry(id).or_default().on_frame(width, height, len);
}
fn count(&self, id: u32, f: impl FnOnce(&mut SourceStats)) {
let mut stats = self.stats.lock().expect("stats mutex poisoned");
f(stats.entry(id).or_default());
}
/// 超过 `max_idle` 没有广播的板子从列表里移除。
///
/// 界面上不会留下一堆永远离线的板子。
pub fn forget_silent_cameras(&self, max_idle: Duration) {
self.cameras
.lock()
.expect("cameras mutex poisoned")
.retain(|_, c| c.silence() < max_idle);
}
}
/// 客户端身份。
///
/// 常量而非 MAC 派生:队伍客户端和裁判终端各自唯一,进程内也不需要区分自己,
/// 而一个确定的常量让日志更好读。
pub const CLIENT_ID: u32 = 0x0C11_0001;
/// 接收器。
///
/// socket 在这里创建而不是在接收线程里,这样"端口被占用"会立刻报错,而不是被线程
/// 吞掉、表现为一个永远不出画面的空窗口。
pub struct Receiver {
discovery: Arc<Discovery>,
frames: Arc<LatestFrames>,
role: SubscribeRole,
camera: Ipv4Addr,
stop: Arc<AtomicBool>,
}
impl Receiver {
/// 绑定视频端口并准备好发现。
///
/// 绑定固定端口(而非临时端口)是必须的:板子是从订阅的**源端口**学到推流目标的,
/// 临时端口也能工作,但固定端口让三台客户端行为一致,也方便抓包排查。
pub fn start(camera: Ipv4Addr, role: SubscribeRole) -> std::io::Result<Self> {
let socket =
UdpSocket::bind(SocketAddrV4::new(Ipv4Addr::UNSPECIFIED, DEFAULT_PORT))?;
socket.set_read_timeout(Some(Duration::from_millis(100)))?;
let broadcast = socket.set_broadcast(true).is_ok();
if !broadcast {
log::warn!("could not enable broadcast on the socket");
}
Ok(Self {
discovery: Arc::new(Discovery::new()),
frames: Arc::new(LatestFrames::default()),
role,
camera,
stop: Arc::new(AtomicBool::new(false)),
})
}
pub fn discovery(&self) -> Arc<Discovery> {
Arc::clone(&self.discovery)
}
pub fn frames(&self) -> Arc<LatestFrames> {
Arc::clone(&self.frames)
}
/// 运行接收循环,直到 [`Receiver::shutdown`]。
///
/// 阻塞,所以放在单独的线程上。
pub fn run(&self, socket: UdpSocket) {
let target = SocketAddrV4::new(self.camera, DEFAULT_PORT);
let subscribe = Subscribe {
role: self.role,
client_id: CLIENT_ID,
};
let subscribe_bytes = subscribe.encode();
// 每个发送方一个重组器:两块板的 frame_id 各自独立,混在一起会互相顶掉。
let mut assemblers: HashMap<u32, FrameAssembler> = HashMap::new();
let mut buf = [0u8; RECV_BUFFER];
let mut last_subscribe = Instant::now() - SUBSCRIBE_INTERVAL;
let mut last_sweep = Instant::now();
log::info!(
"listening on UDP {DEFAULT_PORT}, announcing {:?} to {target}",
self.role
);
while !self.stop.load(Ordering::Relaxed) {
// 周期性续订就是保活:板子在 6 秒收不到就认为我们走了。
if last_subscribe.elapsed() >= SUBSCRIBE_INTERVAL {
if let Err(err) = socket.send_to(&subscribe_bytes, target) {
log::warn!("could not subscribe at {target}: {err}");
}
last_subscribe = Instant::now();
}
match socket.recv_from(&mut buf) {
Ok((n, from)) => {
self.handle_datagram(&buf[..n], from, &mut assemblers);
}
Err(err)
if err.kind() == ErrorKind::WouldBlock
|| err.kind() == ErrorKind::TimedOut =>
{
// 帧之间的常态。读超时正是让退出能及时响应的原因。
}
Err(err) => {
log::warn!("recv failed: {err}");
thread::sleep(Duration::from_millis(100));
}
}
// 放弃等太久的半截帧。否则板子在帧中途死掉时,重组器会一直留着它。
for (id, asm) in assemblers.iter_mut() {
if asm.expire().is_some() {
self.discovery.count(*id, |s| s.frames_dropped += 1);
}
}
if last_sweep.elapsed() >= Duration::from_secs(5) {
self.discovery.forget_silent_cameras(Duration::from_secs(60));
last_sweep = Instant::now();
}
}
log::info!("receive loop stopped");
}
/// 处理一个数据报。
fn handle_datagram(
&self,
bytes: &[u8],
from: std::net::SocketAddr,
assemblers: &mut HashMap<u32, FrameAssembler>,
) {
if bytes.len() < HEADER_LEN {
return;
}
// 板子的在线通告。注意这里不看来源端口:广播来的源端口没有意义,身份由
// 头里的 sender_id 决定。
if let Some(hello) = protocol::Hello::decode(bytes) {
if let SocketAddr::V4(v4) = from {
let is_new = !self.discovery.cameras().iter().any(|c| c.id == hello.sender_id);
self.discovery.note_hello(hello.sender_id, v4, hello.firmware);
if is_new {
log::info!(
"found camera 0x{:08X} at {v4}",
hello.sender_id
);
}
}
return;
}
let Some(frag) = Fragment::decode(bytes) else {
// 既不是 Hello 也不是合法分片:将来的其它消息,或一次损坏。
return;
};
let id = frag.header.sender_id;
match assemblers
.entry(id)
.or_insert_with(|| FrameAssembler::new(FRAME_TIMEOUT))
.push(frag)
{
Ok(frame) => {
self.discovery.count(id, |s| s.datagrams += 1);
self.discovery
.note_frame(id, frame.width, frame.height, frame.data.len());
self.frames.put(frame);
}
Err(protocol::DropReason::Superseded) => {
// 被更新的帧追上了。这不是丢帧:半截的那张只是被超越,在健康链路上
// 这是帧被放弃的**正常**方式。
self.discovery.count(id, |s| s.datagrams += 1);
}
Err(protocol::DropReason::TimedOut) => {
self.discovery.count(id, |s| {
s.datagrams += 1;
s.frames_dropped += 1;
});
}
}
}
pub fn shutdown(&self) {
self.stop.store(true, Ordering::Relaxed);
}
}
/// 解码后的画面。
pub struct DecodedFrame {
pub rgba: Vec<u8>,
pub width: u32,
pub height: u32,
}
/// 解码一张 JPEG。
///
/// 返回 `None` 表示字节不是可解码的图像——协议层挡不住这种:它只检查分片结构,
/// 不检查 JPEG 本身是否有效。
pub fn decode_frame(jpeg: &[u8]) -> Option<DecodedFrame> {
let img = image::load_from_memory_with_format(jpeg, image::ImageFormat::Jpeg).ok()?;
let rgba = img.to_rgba8();
Some(DecodedFrame {
width: rgba.width(),
height: rgba.height(),
rgba: rgba.into_raw(),
})
}
/// 装甲板血量的来源。
///
/// **预留,当前没有实现也没有数据源。** 血量显示为占位符。
///
/// 接口和端口都已就位([`protocol::ARMOR_PORT`]、[`protocol::ArmorHealth`]):装甲板
/// 接上以后往那个端口发消息即可,裁判界面与其它代码都不用改。
pub trait ArmorSource: Send + Sync {
/// 某一边的血量。
///
/// 返回 4 个装甲板的当前血量,按 index 排列。没有数据时返回 `None`,界面据此
/// 显示占位符而不是 0——0 是一个看起来像真值的假值。
fn health(&self, team: u8) -> Option<[u16; 4]>;
}
/// 一边的装甲板数量。
///
/// RoboMaster 规则是每队 4 块。界面按这个数量画格子,所以数据还没接上时也能看到
/// 位置——空着比画 0 个格子更让人以为布局坏了。
pub const ARMOR_BOARDS: usize = 4;
/// 目前唯一的实现:什么都不提供。
///
/// 存在的意义是让界面代码从一开始就对着一个真实接口开发,而不是等装甲板就绪后再改
/// 一次结构。
pub struct NoArmorYet;
impl ArmorSource for NoArmorYet {
fn health(&self, _team: u8) -> Option<[u16; 4]> {
None
}
}