1、前言

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

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

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

2、ZMQ 块的类型

SINK	SOURCE	特征
PUB	SUB	广播，可一对多
PUSH	PULL	点播，点对点对等网络
REQ	REP	点对点链路，一个请求一个回复，类似客户端服务器

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

上面代码：

kind	port	method	func	C/S
REQ	50246	connect	recv()	server
REP	50247	bind	send()	client

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

参考链接

---

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