1、前言

学会使用 GNU Radio 中的 ZMQ，是从低阶使用者向高阶迈进的第一步！

因为学会了 ZMQ，就可以将 GNU Radio 中的实时数据流通过 socket 引到外面的 python/c 等大型应用程序中，做数据分析、展示、人工智能等。

来自 ZeroMQ 官方介绍：ZeroMQ (0MQ, ZMQ)，看起来像是一个可嵌入的网络库，同时起到了并发框架的作用。它为您提供了在进程内、进程间、TCP和多播等各种传输中承载原子消息的 socket 。

2、ZMQ 块的类型

Data Blocks：
ZMQ data blocks 传输原始流数据；没有格式化。数据类型和采样率由馈送 ZMQSink 的流程图确定。因此，接收数据的流程图或程序必须知道这些参数，以便正确地解释数据。

Message Blocks
不像普通的 ZeroMQ 字符串，GNU Radio ZMQ Message Blocks 使用 PMT 对数据进行编码和解码。

3、ZMQ 块的使用

• ZMQ 块的用户应该对ZeroMQ有一些熟悉。特别是，应该认识到ZMQ套接字和BSD套接字之间的区别。有关概述，请参阅 ZMQ Socket API。
• ZMQ 块使用 endpoints 来描述 ZMQ 应该如何传递数据。常见 endpoints 使用 TCP 传输数据，当然也可以采用其他协议。想要了解不同协议的 endpoints，可以参阅 zmq_tcp 和 zmq_ipc。
• 可以在49152–65535范围内分配专用端口。
• 不建议在单个流程图中使用ZMQ块，因为 Virtual_Source 和 Virtual_Sink 块的效率要高得多。

TCP Bind vs Connect

一些用户可能会想直接连接到GNU Radio ZMQ Blocks。虽然这是可能的，但需要谨慎。

首先要注意，在任何拓扑中，必须有一个到给定端点的绑定，而可能有多个到同一端点的连接。（A-B 之间绑定一次，可能会出现多个连接）

在 GNU Radio 中，stream sinks bind and stream sources connect。Message blocks 取决于参数设置。

还要注意：TCP端点的语义在绑定和连接之间有所不同。

TCP Bind

当绑定一个 TCP 端点时，您可以指定要侦听的连接点。

如果您指定了一个IP地址，则说明 socket 只接受与该地址相关联的网络上的连接（例如：127.0.0.1 or 192.168.1.123）

在某些情况下，您可能希望在连接到节点的所有网络上进行侦听。对于 GNU Radio，您应该使用 0.0.0.0 作为通配符地址；尽管 ZMQ 接受 * 作为通配符，但它并不是在所有情况下都能很好地工作。因此，您可以选择绑定到 tcp://0.0.0.0:54321

请注意，如果您没有输入IP地址，bind 会将该值视为网络适配器名称（例如 eth0）。详细信息参阅：zmq_tcp

TCP Connect

连接 TCP 端点时，您可以指定要连接的远程端点。您可以指定 IP 地址或 DNS 可解析名称。

Wire Format

ZMQ stream blocks 具有传递标记的选项。此外，PUB/SUB块支持过滤。这两个选项都会影响 ZMQ-wire 协议。

当过滤器字符串被提供给 PUB/SUB 块时，GNU Radio 使用多部分消息来发送过滤器字符串，然后是有效载荷。尝试与 GNU Radio ZMQ 块接口的非 GNU 无线电代码必须为此部分准备好，并将其丢弃。请注意，只有在指定了非空筛选器的情况下，发送方才会发送此消息部分。

接下来，如果启用了发送标签，则要发送的数据窗口内的任何标签都将以特殊格式编码，并在有效载荷数据之前进行预处理。如果未启用标记，则会忽略此标头。

这两个特征使得发送器配置与接收器配置的匹配变得至关重要。否则将导致流程图中出现运行时错误。

4、DEMO

4.1 同一台电脑上的两个流程图

在同一个电脑上时，localhost 的 IP 地址应为 127.0.0.1。它的开销比完整的IP要小。

下面使用 PUB/SUB 的流程图来自 Simulation_example:_AM_transmitter_and_receiver：

这个流程图我也在 B 站上面有详细的视频介绍：GNU Radio 系列教程（十二）－－ 窄带 FM 收发系统（基于ZMQ模拟射频发送）

4.2 不同电脑上的两个流程图

在不同电脑上时，则必须在该连接的每一端指定接收块（sink block) 的 IP 和端口号。例如，如果 Sink 位于 IP 192.168.1.194:50241，Source 位于 IP 192.168.1.85，则 Source 和 Sink 块都必须指定 Sink IP 和端口 192.168.1.194:00241。

4.3 作为 REQ/REP 服务器的 Python 程序

下面的 Python 程序在其 REQ socket 上接收字符串消息，将文字变成大写，然后在其 REP socket 上发送出去。术语在这里变得混乱，因为传入的 REQ 来自 GR ZMQ_REP_Message_Sink，并返回到 ZMQ_REQ_Message_Source。

只需记住： sink 是流程图的终点，source 是流程图的起点。

#!/usr/bin/python3
# -*- coding: utf-8 -*-

# zmq_REQ_REP_server.py

# This server program capitalizes received strings and returns them.
# NOTES:
#   1) To comply with the GNU Radio view, messages are received on the REQ socket and sent on the REP socket.
#   2) The REQ and REP messages must be on separate port numbers.

import pmt
import zmq

_debug = 0          # set to zero to turn off diagnostics

# create a REQ socket
_PROTOCOL = "tcp://"
_SERVER = "127.0.0.1"          # localhost
_REQ_PORT = ":50246"
_REQ_ADDR = _PROTOCOL + _SERVER + _REQ_PORT
if (_debug):
    print ("'zmq_REQ_REP_server' version 20056.1 connecting to:", _REQ_ADDR)
req_context = zmq.Context()
if (_debug):
    assert (req_context)
req_sock = req_context.socket (zmq.REQ)
if (_debug):
    assert (req_sock)
rc = req_sock.connect (_REQ_ADDR)
if (_debug):
    assert (rc == None)

# create a REP socket
_PROTOCOL = "tcp://"
_SERVER = "127.0.0.1"          # localhost
_REP_PORT = ":50247"
_REP_ADDR = _PROTOCOL + _SERVER + _REP_PORT
if (_debug):
    print ("'zmq_REQ_REP_server' version 20056.1 binding to:", _REP_ADDR)
rep_context = zmq.Context()
if (_debug):
    assert (rep_context)
rep_sock = rep_context.socket (zmq.REP)
if (_debug):
    assert (rep_sock)
rc = rep_sock.bind (_REP_ADDR)
if (_debug):
    assert (rc == None)

while True:
    #  Wait for next request from client
    data = req_sock.recv()
    message = pmt.to_python(pmt.deserialize_str(data))
    print("Received request: %s" % message)

    output = message.upper()

    #  Send reply back to client
    rep_sock.send (pmt.serialize_str(pmt.to_pmt(output)))

安装 NetCat：方便我们测试 TCP
-《NetCat使用指南》
-《Sending TCP/UDP packets using Netcat》
-《Simple client / server with nc not working》
注意，这鬼软件有好几个不同的软件，我用的是 openbsd-netcat

sudo pacman -S openbsd-netcat

上面代码：

while 循环中用 REQ 等待接收，然后转为大写，用 REP 发送出去:（比较坑的是，我用 netcat 建立 tcp 服务器和客户端，无法与上面 python 脚本通信，似乎一启动，建立连接，server 就异常退出了，最终还是得用 GNN Radio 开启两个 ZMQ 工程，然后与这个 python 脚本通信，整体信息流如下：）

4.4 作为 PUSH/PULL 服务器的 Python 程序

与上面 demo 类似，是基于 PUSH/PULL 传递消息。

#!/usr/bin/python3
# -*- coding: utf-8 -*-

# zmq_PUSH_PULL_server.py

import sys
import pmt
import zmq

_debug = 0          # set to zero to turn off diagnostics

# create a PUSH socket
_PROTOCOL = "tcp://"
_SERVER = "127.0.0.1"          # localhost
_PUSH_PORT = ":50252"
_PUSH_ADDR = _PROTOCOL + _SERVER + _PUSH_PORT
if (_debug):
    print ("'zmq_PUSH_PULL_server' version 20068.1 binding to:", _PUSH_ADDR)
push_context = zmq.Context()
if (_debug):
    assert (push_context)
push_sock = push_context.socket (zmq.PUSH)
if (_debug):
    assert (push_sock)
rc = push_sock.bind (_PUSH_ADDR)
if (_debug):
    assert (rc == None)

# create a PULL socket
_PROTOCOL = "tcp://"
_SERVER = "127.0.0.1"          # localhost
_PULL_PORT = ":50251"
_PULL_ADDR = _PROTOCOL + _SERVER + _PULL_PORT
if (_debug):
    print ("'zmq_PUSH_PULL_server' connecting to:", _PULL_ADDR)
pull_context = zmq.Context()
if (_debug):
    assert (pull_context)
pull_sock = pull_context.socket (zmq.PULL)
if (_debug):
    assert (pull_sock)
rc = pull_sock.connect (_PULL_ADDR)
if (_debug):
    assert (rc == None)

while True:
    #  Wait for next request from client
    data = pull_sock.recv()
    message = pmt.to_python(pmt.deserialize_str(data))
    # print("Received request: %s" % message)

    output = message.upper()    # capitalize message

    #  Send reply back to client
    push_sock.send (pmt.serialize_str(pmt.to_pmt(output)))

4.5 处理流程图数据的 Python 程序

个 demo 是几乎贯穿后面 GNU Radio 高阶用法的最重要的 DEMO。 因为，通常情况下我们会使用 GNU Radio 进行信号处理，但希望数据流流入普通 python 程序，然后做丰富的数据分析等逻辑。这里，PUB 和 PUSH 可以让应用程序获得这些数据流。(这里我们将 127.0.0.1 换成了 *，这样能够让同一局域网内的设备都能访问)

一般的，流程图中采用 PUB/PUSH Sink，将数据送出：

然后，普通 python 脚本就可以对其进行 recv：

#!/usr/bin/python3
# -*- coding: utf-8 -*-

# zmq_SUB_proc.py
# Author: Marc Lichtman

import zmq
import numpy as np
import time
import matplotlib.pyplot as plt

context = zmq.Context()
socket = context.socket(zmq.SUB)
socket.connect("tcp://127.0.0.1:55555") # connect, not bind, the PUB will bind, only 1 can bind
socket.setsockopt(zmq.SUBSCRIBE, b'') # subscribe to topic of all (needed or else it won't work)

while True:
    if socket.poll(10) != 0: # check if there is a message on the socket
        msg = socket.recv() # grab the message
        print(len(msg)) # size of msg
        data = np.frombuffer(msg, dtype=np.complex64, count=-1) # make sure to use correct data type (complex64 or float32); '-1' means read all data in the buffer
        print(data[0:10])
        # plt.plot(np.real(data))
        # plt.plot(np.imag(data))
        # plt.show()
    else:
        time.sleep(0.1) # wait 100ms and try again

参考链接

[1]. GNU Radio 系列教程（一） —— 什么是 GNU Radio
[2]. GNU Radio 系列教程（二） —— 绘制第一个信号分析流程图
[3]. GNU Radio 系列教程（三） —— 变量的使用
[4]. GNU Radio 系列教程（四） —— 比特的打包与解包
[5]. GNU Radio 系列教程（五） —— 流和向量
[6]. GNU Radio 系列教程（六） —— 基于层创建自己的块
[7]. GNU Radio 系列教程（七）—— 创建第一个块
[8]. GNU Radio 系列教程（八）—— 创建能处理向量的 Python 块
[9]. GNU Radio 系列教程（九）—— Python 块的消息传递
[10]. GNU Radio 系列教程（十）—— Python 块的 Tags
[11]. GNU Radio 系列教程（十一）—— 低通滤波器
[12]. GNU Radio 系列教程（十二）—— 窄带 FM 收发系统（基于ZMQ模拟射频发送）
[13]. GNU Radio 系列教程（十三）—— 用两个 HackRF 实现 FM 收发
[14]. SDR 教程实战 —— 利用 GNU Radio + HackRF 做 FM 收音机
[15]. SDR 教程实战 —— 利用 GNU Radio + HackRF 做蓝牙定频测试工具（超低成本）

: 如果觉得不错，帮忙点个支持哈～