Python 代码小组件#

Python 代码小组件允许用户在 Python 中构建应用程序的某些部分。 这也允许用户在不创建自定义扩展的情况下添加自定义实现。 本节概述了如何在 Python 中实现代码小组件以及如何在应用程序中使用它们。

代码小组件是用户实现的单元，可以作为图的一部分运行。 Python 代码小组件允许用户在 Python 中实现此功能。 与 C++ 代码小组件类似，Python 代码小组件具有以下用户必须实现的功能

  • start() - 代码小组件启动时调用一次

  • tick() - 每次 tick 时调用

  • stop() - 代码小组件停止时调用一次

与 C++ 代码小组件不同，所有 Python 代码小组件都通过相同的接口注册，即： nvidia::gxf::PyCodeletV0。

要在 Python 中实现代码小组件，用户必须实现类 CodeletAdapter。 如上所述，用户在 Python 中实现 start()、 stop() 和 tick() 方法。

运行包含 Python 代码小组件的应用程序必须使用 gxe.py 而不是 gxe 二进制文件。 包含 Python 代码小组件的图无法直接使用 gxe 二进制文件运行，因为必须先启动 Python 解释器才能运行任何 Python 代码。

如果已安装 graph-composer 和 registry，也可以运行图。 使用以下命令安装相关扩展

registry graph install -g path/to/graph.yaml -m output/path/to/generated/manifest.yaml -d path/to/target.yaml --output-directory output/directory

示例目标文件可以在 /opt/nvidia/graph-composer/ 中找到。

以上命令会将相关扩展和其他必需文件安装到 output/directory。

复制核心扩展并运行以下命令

cp -r /opt/nvidia/graph-composer/core output/directory/gxf

最后，使用以下命令将 output/directory 添加到 PYTHONPATH

export PYTHONPATH=output/directory

使用 gxe.py 运行图

python3 output/directory/gxf/std/gxe.py --app path/to/graph.yaml --manifest path/to/manifest/file.yaml

以下是一些 Python 代码小组件示例，可以在 gxf/python/tests 中找到

Ping Codelets 演示了 Python 代码小组件的基本用法

  1. PingTx.py

    • 构造一个空消息，该消息是一个 Entity

    • 在第一个发射器上发送它

  2. PingRx.py

    • 在第一个发射器上接收消息

    • 如果为非空则通过，否则引发异常

在 Python 中获取张量数据

  1. VerifyEqual.py

    • 此代码小组件有两个接收器。

    • 它从每个接收器接收两条消息。

    • 从两条消息中提取张量数据。

    • 如果张量在设备上，则将数据复制到主机上。

    • 断言张量上的数据相等。

从 Python 生成张量数据

  1. StreamGenerator

    • 在每次 tick 时，此代码小组件生成四个张量：两个在主机上，两个在设备上。

    • 它使用 TensorDescription 对象将张量重塑为所需的形状。

    • 创建主机消息和设备消息。

    • 将设备张量添加到设备消息，并将主机张量添加到主机消息

    • 在第一个发射器上发布主机消息，在第二个发射器上发布设备消息。

请参阅 gxf/python/tests 目录中的示例图文件，以了解如何在应用程序中使用 Python 代码小组件。

通用概念#

Python 使用户能够以非常少的样板代码添加大量功能，因此在机器学习社区中得到广泛使用。

例如，模拟传感器获取数据、可视化生成的输出或在数据上运行各种 ML 模型都可以很容易地在 Python 中实现。 Python 代码小组件允许用户在 Python 中实现这些功能。

与 C++ 代码小组件不同，Python 代码小组件不是单独注册的。 要创建 Python 代码小组件，用户需要实现 CodeletAdapter（一个 Python 基类），并且所有 Python 代码小组件都在注册表中注册为 PyCodeletV0，这是一个经典的 C++ 代码小组件。 此 C++ 代码小组件调用 Python 代码小组件实现的 start()、 tick() 和 stop() 方法。

PyCodeletV0

用户的 Python 代码小组件

start()

start()

tick()

tick()

stop()

stop()

实现 Python 代码小组件#

实现类#

要实现 Python 代码小组件，用户需要实现 CodeletAdapter。 这在 gxf.python_codelet.codelet 中可用。 具体来说，用户需要实现以下函数

start() - 设置代码小组件。 代码小组件启动时调用一次。

tick() - 业务逻辑。 根据调度术语调用。

stop() - 清理。 包含 Python 代码小组件的实体终止时调用。

将 Python 代码小组件添加到图#

将 Python 代码小组件添加到图与添加 C++ 代码小组件不同。 与 C++ 代码小组件不同，Python 代码小组件的 type 是 nvidia::gxf::PyCodeletV0，而不是 nvidia::gxf::Codelet。 代码小组件实现的链接直接通过参数完成。 此外，Python 代码小组件只能获取具有相应 Python 绑定的参数。

nvidia::gxf::PyCodeletV0 接受以下参数

参数

强制/可选

描述

codelet_name

强制

实现 CodeletAdapter 的用户 Python 代码小组件类的名称

codelet_file

强制

包含实现的文件的绝对路径

codelet_params

可选

用户可以解析以设置其他参数的字符串

以下是名为 rx 的实体，其中包含以下组件

  • 一个名为 python_receiver 的 Python 代码小组件

  • 一个名为 signal 的 DoubleBufferReceiver

  • 一个匿名的 MessageAvailableSchedulingTerm

Python 代码小组件的实现在 some/path/to/PythonCodelets.py 文件中，类名为 PingRx，它应该实现 CodeletAdapter。 Python 代码小组件还接受另外两个参数

  • receivers：包含单个项 signal 的列表，它是一个双缓冲接收器组件。

  • codelet_params：一个自定义字符串，用户可以在 start()、 tick() 或 stop() 方法中解析并相应地使用。

---
name: rx
components:
- name: signal
type: nvidia::gxf::DoubleBufferReceiver
- type: nvidia::gxf::MessageAvailableSchedulingTerm
parameters:
   receiver: signal
   min_size: 1
- name: python_receiver
type: nvidia::gxf::PyCodeletV0
parameters:
   codelet_name: "PingRx"
   codelet_file: "some/path/to/PingRx.py"
   codelet_params:
     log_count: 5
     receiver: signal

访问参数#

CodeletAdapter 有一个名为 get_params 的方法，该方法返回 YAML 文件中提及的所有参数的字典。

class SampleCodelet(CodeletAdapter):
"""
Sample class to show how to access params
"""
    def start(self):
        self.params = self.get_params()

    def tick(self):
        print(self.params['log_count'])

    def stop(self):
        return

访问其他组件#

用户还可以按如下方式访问其他组件，例如发射器和接收器

from gxf.std import Receiver

class SampleCodelet(CodeletAdapter):
"""
Sample class to show how to access params
"""
    def start(self):
        self.params = self.get_params()

    def tick(self):
        rx = Receiver.get(self.context(),\
                                self.cid(),\
                                self.params["receiver"])
        msg = rx.receive()

    def stop(self):
        return

CodeletAdapter 还实现了以下实用程序方法

方法

描述

eid()

返回包含此代码小组件的实体的唯一 ID。

cid()

返回代码小组件组件的唯一 ID。

name()

返回 Python 代码小组件的名称。

get_execution_timestamp()

返回代码小组件上次启动、tick 或停止时的纳秒时间戳。

get_execution_time()

与 get_execution_timestamp() 相同，返回以秒为单位的浮点数。

get_delta_time()

返回当前调用（start()、 tick() 或 stop()）与上次调用之间的时间差。

get_execution_count()

返回代码小组件已执行的次数。

is_first_tick()

如果 tick() 在 start() 之后首次调用，则返回 True。
在本页上