初级教程（3）创建和修改 Python 块3-Python 块消息传递

2年 ago

g_turing

2 minutes

3.Python 块消息传递

本教程介绍如何使用Embedded Python Block读取和写入消息。

上一教程Python Block with Vectors演示了如何编写带有矢量输入和输出的嵌入式 Python Block。下一个教程Python 块标签描述了如何在 Python 块中读取和写入标签。

内容

消息概述

消息是一种在块之间发送信息的异步方式。消息擅长传递控制数据，跨块保持一致的状态，并向流程图中的块提供某种形式的非数据反馈。

消息有几个独特的属性：

消息到达时没有基于采样时钟的保证
消息不与标签等特定样本相关联
消息输入输出端口在GRC中不必连接
消息端口使用多态类型 (PMT)

消息端口用灰色表示，它们的连接用虚线区分：

有关使用 PMT 传递消息的更多信息，请参见此处：消息传递

流程图概述

以下流程图演示了如何：

给Python块添加消息收发端口
发送信息
接收和处理消息
根据收到的消息调整work()函数中的块行为

创建两个自定义嵌入式 Python 块以：

根据收到的消息选择或多路复用两个输入信号之一
计算样本数量并向多路复用块发送消息以切换输入

本教程假设您已经创建了至少一个Embedded Python Block。如果没有，请在继续之前完成创建您的第一个块的教程。

首先将以下块添加到流程图并连接它们：

噪声源
信号源
蟒蛇块
风门
QT GUI 时间接收器

多路复用器：定义块

双击Embedded Python Block并在编辑器中打开源代码：

不需要 example_param，因此从__init __()函数中删除变量example_param：

def __init__(self): # 这里只有默认参数

并删除该行：

self.example_param = example_param

将块的名称更改为Multiplexer：

name='多路复用器',

向块添加第二个输入：

in_sig=[np.complex64, np.complex64],

删除乘以example_param：

output_items[0][:] = input_items[0]

回想一下，Python 需要适当的缩进。默认情况下，嵌入式 Python 块使用 4 个空格的倍数缩进。混用制表符和空格会引发语法错误：

保存代码 (CTRL+S) 并返回 GRC。请注意块的名称发生了怎样的变化，并且块现在有两个输入。将噪声源和信号源连接到两个输入：

多路复用器：定义消息输入端口

返回代码编辑器。需要添加一个输入消息端口。创建一个变量来存储消息端口名称：

self.selectPortName = 'selectPort'

添加一行以创建或注册消息输入端口：

self.message_port_register_in(pmt.intern(self.selectPortName))

添加一行以将输入端口与消息处理程序连接起来。

self.set_msg_handler（pmt.intern（self.selectPortName），self.handle_msg）

保存代码 (CTRL+S)。请注意，语法错误列在Embedded Python Block的属性中：

此错误表示需要导入pmt库。返回代码编辑器并添加正确的导入语句：

导入PMT

多路复用器：创建消息处理程序

消息处理程序是在收到消息时调用的函数。

必须定义消息处理函数。此消息处理程序根据收到的消息在两个输入端口之间切换。收到的消息是一个布尔值，即 True 或 False。在__init__()下定义一个新变量，它是输入选择器，

self.selector = True

定义handle_msg()函数：

def handle_msg（自我，味精）：
    self.selector = pmt.to_bool(msg)

函数pmt.to_bool()获取消息 PMT，然后将数据类型转换为 Python 的布尔数据类型。PMT 用于将消息传递给抽象数据类型。例如，消息可用于发送和接收字符串、浮点数、整数甚至列表。有关 PMT 的更多信息，请访问多态类型 (PMT) 维基页面。

多路复用器：在work()中使用消息

多路复用器的外部接口是完整的。修改块的work()函数以添加多路复用操作。将以下代码添加到work()函数中：

如果（自我选择器）：
    output_items[0][:] = input_items[0]
别的：
    output_items[0][:] = input_items[1]

如果self.selector = True 多路复用器块选择端口0 ，如果self.selector = False则选择端口1。self.selector的默认值在__init__()函数中定义。

保存代码 (CTRL+S) 并返回 GRC。Multiplexer块有一个消息端口selectPort：

在继续之前运行流程图以确保一切正确。如简介中所述，不必连接消息端口即可运行流图。因为self.selector的默认值为True ，多路复用器的work()函数将选择端口0并将其发送到输出。QT GUI Time Sink显示噪音：

选择器控件：定义块

另一个嵌入式 Python 块用于计算它收到的样本数，然后将控制消息发送到多路复用器块以切换选择器。

首先在Multiplexer和Throttle之间的流程图中添加一个新的Python 块。

警告！从块库中拖放一个新的 Python 块！不要复制和粘贴现有的多路复用器块，它只会创建该块的第二个副本。

编辑Embedded Python Block 的代码。更改__init__()函数中的参数example_param：

def __init__(self, Num_Samples_To_Count=128):

更改块的名称：

name='选择器控件',

将Num_Samples_To_Count存储为私有变量：

self.Num_Samples_To_Count = Num_Samples_To_Count

移除work()函数中的example_param乘法：

选择器控件：定义消息输出端口

导入pmt库：

导入PMT

在__init__()函数中创建一个变量 ( self.portName ) ，其中包含字符串形式的输出端口名称messageOutput：

self.portName = 'messageOutput'

通过在__init__()函数中添加以下行来创建或注册消息端口：

self.message_port_register_out（pmt.intern（self.portName））

保存代码 (CTRL + S) 并返回 GRC。选择器控制块有一个消息输出端口：

选择器控件：在work()中发送消息

不需要为输出端口定义消息处理程序。但是，需要修改 work()函数以创建发送消息的逻辑。

首先在__init__()中创建两个变量：

self.state = 真
self.counter = 0

添加行以增加每次调用work()的计数样本数：

self.counter = self.counter + len(output_items[0])

添加逻辑以在超过计数器后发送消息：

如果（self.counter > self.Num_Samples_To_Count）：
    PMT_msg = pmt.from_bool(self.state)
    self.message_port_pub（pmt.intern（self.portName），PMT_msg）
    self.state = not(self.state)
    self.counter = 0

该逻辑使用pmt.from_bool()函数调用将self.state的 Python 布尔数据类型转换为 PMT ，然后在输出消息端口上发送或发布消息。self.state变量被切换到它的相反值并且计数器被重置。

保存代码并返回到 GRC。在Selector Control块的属性中为Num_Samples_To_Count输入32000：

添加Message Debug块并将messageOutput端口连接到它。运行流程图显示消息正在以每秒一次的速率发送，在#t (True) 和#f (False) 之间交替：

但是，选择器控件的输出消息端口尚未连接到多路复用器的输入消息端口，因此QT GUI Time Sink仅显示噪声。

最终流程图

将Selector Control的输出消息端口连接到Multiplexer的输入消息端口。请注意，虚线在两个块后面移动并且很难看到：

Virtual Sinks和Virtual Sources可用于清理一些连接并使流程图更易于理解。单击虚线并将其删除。将Virtual Sink和Virtual Source拖放到工作区中。将Virtual Sink和Virtual Source的Stream ID更改为message，然后在流程图中连接它们：

运行流程图。QT GUI Time Sink显示噪声源和信号源之间的交替输出：

下一个教程Python 块标签描述了如何在 Python 块中读取和写入标签。