#!/usr/bin/env python3 """Measure real UDP video throughput, complete-frame loss and receive jitter. Uses an ephemeral port by default so it can run beside the desktop client. The timings are receive-side metrics, not camera-to-display latency. """ import argparse from collections import Counter import json from pathlib import Path import socket import secrets import struct import time from udp_probe import decode PORT = 5600 FRAGMENT = 1400 def percentile(values, p): if not values: return None values = sorted(values) return round(values[int((len(values) - 1) * p)], 3) def summary(source, duration): counts = source['counts'] frames = counts['complete_frames'] dropped = counts['superseded_frames'] + counts['missing_frames'] + counts['invalid_jpeg'] + counts['timed_out_frames'] span = source['last_complete'] - source['first_complete'] if frames > 1 else 0 return dict( address=source['address'], counts=dict(counts), window_fps=round(frames / duration, 3) if duration else 0, active_fps=round((frames - 1) / span, 3) if span else 0, frame_loss_percent=round(100 * dropped / max(1, frames + dropped), 3), resolution=source.get('resolution'), mean_jpeg_bytes=round(counts['jpeg_bytes'] / frames) if frames else 0, inter_frame_p50_ms=percentile(source['intervals'], .5), inter_frame_p95_ms=percentile(source['intervals'], .95), inter_frame_max_ms=round(max(source['intervals']), 3) if source['intervals'] else None, assembly_p95_ms=percentile(source['assembly'], .95), pending_frames=int(source['pending'] is not None), ) def main(): parser = argparse.ArgumentParser(description=__doc__) parser.add_argument('host', help='Board IPv4 address; avoids VPN limited-broadcast routing') parser.add_argument('--seconds', type=float, default=30) parser.add_argument('--local-port', type=int, default=0) parser.add_argument('--local-ip', default='') parser.add_argument('--receive-buffer', type=int, default=1024 * 1024, help='0 keeps the OS default') parser.add_argument('--json', type=Path) parser.add_argument('--save-frame', type=Path) args = parser.parse_args() client_id = secrets.randbelow(0xFFFFFFFF) + 1 if client_id == 0x0C110001: client_id += 1 subscribe = struct.pack(' .5: source['counts']['timed_out_frames'] += 1 source['pending'] = None if now >= renew: sock.sendto(subscribe, (args.host, PORT)) renew = now + .8 if now >= report_at: frames = sum(s['counts']['complete_frames'] for s in sources.values()) print(f'{now-started:5.1f}s {(frames-previous_frames)/(now-previous_report):5.1f} fps complete={frames}', flush=True) previous_frames, previous_report, report_at = frames, now, now + 5 try: data, address = sock.recvfrom(2048) except socket.timeout: continue now = time.perf_counter() parsed = decode(data) if parsed is None: invalid += 1 continue kind, h, payload = parsed if kind != 1: continue size = h['total_len'] index, count = h['frag_index'], h['frag_count'] offset = index * FRAGMENT if h['version'] != 1 or size > 4 * 1024 * 1024 or count != max(1, (size + FRAGMENT - 1) // FRAGMENT) or index >= count or offset > size or len(payload) != min(FRAGMENT, size-offset): invalid += 1 continue sender, frame = h['sender_id'], h['frame_id'] if sender not in sources: sources[sender] = dict(address=address[0], counts=Counter(), latest=None, last_seen=now, pending=None, intervals=[], assembly=[], first_complete=None, last_complete=None) s = sources[sender] counts = s['counts'] counts['video_datagrams'] += 1 latest = s['latest'] if latest is not None: distance = (frame - latest) & 0xFFFFFFFF if distance >= 0x80000000 and now-s['last_seen'] >= 1: counts['restarts'] += 1 latest = s['latest'] = None s['pending'] = None elif distance >= 0x80000000 or (distance == 0 and s['pending'] is None): counts['late_or_repeated_datagrams'] += 1 continue elif distance > 0: counts['missing_frames'] += distance - 1 counts['superseded_frames'] += int(s['pending'] is not None) s['pending'] = None s['latest'], s['last_seen'] = frame, now if s['pending'] is None: s['pending'] = dict(buffer=bytearray(size), seen=set(), count=count, size=size, width=h['width'], height=h['height'], started=now) pending = s['pending'] if (size, count, h['width'], h['height']) != (pending['size'], pending['count'], pending['width'], pending['height']): counts['inconsistent_metadata'] += 1 continue if index in pending['seen']: counts['duplicate_fragments'] += 1 continue pending['seen'].add(index) pending['buffer'][offset:offset+len(payload)] = payload if len(pending['seen']) != count: continue jpeg = bytes(pending['buffer']) if jpeg[:2] == b'\xff\xd8' and jpeg[-2:] == b'\xff\xd9': completed = time.perf_counter() counts['complete_frames'] += 1 counts['jpeg_bytes'] += size if s['last_complete'] is not None: s['intervals'].append((completed-s['last_complete'])*1000) s['last_complete'] = completed if s['first_complete'] is None: s['first_complete'] = completed s['assembly'].append((completed-pending['started'])*1000) s['resolution'] = [h['width'], h['height']] if args.save_frame and not saved: args.save_frame.write_bytes(jpeg) saved = True else: counts['invalid_jpeg'] += 1 s['pending'] = None finally: sock.close() elapsed = time.perf_counter()-started result = dict(seconds=round(elapsed, 3), host=args.host, local_port=local_port, receive_buffer_bytes=buffer_bytes, invalid_datagrams=invalid, sources={f'0x{sender:08X}': summary(s, elapsed) for sender, s in sources.items()}, timing_scope='receiver inter-frame spacing and first-to-last fragment assembly; not camera-to-display latency') if args.json: args.json.write_text(json.dumps(result, ensure_ascii=False, indent=2)+'\n', encoding='utf-8') print(json.dumps(result, ensure_ascii=False, indent=2)) return 0 if any(s['counts']['complete_frames'] for s in sources.values()) else 1 if __name__ == '__main__': raise SystemExit(main())