Python ZMQ Publisher对接GNUradio SUB Message Source块报错处理

开发过程中尝试通过Python代码向GNU Radio的SUB Message块发送数据时出现报错，最初排查方向为socket配置错误、数据需加自定义前导码，未在公开资料中找到对应解决方案。

初始报错场景 #

初始测试代码 #

import zmq
import time

ctx = zmq.Context()
socket = ctx.socket(zmq.PUB)
socket.bind("tcp://127.0.0.1:4444")

tab = [0xFF] * 256
tab = bytes(tab)
print("Starting loop...")
while True:
   socket.send(tab)
   time.sleep(5)

对应GNU Radio报错信息 #

Generating: '/home/X/test.py'
Executing: /usr/bin/python3 -u /home/X/test.py
Warning: failed to XInitThreads()
terminate called after throwing an instance of 'pmt::exception'
  what():  pmt::deserialize: malformed input stream, tag value = : 255
Done (return code -6)

对照测试验证 #

修改发送数组为全0x00后代码：

import zmq
import time

ctx = zmq.Context()
socket = ctx.socket(zmq.PUB)
socket.bind("tcp://127.0.0.1:4444")

tab = [0x00] * 256
tab = bytes(tab)
print("Starting loop...")
while True:
   socket.send(tab)
   time.sleep(5)

对应输出：

Generating: '/home/X/test.py' 
Executing: /usr/bin/python3 -u /home/X/test.py 
Warning: failed to XInitThreads()
******* MESSAGE DEBUG PRINT ********
#t
********************************************

修改发送数组为全0x01后输出：

Generating: '/home/X/test.py'
Executing: /usr/bin/python3 -u /home/X/test.py
Warning: failed to XInitThreads()
******* MESSAGE DEBUG PRINT ********
#f
************************************ 

测试发现直接发送裸字节时，全0被识别为PMT布尔值真（#t）、全1被识别为PMT布尔值假（#f），不会触发报错但传输的数据不符合预期。

根因说明 #

GNU Radio的SUB Message块默认接收的ZMQ消息必须为序列化后的PMT（GNU Radio自定义的多态数据类型）格式，直接发送原始字节会触发PMT反序列化异常，或被识别为其他类型的PMT数据。

最终可用代码 #

引入PMT库将数据转换为PMT对象并序列化后再发送即可正常传输：

import zmq
import time
import pmt

ctx = zmq.Context()
socket = ctx.socket(zmq.PUB)
socket.bind("tcp://127.0.0.1:4444")

tab = [0xFF] * 256
print("Starting loop...")
while True:
   socket.send(pmt.serialize_str(pmt.to_pmt(tab)))
   time.sleep(5)

修改后GNU Radio侧可正常接收到完整的256个值为255的元素组成的数组。

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