使用Head块终止GNU Radio流图后，如何重启以周期性保存样本？

运行GNU Radio流图时，希望每30秒将固定数量的样本保存到文件中，同时对已保存的样本进行信号处理。尝试用Head块实现，虽然能正常保存第一组样本，但会直接终止流图，无法继续保存后续样本。

原本打算用Head块的reset()方法解决，但根据GNU Radio邮件列表说明：

reset()仅在Head块达到指定样本数前有效，一旦达到指定数量，Head块完成任务，流图将终止，因此reset()无法实现预期效果。

相关代码片段：

blocks_head_0 = blocks.head(gr.sizeof_float*1, 12000*15)

def rx(stop_rx):
        while not stop_rx.isSet():
            print("\nStarting RX")
            check_time(rx_cycle)
            parse_rx()
            time.sleep(8)
            blocks_head_0.reset()
            print("\nExiting RX")

可行解决方案 #

1. 使用Repeat Head块（推荐） #

GNU Radio的gr-blocks库中有repeat_head块（部分版本需手动导入），它专门用于重复截取固定数量的样本，不会在达到样本数后终止流图。替换原有的Head块即可，无需额外reset操作，它会自动循环截取指定数量的样本。

示例代码替换：

blocks_repeat_head_0 = blocks.repeat_head(gr.sizeof_float*1, 12000*15)

2. 自定义可控的样本截取块 #

如果找不到repeat_head块，可以自定义一个简单的同步块，通过外部变量控制样本传输，达到固定数量后暂时阻塞，定时重置计数器后恢复传输：

class ControlledHead(gr.sync_block):
    def __init__(self, itemsize, count):
        gr.sync_block.__init__(self,
            name="ControlledHead",
            in_sig=[itemsize],
            out_sig=[itemsize])
        self.count = count
        self.counter = 0
        self.enable = True

    def work(self, input_items, output_items):
        if not self.enable:
            return 0  # 阻塞，不输出样本
        n = min(len(input_items[0]), len(output_items[0]), self.count - self.counter)
        if n == 0:
            self.enable = False  # 达到数量后禁用
            return 0
        output_items[0][:n] = input_items[0][:n]
        self.counter += n
        return n

    def reset(self):
        self.counter = 0
        self.enable = True

# 使用自定义块
custom_head = ControlledHead(gr.sizeof_float*1, 12000*15)

之后在rx函数中调用custom_head.reset()即可，流图不会终止。

3. 结合Stream Tag与Tag Gate块 #

通过定时生成Stream Tag，用Tag Gate块控制样本传输：

• 用Message Strobe或自定义定时模块，每30秒发送一个开启Tag；
• 用Tag Counter块统计通过的样本数，达到指定数量后发送关闭Tag；
• Tag Gate块根据Tag状态允许/阻塞样本流向文件Sink。
  这种方式灵活性更高，适合复杂流控制场景。

4. 定时切换File Sink文件 #

如果不需要严格在流内截取样本，也可以定时切换File Sink的输出文件：

• 保持流图持续运行，每30秒调用file_sink.open("new_file_path")切换文件；
• 配合blocks.keep_one_in_n或计数器，确保每个文件的样本数一致。

内容的提问来源于stack exchange，提问作者Jim