[SDR] GNU Radio 系列教程（十四） —— GNU Radio 低階到高階用法的分水嶺 ZMQ 的使用詳解

1、前言
2、ZMQ 塊的類型
3、ZMQ 塊的使用
4、DEMO
參考鏈接

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

參考鏈接

: 如果覺得不錯，幫忙點個支持哈～

posted @ 2023-04-26 02:22 beautifulzzzz 閱讀(2802) 評論(4) 收藏舉報

刷新頁面返回頂部

beautifulzzzz

創造傳奇，不是賣藝！