如何在GNU Radio中用嵌入式Python块制作消息源与消息选通块

问题根因说明 #

GNU Radio 流图调度器仅在流端口有可处理样本时才会调用work方法，当in_sig和out_sig均为None时，调度器不会触发work执行，因此写在work中的消息发送代码永远不会运行。

完整可运行实现代码 #

import time
import threading
from gnuradio import gr
import pmt

class msg_block(gr.sync_block):
    def __init__(self, interval=1.0):  # interval参数为发送周期，单位秒
        gr.sync_block.__init__(
            self,
            name="dynamic_msg_strobe",
            in_sig=None,
            out_sig=None
        )
        # 注册消息输出端口
        self.message_port_register_out(pmt.intern('msg_out'))
        
        # 运行参数配置
        self.interval = interval
        self._stop_flag = threading.Event()
        self._sender_thread = None

    def _msg_sender(self):
        while not self._stop_flag.is_set():
            # ======================
            # 此处替换为自定义动态消息计算逻辑
            dynamic_content = f"生成的动态消息，时间戳: {time.time()}"
            # 转换为PMT格式，支持字符串、数值、向量等多种类型
            send_pmt = pmt.from_string(dynamic_content)
            # ======================
            
            # 发送消息到输出端口
            self.message_port_pub(pmt.intern('msg_out'), send_pmt)
            # 等待一个发送周期
            time.sleep(self.interval)

    def start(self):
        # 流图启动时自动启动发送线程
        self._stop_flag.clear()
        self._sender_thread = threading.Thread(target=self._msg_sender, daemon=True)
        self._sender_thread.start()
        return True

    def stop(self):
        # 流图停止时安全回收线程，避免后台僵尸进程
        self._stop_flag.set()
        if self._sender_thread is not None:
            self._sender_thread.join()
        return True

核心逻辑说明 #

• 独立线程实现发送逻辑：避开流调度的触发限制，完全自主控制发送时机
• 重写start/stop方法：和GNU Radio流图生命周期对齐，保证线程的安全启停，避免流图关闭时报错
• message_port_pub正确用法：第一个参数为注册输出端口时使用的PMT标识，第二个参数为PMT格式的消息，可根据需要调用pmt.from_int/pmt.from_float/pmt.to_pmt等方法转换Python原生变量为PMT格式
• 不推荐使用threading.Timer实现周期循环：单次Timer每次触发都要新建线程，线程管理成本高，单线程加sleep的方案更稳定可控

自定义修改指南 #

• 调整__init__的interval参数即可修改发送周期
• 替换_msg_sender函数中消息生成部分的代码，即可实现每次发送动态计算的变量

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