无线电安全基础-GNURADIO流程图入门

2025/4/5 无线电安全入门无线电安全入门 GNURADIO

回顾课堂学的信号系统，高频电子线路，数字信号处理，通信原理四大神书

一.最基础的流程图

信号与系统知识回顾

上面为时域图。时域的定义是描绘时间随信号如何变化

下面为频域图。频域用来描述信号包含哪些频率成分

时域：信号随时间怎么变化
（你看到的是波形——起伏、快慢、周期性）

频域：信号包含哪些频率成分、各有多强
（你看到的是频谱——哪些频率占主导，谁强谁弱）

从时域到频域使用的是傅里叶变换

二.比特的打包与解包

主要演示的是如何将原始字节比特打包成符号。

意义：把一串比特打包成一个数值（可以是十进制，也可以转成十六进制字符）

三.基于层创建 block

先设置一个流程图，接着点启动按钮旁边的按键保存

这里可以看到保存好了

四、滤波器的实现

原理

低通滤波器：只允许低频信号通过挡住高频信号

高通滤波器：只允许高频信号通过挡住低频信号

带通滤波器：只允许某一段频率范围的信号通过，其它全过滤

带阻滤波器：不允许某一段频率范围的信号通过，其它全接收

滤波器对不同频率信号的响应情况可以通过通频带、阻频带、截止频率等参数来描述。

五、ZMQ 的使用

ZMQ 最主要的作用是可以将 GNU Radio 中的实时数据流通过 socket 引到外面的 python/c 等大型应用程序中做数据分析、展示等。

https://www.cnblogs.com/zjutlitao/p/17354483.html

ZMQ 的几种应用场景

1.本机上两个不同的 gnuradio 文件之间相互通信。

以 AM 信号发送接收机为例

回顾一下上课知识

AM 调频的核心思想：

以下内容按顺序为从左到右

首先使用 audio source 输出一个采样率为 48Khz 的模拟信号表示 m(t)
通过 repeat 将 48KHZ 的音频信号插值到 768Khz。插值后的信号依然是m(t)，只不过变成了 768 个样点
Multiply Const（乘以 1 ）实际可以用来控制音频幅度
Add Const + 1 .这是因为由 AM 公式可得如果 m(t) 有负数，就可能使调制深度超过 100%，造成“过调制”所以加 1 是为了把信号偏移到正区间，避免过调制
Signal source（载波）频率和采样率与音频保持一致
Multiply（调制器）这一步实现了真正的调幅输出为 S(t)
结果输出到 ZMQ 和时域图

（注意调制信号和载波都要设置为相同的值）

在前面我们已经演示如何发送端输出已调信号。现在在接收端要做的事情是解调将其还原为最初的 mt 信号

那么 AM 最经典的解调方式是包络解调

首先 source 的信号就是发送机发送的 s(t)
Frequency Xlating FIR Filter（频率平移 + 滤波 + 降采样）中心设为 48Khz 再将频率搬移到基带 0hz 附近。使得目标信号在滤波器通带中心。接着使用低通滤波器只保留 [1 + m(t)] 的频带部分。降采样（Decimation=16）
AGC(自动增益控制)防止包络检测失真
Complex to Mag包络检测提取调幅信号的包络部分（调制内容）模块自动完成底层的计算工作
Band Pass Filter 低通滤波器过滤直流分量 1 只保留 m(t)部分
Multiply Const 同上音量调整
结果输出为时域图

2.不同电脑可以使用局域网进行传输

这里主要是由 REQ/REP 的方式传递信息原理如图所示

代码比较简单。先通过 ZMQ REQ socket 接收消息，再把消息转换为大写，再然后通过 ZMQ REP socket 发送回去消息。格式为 GNU Radio 的 PMT（Polymorphic Type）。但是这个过程是双向的请求响应，适用于客户端和服务器的应答

#!/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)))

再来看一段基于 PUSH_PULL 的消息传输脚本。使用 ZMQ PULL socket 接收 PMT 消息（从客户端或 GNU Radio）。将消息转换为大写（字符串处理）再通过 ZMQ PUSH socket 发送回去（可以广播给多个客户端。这构建的是一个 消息处理节点，支持多个接收者（fan-out）。

#!/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)))

如何将 GNURADIO 的数据进行导出做信号处理

该流程图其中 signal source 即是我们所收集到的信号。这个流程图可以使得信号通过 TCP 的方式在局域网传输

以下是接收脚本的例程，以方便做进一步的数据分析

#!/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