diff --git a/protocol/src/lib.rs b/protocol/src/lib.rs index 3dcce8d..83c1f8a 100644 --- a/protocol/src/lib.rs +++ b/protocol/src/lib.rs @@ -540,8 +540,13 @@ pub struct FrameAssembler { height: u16, since: Option, timeout: std::time::Duration, - /// 已收到的分片数，含重复。用来统计网络丢了多少。 + /// 已收到的分片数（不含重复）。用来统计网络丢了多少。 seen: usize, + /// 每个分片是否已收到。 + /// + /// 存在的原因是 `seen` 会被重复包灌水：UDP 会重复投递，一个丢失的分片可以 + /// 被另一个包的副本凑数，于是"收够了"却留着空洞，交出去一张尾部是 0 的坏图。 + have: Vec, } impl FrameAssembler { @@ -557,6 +562,7 @@ impl FrameAssembler { since: None, timeout, seen: 0, + have: Vec::new(), } } @@ -602,12 +608,27 @@ impl FrameAssembler { 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 start = frag.header.frag_index as usize * MAX_FRAGMENT_PAYLOAD; + let index = frag.header.frag_index as usize; + + // 重复分片直接跳过，且**不**增加 seen。 + // + // 这里曾经用 `seen >= expected` 判断收齐，那是错的：UDP 会重复 delivered + // 重复包，一个丢失的分片可以被另一个包的副本凑数，于是 seen 达标而缓冲区 + // 里留着空洞，交出去一张 JPEG 尾部是 0 的坏图。这个 bug 很隐蔽——它不会让 + // 帧率下降，只会让偶尔几张图解不出来，而接收端没有任何线索知道原因。 + 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() { @@ -616,7 +637,13 @@ impl FrameAssembler { self.buffer[start..end].copy_from_slice(&frag.payload); self.seen += 1; - if self.seen >= self.expected { + // 计数达标还不够：最后一个分片必须真的到过，否则中间可能缺一块。 + // 显式检查尾块，比依赖 seen 的算术更直接，也更容易看出意图。 + let last = self.expected - 1; + let tail_expected = frag.header.total_len as usize; + let tail_index = tail_expected.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())); @@ -660,6 +687,7 @@ impl FrameAssembler { self.buffer.shrink_to_fit(); self.since = None; self.seen = 0; + self.have = Vec::new(); } } @@ -852,6 +880,50 @@ mod tests { 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 = (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); diff --git a/scripts/pick-flash-pad.sh b/scripts/pick-flash-pad.sh deleted file mode 100755 index 726bac2..0000000 --- a/scripts/pick-flash-pad.sh +++ /dev/null @@ -1,55 +0,0 @@ -#!/usr/bin/env bash -# Finds a FLASH_PAD value that moves the firmware off a defective SPI flash byte. -# -# The module this was developed against has one byte at flash address 0xFADE7 -# whose bits 5..7 cannot be programmed, so whatever lands there gets corrupted -# and the image checksum (and therefore the app) fails to boot. Growing -# FLASH_PAD shifts .flash.text, which moves a different byte onto that address. -# A byte is safe when its top three bits are zero, i.e. value <= 0x1F. -# -# Usage: scripts/pick-flash-pad.sh [max_attempts] -# Prints the FLASH_PAD value to use, or exits non-zero if none was found. -set -euo pipefail - -cd "$(dirname "$0")/.." - -# Flash address of the defective cell, and the offset of the app partition. -readonly BAD_FLASH_ADDR="${BAD_FLASH_ADDR:-0xFADE7}" -readonly APP_OFFSET="${APP_OFFSET:-0x10000}" -readonly MAX_ATTEMPTS="${1:-40}" - -image=$(mktemp -t esp32-app.XXXXXX.bin) -trap 'rm -f "$image"' EXIT - -bad_offset=$(python3 -c "print($BAD_FLASH_ADDR - $APP_OFFSET)") - -echo "bad cell: flash $BAD_FLASH_ADDR (image offset $bad_offset)" - -for ((pad = 0; pad < MAX_ATTEMPTS; pad++)); do - FLASH_PAD="$pad" ./scripts/build.sh >/dev/null 2>&1 || { - echo "build failed for FLASH_PAD=$pad" >&2 - exit 1 - } - - espflash save-image --chip esp32s3 --format esp-idf \ - target/xtensa-esp32s3-espidf/release/esp32 "$image" >/dev/null 2>&1 - - byte=$(python3 - "$image" "$bad_offset" <<'PY' -import sys -data = open(sys.argv[1], "rb").read() -off = int(sys.argv[2]) -print(data[off] if off < len(data) else -1) -PY -) - - if (( byte >= 0 && (byte & 0xE0) == 0 )); then - echo "FLASH_PAD=$pad works: byte at $BAD_FLASH_ADDR would be 0x$(printf '%02X' "$byte")" - echo "set FLASH_PAD to $pad (in src/main.rs, or export it at build time)" - exit 0 - fi - - echo " FLASH_PAD=$pad -> 0x$(printf '%02X' "$byte" 2>/dev/null || echo '??') (unsafe)" -done - -echo "no safe FLASH_PAD found in 0..$((MAX_ATTEMPTS - 1)); this module's flash is too broken" >&2 -exit 1 \ No newline at end of file diff --git a/scripts/udp_probe.py b/scripts/udp_probe.py index 773b5c8..3cba8bf 100644 --- a/scripts/udp_probe.py +++ b/scripts/udp_probe.py @@ -1,110 +1,192 @@ #!/usr/bin/env python3 -"""Minimal UDP listener for the camera stream, used to verify the firmware. +"""独立验证相机 UDP 图传线路的诊断脚本。 -Not the real client - just enough to confirm datagrams arrive, that they -reassemble into a valid JPEG, and to measure the frame rate independently of -the Slint app. +不依赖 Rust，也不依赖 protocol crate——正因为如此，它能当作线路格式的 +**独立**检查：如果 protocol crate 的编码有错，这个脚本（手写的常量）仍然 +按文档解析，会立刻报出不一致。 + +用法： + + python3 scripts/udp_probe.py [秒数] + +它做三件事： + 1. 广播一次 Subscribe，触发板子开始推流 + 2. 解析收到的数据报，按 sender_id 重组 + 3. 报告帧率、丢帧、以及是否发现了板子的 Hello + +常量是手写的，与 protocol/src/lib.rs 保持一致。改了协议记得改这里—— +这正是它作为交叉检查的价值所在。 """ -import socket, sys, time, collections +import socket +import sys +import time +from collections import defaultdict -HOST = "192.168.1.139" -PORT = 5600 -DURATION = float(sys.argv[1]) if len(sys.argv) > 1 else 10.0 +# ---- 与 protocol/src/lib.rs 一致 ---- +HEADER = 27 # magic4 + type1 + ver2 + sender4 + frame4 + idx2 + cnt2 + len4 + w2 + h2 +MAGIC = 0x4D564331 # "MVC1" +FRAG = 1400 # MAX_FRAGMENT_PAYLOAD +PORT = 5600 # DEFAULT_PORT +BROADCAST = "255.255.255.255" -# Must match protocol/src/lib.rs: MAGIC, HEADER_LEN, MAX_FRAGMENT_PAYLOAD, -# DEFAULT_PORT. Duplicated here on purpose - this is a throwaway diagnostic, and -# making it depend on the Rust crate would defeat the point of it being an -# independent check of the wire format. -HEADER = 20 -MAGIC = 0x4D564331 -FRAG = 1400 +MSG_VIDEO = 1 +MSG_SUBSCRIBE = 2 +MSG_HELLO = 3 +MSG_HEARTBEAT = 4 -sock = socket.socket(socket.AF_INET, socket.SOCK_DGRAM) -sock.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1) -sock.bind(("", PORT)) +TYPE_NAMES = { + MSG_VIDEO: "video", + MSG_SUBSCRIBE: "subscribe", + MSG_HELLO: "hello", + MSG_HEARTBEAT: "heartbeat", +} -# Announce ourselves, exactly as the Slint client will. -sock.sendto(b"\x00", (HOST, PORT)) -sock.settimeout(0.5) - -frag_bytes = FRAG -frames = {} -stats = collections.Counter() -sizes = [] -start = time.time() -last_print = start - -while time.time() - start < DURATION: - try: - data, _ = sock.recvfrom(2048) - except socket.timeout: - continue - - stats["datagrams"] += 1 +def decode(data): + """把一个数据报解析成 (msg_type, header_dict, payload)。""" if len(data) < HEADER: - stats["short"] += 1 - continue + return None magic = int.from_bytes(data[0:4], "little") if magic != MAGIC: - stats["bad_magic"] += 1 - continue + return None - frame_id = int.from_bytes(data[4:8], "little") - frag_index = int.from_bytes(data[8:10], "little") - frag_count = int.from_bytes(data[10:12], "little") - total_len = int.from_bytes(data[12:16], "little") - width = int.from_bytes(data[16:18], "little") - height = int.from_bytes(data[18:20], "little") + msg_type = data[4] + header = { + "version": int.from_bytes(data[5:7], "little"), + "sender_id": int.from_bytes(data[7:11], "little"), + "frame_id": int.from_bytes(data[11:15], "little"), + "frag_index": int.from_bytes(data[15:17], "little"), + "frag_count": int.from_bytes(data[17:19], "little"), + "total_len": int.from_bytes(data[19:23], "little"), + "width": int.from_bytes(data[23:25], "little"), + "height": int.from_bytes(data[25:27], "little"), + } + return msg_type, header, data[HEADER:] - payload = data[HEADER:] - if frame_id not in frames: - if frames: - stats["superseded"] += 1 - frames[frame_id] = { - "buf": bytearray(total_len), - "seen": 0, - "count": frag_count, - "total": total_len, - "w": width, - "h": height, - "t0": time.time(), - } +def main(): + duration = float(sys.argv[1]) if len(sys.argv) > 1 else 10.0 + host = sys.argv[2] if len(sys.argv) > 2 else BROADCAST - f = frames[frame_id] - off = frag_index * frag_bytes - if off + len(payload) <= total_len: - f["buf"][off:off + len(payload)] = payload - f["seen"] += 1 - else: - stats["overflow"] += 1 + sock = socket.socket(socket.AF_INET, socket.SOCK_DGRAM) + sock.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1) + sock.setsockopt(socket.SOL_SOCKET, socket.SO_BROADCAST, 1) + sock.bind(("", PORT)) + sock.settimeout(0.5) - if f["seen"] >= f["count"]: - jpeg = bytes(f["buf"]) - if jpeg[:2] == b"\xff\xd8" and jpeg[-2:] == b"\xff\xd9": - stats["good"] += 1 - sizes.append((f["total"], f["w"], f["h"])) - else: - stats["corrupt"] += 1 - del frames[frame_id] + # 订阅：板子看到它就开始往本机推流。 + # + # role 字节紧跟在 HEADER 之后，也就是偏移 HEADER 处。之前这里多填了 12 个 + # 零字节，role 落到了偏移 39——板子的 Subscribe::decode 严格按偏移读，于是 + # 认不出 role 而拒收，表现为"板子在线但 0 clients subscribed"。 + # 这里刻意手写而不是复用 protocol crate：这个脚本的意义就在于独立验证， + # role 的位置写错时它会给出和固件一致的结论，而不是把错误藏起来。 + header = ( + MAGIC.to_bytes(4, "little") + + bytes([MSG_SUBSCRIBE]) + + (1).to_bytes(2, "little") # version + + (0x0C110001).to_bytes(4, "little") # client_id + + b"\x00" * 4 # frame_id + + b"\x00" * 4 # frag_index + frag_count + + b"\x00" * 4 # total_len + + b"\x00" * 4 # width + height + ) + sub = header + bytes([2]) # role = referee，紧跟 27 字节头 + sock.sendto(sub, (host, PORT)) + print(f"已向 {host}:{PORT} 广播订阅（role=referee），监听 {duration:.0f} 秒…") - now = time.time() - if now - last_print >= 2.0: - elapsed = now - last_print - print( - f" {stats['good']/elapsed:5.1f} fps " - f"datagrams={stats['datagrams']:5d} " - f"superseded={stats['superseded']:4d} corrupt={stats['corrupt']:3d} " - f"good={stats['good']:4d}" + # 每个发送方一份重组状态：两块板的 frame_id 各自独立，混在一起会互相顶掉。 + pending = {} + boards = {} + counters = defaultdict(int) + sizes = [] + start = time.time() + last_print = start + + while time.time() - start < duration: + try: + data, addr = sock.recvfrom(2048) + except socket.timeout: + continue + + parsed = decode(data) + if parsed is None: + counters["无法解析"] += 1 + continue + + msg_type, h, payload = parsed + counters[f"type:{TYPE_NAMES.get(msg_type, msg_type)}"] += 1 + + if msg_type in (MSG_HELLO, MSG_HEARTBEAT): + boards[h["sender_id"]] = addr + continue + + if msg_type != MSG_VIDEO: + continue + + sid = h["sender_id"] + + # 与 Fragment::decode 相同的自洽性检查。 + if h["frag_count"] == 0 or h["frag_index"] >= h["frag_count"]: + counters["分片索引非法"] += 1 + continue + if len(payload) > h["total_len"]: + counters["载荷超过总长"] += 1 + continue + + state = pending.setdefault( + sid, + {"buf": bytearray(h["total_len"]), "seen": 0, "expect": h["frag_count"], + "have": set(), "total": h["total_len"], "w": h["width"], "h": h["height"], + "started": time.time()}, ) - last_print = now -sock.close() -print() -print("totals:", dict(stats)) -if sizes: - avg = sum(s[0] for s in sizes) / len(sizes) - print(f"avg frame: {avg:.0f} bytes, {sizes[0][1]}x{sizes[0][2]}") - print(f"throughput: {sum(s[0] for s in sizes)/DURATION/1024:.1f} kB/s") + # 重复分片直接跳过且不计数：否则一个丢失的分片会被副本凑数， + # seen 达标而缓冲区留空洞，交出去一张尾部是 0 的坏图。 + if h["frag_index"] in state["have"]: + counters["重复分片"] += 1 + continue + state["have"].add(h["frag_index"]) + + off = h["frag_index"] * FRAG + state["buf"][off:off + len(payload)] = payload + state["seen"] += 1 + + tail = (h["total_len"] - 1) // FRAG + if state["seen"] >= state["expect"] and tail in state["have"]: + jpeg = bytes(state["buf"][:state["total"]]) + if jpeg[:2] == b"\xff\xd8" and jpeg[-2:] == b"\xff\xd9": + counters["完整帧"] += 1 + sizes.append((state["total"], state["w"], state["h"])) + else: + counters["JPEG 不完整"] += 1 + del pending[sid] + + now = time.time() + if now - last_print >= 2.0: + elapsed = now - last_print + good = counters["完整帧"] + print( + f" {good / elapsed:5.1f} fps " + f"视频包={counters['type:video']:5d} " + f"完整帧={good:4d} 丢弃={counters['JPEG 不完整']:3d}" + ) + last_print = now + + sock.close() + + total_time = time.time() - start + print() + print("板子:", ", ".join(f"0x{k:08X} @ {v[0]}" for k, v in boards.items()) or "（未发现）") + print("完整帧:", counters["完整帧"], " 累计视频包:", counters["type:video"]) + if counters["完整帧"]: + avg = sum(s[0] for s in sizes) / len(sizes) + print(f"平均帧: {avg:.0f} 字节 {sizes[0][1]}x{sizes[0][2]}") + print(f"吞吐: {sum(s[0] for s in sizes) / total_time / 1024:.1f} kB/s") + bad = {k: v for k, v in counters.items() if k.startswith(("无法", "JPEG", "分片", "载荷"))} + print("异常:", bad or "无") + + +if __name__ == "__main__": + main()