1+ #! /usr/bin/python3
2+ ##--------------------------------------------------------------------\
3+ # hackrfpy 'examples/benchmark.py'
4+ # Measure the real performance gaps, replacing the comparison doc's
5+ # hand-waving with numbers from YOUR hardware + machine. Read-only.
6+ #
7+ # Measures three things:
8+ # 1. Decode throughput (MB/s) -- pure CPU, no device. How fast the
9+ # int8->complex64 path runs vs the ~40 MB/s the device maxes at.
10+ # 2. Sustained-rate drop test -- the important one. Capture at a target
11+ # sample rate and report whether hackrf_transfer dropped samples
12+ # (it warns on buffer overrun) and whether the consumer kept up.
13+ # 3. Callback block cadence -- inter-block timing + jitter, the practical
14+ # proxy for "latency floor is the pipe."
15+ #
16+ # IMPORTANT: results are specific to THIS board, machine, USB stack, and
17+ # system load. They are honest measurements, not universal claims. Run it a
18+ # few times; close other heavy apps for the drop test.
19+ #
20+ # Usage:
21+ # uv run python examples/benchmark.py
22+ # uv run python examples/benchmark.py --rate 20e6 --seconds 3
23+ # uv run python examples/benchmark.py --tools-dir "C:\hackrf-tools-windows"
24+ ##--------------------------------------------------------------------\
25+ import argparse
26+ import sys
27+ import time
28+ import numpy as np
29+ from hackrfpy import HackRF
30+
31+
32+ def bench_decode (h ):
33+ # pure decode throughput; no device needed
34+ print ("\n == 1. decode throughput (CPU only) ==" )
35+ block = np .random .randint (0 , 256 , 262144 , dtype = np .uint8 ).tobytes () # 256KB
36+ N = 2000
37+ mb = N * len (block ) / 1e6
38+ t0 = time .perf_counter ()
39+ for _ in range (N ):
40+ h .decode_iq (block )
41+ dt = time .perf_counter () - t0
42+ rate = mb / dt
43+ print (f" { rate :.0f} { rate / 40 :.1f} )
44+ print (f" -> decode is { 'NOT ' if rate > 80 else '' } )
45+ return rate
46+
47+
48+ def bench_drop_test (h , rate , seconds ):
49+ # the important one: does a sustained capture drop samples?
50+ #
51+ # MEASUREMENT NOTE: process spin-up (launching hackrf_transfer, USB setup,
52+ # device settling) is a fixed ~1-2s cost. Dividing total samples by total
53+ # wall-clock makes a short capture look slow even when every sample
54+ # arrived. So we measure STEADY STATE: discard a warm-up window, then time
55+ # only the samples after it. We also report whether the full requested
56+ # count actually arrived (no real loss).
57+ print (f"\n == 2. sustained drop test @ { rate / 1e6 :g} { seconds } )
58+ n = int (rate * seconds )
59+ warmup_s = 1.5
60+ got = 0
61+ warm_got = 0
62+ warm_t0 = None
63+ t_start = time .perf_counter ()
64+ try :
65+ with h .capture_stream (100_000_000 , rate ) as stream :
66+ for iq in stream :
67+ got += len (iq )
68+ now = time .perf_counter ()
69+ if warm_t0 is None and (now - t_start ) >= warmup_s :
70+ warm_t0 = now
71+ warm_got = 0
72+ elif warm_t0 is not None :
73+ warm_got += len (iq )
74+ if got >= n :
75+ break
76+ except Exception as e :
77+ print (f" capture error: { e } , file = sys .stderr )
78+ return
79+ total_dt = time .perf_counter () - t_start
80+ steady_dt = (time .perf_counter () - warm_t0 ) if warm_t0 else None
81+
82+ print (f" received { got :,} { n :,} { 100 * got / n :.1f}
83+ f"in { total_dt :.2f} )
84+ if steady_dt and steady_dt > 0.2 :
85+ steady_rate = warm_got / steady_dt / 1e6
86+ print (f" STEADY-STATE rate (after { warmup_s }
87+ f"{ steady_rate :.1f} { rate / 1e6 :g} )
88+ if steady_rate >= rate / 1e6 * 0.95 :
89+ print (f" -> sustains { rate / 1e6 :g} )
90+ else :
91+ print (f" -> steady rate below target; possible real limit here" )
92+
93+ # Drop detection: authoritative hackrf_debug -S shortfall count + the
94+ # hackrf_transfer 'overruns' count from stderr.
95+ import re
96+ err = "" .join (getattr (h , "_err_chunks" , []))
97+ overruns = sum (int (m ) for m in re .findall (r"(\d+)\s+overruns" , err ))
98+ shortfall = None
99+ try :
100+ out , _ , _ = h ._run (["debug" , "-S" ], mode = "blocking" , text = True )
101+ m = re .search (r"shortfall[^0-9]*(\d+)" , out , re .I )
102+ if m :
103+ shortfall = int (m .group (1 ))
104+ except Exception :
105+ pass
106+
107+ if got >= n * 0.999 :
108+ print (f" note: ~100% of requested samples received -- no bulk loss" )
109+ if shortfall is not None :
110+ if shortfall == 0 :
111+ print (f" hackrf_debug -S: 0 shortfalls -- clean" )
112+ elif shortfall <= 3 :
113+ print (f" hackrf_debug -S: { shortfall }
114+ f"startup transient, not sustained loss" )
115+ else :
116+ print (f" ! hackrf_debug -S: { shortfall }
117+ f"sustained drops at { rate / 1e6 :g} )
118+ elif overruns :
119+ print (f" hackrf_transfer reported { overruns } )
120+
121+
122+ def bench_callback_jitter (h , rate , seconds ):
123+ # inter-block cadence + jitter -- the practical latency-floor proxy
124+ print (f"\n == 3. callback cadence @ { rate / 1e6 :g} )
125+ stamps = []
126+
127+ def on_block (iq , total ):
128+ stamps .append (time .perf_counter ())
129+
130+ n = int (rate * seconds )
131+ try :
132+ h .capture_callback (100_000_000 , rate , on_block = on_block , max_samples = n )
133+ except Exception as e :
134+ print (f" error: { e } , file = sys .stderr )
135+ return
136+ if len (stamps ) < 3 :
137+ print (" too few blocks to measure" )
138+ return
139+ gaps = np .diff (stamps ) * 1000.0 # ms between callbacks
140+ print (f" { len (stamps )}
141+ f"mean { gaps .mean ():.2f} { np .median (gaps ):.2f}
142+ f"max { gaps .max ():.2f} )
143+ print (f" jitter (std): { gaps .std ():.2f} )
144+ print (f" -> practical block-delivery latency is ~{ np .median (gaps ):.1f}
145+ f"on this machine (NOT antenna-to-callback; that needs a reference)" )
146+
147+
148+ def main ():
149+ p = argparse .ArgumentParser (description = "Measure real performance (read-only)." )
150+ p .add_argument ("--tools-dir" , default = None )
151+ p .add_argument ("--rate" , type = float , default = 20e6 ,
152+ help = "sample rate for the drop/jitter tests (default 20e6)" )
153+ p .add_argument ("--seconds" , type = float , default = 6.0 ,
154+ help = "capture duration for drop/jitter tests (default 6s; "
155+ "the first ~1.5s is discarded as warm-up)" )
156+ p .add_argument ("--no-device" , action = "store_true" ,
157+ help = "decode benchmark only; no board needed" )
158+ args = p .parse_args ()
159+
160+ h = HackRF (tools_dir = args .tools_dir )
161+ bench_decode (h )
162+
163+ if args .no_device :
164+ return 0
165+
166+ det = h .detect ()
167+ if not det ["ready" ]:
168+ print (f"\n no usable HackRF for device tests: { det ['problem' ]} ,
169+ file = sys .stderr )
170+ return 1
171+ print (f"\n [*] board: firmware { det ['boards' ][0 ].get ('firmware' )}
172+ f"tools { det ['tools_version' ]} )
173+ for w in det .get ("warnings" , []):
174+ print (f" ! { w } )
175+
176+ bench_drop_test (h , args .rate , args .seconds )
177+ bench_callback_jitter (h , args .rate , min (args .seconds , 2.0 ))
178+
179+ print ("\n == done ==" )
180+ print (" These numbers are specific to this board/machine/USB/load." )
181+ print (" Re-run a few times; they're honest measurements, not universal." )
182+ return 0
183+
184+
185+ if __name__ == "__main__" :
186+ sys .exit (main ())
0 commit comments