基于Hackrf的简单实时频谱显示

基于 python-hackrf 的实时频谱

简单的频谱显示.

在上一篇 blog 中我们看到了如何用 pyhackrf 库来控制 HACKRF 硬件返回接收的帧数据，那下面我们就来利用接收到的帧数据做一个最简单的实时频谱显示

构造线程

由于 HackRF 的 rx_callback 是在库内部的独立线程里被高频调用的(每次 256KB 数据,USB 传输速率很高)。如果我们在回调里直接调用画图函数,会导致: GUI 库(matplotlib/pyqtgraph)通常不是线程安全的,跨线程直接操作 GUI 控件会崩溃或产生诡异 bug、画图本身较慢(尤其是 FFT + 渲染),会拖慢回调返回速度,导致 USB 缓冲区堆积、丢帧甚至 libusb 报错 等问题

因此我们的做法是:回调线程只做最快的事——把数据丢进一个队列就立刻返回,真正耗时的处理(FFT、绘图)放到主线程/GUI 线程里异步进行。

因此在 Claude 的建议下，我们接收线程采用 python 的 Queue 机制来实现接收队列，同时画图线程用 pyQt5 的 pyqtgraph 实现（之前我自己的 demo 是用 matplotlib 实现，明显画图会更吃力）

构造 Hackrf_Receiver 线程

我们首先构造一个 class 类，叫做 Hackrf_Receiver，括号中的 threading.Thread 表示这个 class 继承自 threading.Thread，他拥有 Thread 所有的能力，同时可以继续添加自己的东西

class HackRF_Receiver(threading.Thread):

在这个 HackRF_Receiver 类中，我们首先来写构造函数，当创建 HackRF_Receiver 这个对象（类) 时，构造函数会首先被调用; 在 init 函数中，由于我们是继承自 Thread 类，因此必须要用 super().init 来把父类的初始化跑一次，同时 daemon=True 表示当主程序结束时，这个线程也会被同时结束，不会卡住主程序

其他的比如 data_queue 就是我们在这个类中基于父类新增的内容了。在 init 最后，我们再创建了一个最重要的 threading.Event，threding.Event 内部就是一个布尔标志位 + 一套线程安全的等待 / 唤醒机制，我们在这个类中主要就是使用它的 .wait() 和 .set() 功能，其中 .wait 功能会将函数阻塞在这里，但不占用 CPU 线程，而 .set 则会唤醒函数，退出 .wait 状态

注意，这里 self._stop_event 实际只是一个名字，我们可以用任何名字来替代它，它的本质就是创建 threading.Event()，并使用 Event 中的 .wait() 和 .set() 两个功能

def __init__(self, data_queue: queue.Queue):
    super().__init__(daemon=True)  # daemon=True,主程序退出时线程自动结束
    self.data_queue = data_queue
    self.device = None
    self._stop_event = threading.Event()

接下来就可以定义 callback 函数，这个 callback 函数和之前的是一样的，唯一的区别是我们增加了一个 put_nowait() 函数来将收到的 iq samples 放到队列 queue 当中，put_nowait 表示如果操作不能完成，就立刻抛异常，而不是一直等待

def rx_callback(self, device, buffer: np.ndarray, buffer_length: int, valid_length: int):
    """
    HackRF 硬件每次填满一个缓冲区就会调用这里。
    注意: 这里绝对不能做耗时操作(比如 FFT、画图、print 太多),
    否则会拖慢返回速度,导致 USB 缓冲区溢出丢包。
    """
    try:
        accepted = valid_length // 2 # I 8bits, Q 8bits
        # buffer 是 int8 类型的 IQ 交织数据,拷贝一份放入队列
        # (拷贝是必须的,因为原始 buffer 内存会被库复用,不拷贝数据会被覆盖)
        iq_bytes = buffer[:valid_length].copy().astype(np.int8)
        iq_samples = iq_bytes[0::2] + 1j * iq_bytes[1::2]  # Convert to complex type (de-interleave the IQ)

        self.data_queue.put_nowait(iq_samples)
    except queue.Full:
        print('Queue is Full, Drop Frame ! \n')
        # 队列满,说明消费端(绘图)处理不过来,直接丢弃本帧,不阻塞接收
        pass
    return 0  # 返回0表示继续接收

然后就是改写的 THreading 函数 .run()，在 .run() 中我们可以理解成他是类似于这个 class 里面的 main 函数，我们首先会初始化 HackRF 相关的东西，然后我们通过 self._stop_event.wait() 来将 .run() 阻塞在 hackrf_start_rx() 之后，类似于之前 blog 中的 while 循环，这样就可以让 HackRF 一直处于 RX 状态，这样的好处是等待时不占用 CPU ！而这个 wait 会直到 _stop_event.set() 函数的到来，才会停止阻塞，继续执行后面的操作，可以看到在 self._stop_event.wait() 后，就是 stop_rx 的操作了

def run(self):
    pyhackrf.pyhackrf_init()
    self.device = pyhackrf.pyhackrf_open()

    self.device.pyhackrf_set_sample_rate(SAMPLE_RATE)
    self.device.pyhackrf_set_freq(int(CENTER_FREQ))
    self.device.pyhackrf_set_amp_enable(False)
    self.device.pyhackrf_set_lna_gain(LNA_GAIN)
    self.device.pyhackrf_set_vga_gain(VGA_GAIN)

    self.device.set_rx_callback(self.rx_callback)
    self.device.pyhackrf_start_rx()
    print("[HackRF Receiver] 接收已启动")

    # 用 Event 等待停止信号,而不是 while True + sleep 空转
    self._stop_event.wait()

    self.device.pyhackrf_stop_rx()
    self.device.pyhackrf_close()
    pyhackrf.pyhackrf_exit()
    print("[HackRF Receiver] 接收已停止")

最后的 stop 函数就很简单了，直接调用 Event().set() 即可；至此我们就完成了整个 HackRF_Receiver 类的构造，其核心就是继承自 threading.Thread 类，然后用 Thread.Event() 中的 wait() 和 set() 来阻塞 / 唤醒程序，使 HackRF 保持在 RX 状态；最后在 HackRF 的 rxcallback 函数中用 queue 来存储 IQ 数据

最后，如果需要进一步的了解 Thread，可以直接在 IDE 里面跳转到 threding.Thread 函数，进一步看父类的构造

构造 PlotWindow 线程

类似的，plotWIndow 类继承自 QtWidgets.QMainWindow，其作用就是从队列中读数据并画图，其原理和前面的 HackRF Receiver 一样，都是继承了父类的 feature，同时自己还可以增加功能，因此在 init 中还是需要首先执行 super().init() 来初始化父类

这里有一个关键是，在 plotWindow 的 init 中，我们还传入了 data_queue 和 rxworker 两个对象，这里 PlotWindow 不负责创建这两个对象,只是”借用”外面传进来的——这是一种常见的设计习惯,叫依赖注入:一个类需要用到的外部资源,通过构造函数参数传进来,而不是自己在内部创建。好处是这个类更容易复用/测试,比如以后想让两个 PlotWindow 共用同一个队列,直接传同一个对象即可。

在 init 中还有一个重点是定时器功能，这个定时器利用 QtTimer 来计时，timer.start(TIME_PERIOD) 表示每 TIME_PERIOD 时间触发一次 timer out 事件，timer.timeout.connect(self.update_plot) 用于指定 timerout 时间对应的关联函数，也就是每次 timer out 之后执行 update_plot 函数

另外 main 函数的 app.exec_() 会在后台不断的检查各种信号，就包括了 QtTimer 的 timerout 信号，然后就会执行其对应的关联函数 update plot；所以准确的说，这个 QtTimer 是在主函数 main 中执行的

def __init__(self, data_queue: queue.Queue, rx_worker: HackRF_Receiver):
    super().__init__()
    self.data_queue = data_queue
    self.rx_worker = rx_worker
    ...
    # 定时器:主线程内定期"拉取"数据并刷新绘图,不会阻塞/被阻塞
    self.timer = QtCore.QTimer()
    self.timer.timeout.connect(self.update_plot)
    self.timer.start(PLOT_INTERVAL_MS)

接下来就是最重要的 update plot 函数，在 update plot 函数中，我们使用 data_queue.get_nowait() 来从 queue 中获取 data，get_nowait() 是指当 get 出现 exception 的时候不会等待，直接跳转到 exeception 中，也就是跳转到 empty 中，所以总的来说，这个 update_plot 里面的 while True 循环会不断尝试从队列取数据,每取一次就把 latest_iq_bytes 覆盖成最新取到的那一份,直到队列空了(抛出 queue.Empty)才用 break 跳出循环。 最终效果:不管队列里积压了多少帧数据,这次 update_plot 执行完,latest_iq_bytes 里存的永远是”当前时刻队列里最新的那一帧”,之前的全部被丢弃

def update_plot(self):
    # 一次性把队列里现有的数据都取出来(避免积压),只保留最新的用于绘图
    latest_iq_bytes = None
    got_any = False
    while True:
        try:
            latest_iq_bytes = self.data_queue.get_nowait()
            got_any = True
            self.frame_count += 1
        except queue.Empty:
            break

    if not got_any or latest_iq_bytes is None:
        return  # 本次没有新数据,跳过这一帧刷新

最后就是 close event 函数了，用于处理我们关闭了绘图窗口，这个时候就会调用 HackRF_Receiver 类中的 stop 方法，来关闭 HackRF RX 进程

def closeEvent(self, event):
    # 关闭窗口时,通知接收线程停止
    print("[PlotWindow] 窗口关闭,停止接收线程...")
    self.rx_worker.stop()
    self.rx_worker.join(timeout=3)
    event.accept()

Main 函数

main 函数就相对很简单了，由于两个类都会从外部引入 Queue，所以我们会在 main 函数中创建 queue.Queue(maxsize=QUEUE_MAXSIZE) 并传入到两个 class 中

然后我们创建一个 HackRF_Receiver 类，并调用其 .start() 函数，这里比较重要的是可以看到我们在定义 HackRF Receiver 类的时候，并没有写任何 .start() 函数，这就是因为我们的 Receiver 类是继承 Thread 类，而 start() 是 Thread 类中的标准函数，在 THread 类函数的注释中可以看到，这个 .start() 函数在每个 thread 类中必须被执行最多一次

最后就是创建一个 QtWidgets.QApplication，然后创建 PlotWindow 类，最后通过 app.exec_() 来触发整个流程开始，到这里就完成了整个实时频谱的 demo

More

可以看到，两个线程的配合大体是这样的，即每 30ms 左右（由 PLOT_INTERVAL_MS 定义）plot class 会调用 update_plot 函数把 Queue 中积累的所有数据读出来，然后同时 HackRF Receiver 类就会不断的采数据放到 queue 中，所以我们好奇 30ms 到底积累了多少数据？所以我们可以增加一些打印来看看每 30ms 调用 update_plot 的时候到底积累了多少数据，通过下面的方法我们打印可以看到，除了第一次以外，其他时候几乎队列积压=1 或 0，这就说明 HACKRF 吐数据的速度比 update_plot 读数据的速度要慢

drained_count = 0
qsize_before = self.data_queue.qsize()
while True:
    try:
        latest_iq_bytes = self.data_queue.get_nowait()
        got_any = True
        self.frame_count += 1
        drained_count += 1
    except queue.Empty:
        break

print(f"[PlotWindow] 本次取出帧数={drained_count}, 取出前队列积压={qsize_before}")

实际上理论计算也如此，因为我们采用的是 2MHz 采样率，而 HackRF 每次传输 131072 个 IQ data，所以采用 131072 个 IQ data 需要 131072*1/2MHz = 65ms，所以当我们以 30ms 的速度 update_plot，确实会出现两次里面一次读到为 0 的情况，也就是 HACKRF 采样更慢，但如果我们 HACKRF 采样率变到 8MHz，则会出现一定的数据挤压了，相当于吐出一次数据仅需要 16ms，而我们是每 30ms 才读一次

• sdr
• cn