高级 Python 流量生成器 API - 2023.2 简体中文

Vitis 统一软件平台文档 应用加速开发 (UG1393)

Document ID
UG1393
Release Date
2023-12-13
Version
2023.2 简体中文

以下列表列示了高级 Python 流量生成器 API 及其用法。

提示: 在 GitHub 上的 External_IO_PY 教程中,提供了一个使用这些 API 的完整系统级示例,该示例可用于 Python 中的外部 I/O。
表 1. 外部流量生成器中使用的高级 Python API 列表
API 行为
ipc_axis_master_util("<sender_io_name>") 用于例化主流量生成器 (TG)
ipc_axis_slave_util("<receiver_io_name>") 用于例化从流量生成器
b_transport(packet) 用于发送 AXI4‑Stream 包的阻塞 API
sample_transaction() 该阻塞 API 用于接收 AXI4‑Stream 包。逐拍接收数据,每次 sample_transaction() 调用提供一拍。
ipc_axis_master_nb_util("<sender_io_name>") 用于例化非阻塞主 TG
ipc_axis_slave_nb_util("<receiver_io_name>") 用于例化非阻塞从 TG
nb_transport(packet) 用于发送 AXI4‑Stream 包的非阻塞 API 调用
nb_sample_transaction() 用于接收 AXI4‑Stream 包的非阻塞 API 调用。逐拍接收数据,每次 sample_transaction() 调用提供一拍。
no_of_available_txns() 用户获取针对非阻塞 TG 接收到的传输事务数量
end_of_simulation()

使用此 API 即可结束并退出仿真

首选在含 sw_emu/hw_emu 的外部流量生成器内使用

disconnect()

用于断开套接字的连接(DataIn、DataOut 等)

首选在含 x86simulatoraiesimulator 的外部流量生成器内使用

ipc_axis_master_util (<sender_io_name>) 

用于例化 Python 脚本中的发送器端口。

API 用法
使用 API 为发送器创建用户对象:
sender_obj = ipc_axis_master_util (<sender_io_name>) 
参数
发送数据的外部端口的名称:
<sender_io_name>
示例
#Instantiating AXI Master Utilities 
data_in = ipc_axis_master_util("DataIn")
DataIn 是外部端口名称,data_in 是用户对象。

ipc_axis_slave_util(<receiver_io_name>)

用于例化 Python 脚本中的接收器端口。

API 用法
使用 API 为接收器创建用户对象:
receiver_obj = ipc_axis_slave_util (<receiver_io_name>)
参数
接收数据的外部端口的名称:
receiver_io_name
示例
#Instantiating AXI Slave Utilities 
data_out = ipc_axis_slave_util("DataOut")
在此示例中,DataOut 是外部端口,data_out 是用户对象。

b_transport(packet)

该阻塞 API 用于发送 AXI4‑Stream
API 用法
使用创建的对象调用 API
sender_obj.b_transport(<packet>) 
参数
AXI4‑Stream 包:
packet
AXI4‑Stream 包含下列属性并需要进行设置。
需发送的数据:
data
用于标记包结束的 TLAST 值。该值由用户根据应用进行选择。
tlast
如不指定该值,则 TLAST 默认值为 true,即,每个数据节拍均为一项传输事务。如果 tlast=false,则不驱动任何 TLAST。如果 tlast=true(由用户根据应用来指定),那么最后一个数据节拍 tlast=true
部分其他可选 AXI4‑Stream 包字段包括:tusertkeep
示例
#Create packet
data_packet = xtlm_ipc.axi_stream_packet() // API to generate the stream packet
data = generate_data() # custom user code to generate the data
data_packet.data = data 

data_packet.tlast = True #AXI Stream Fields
#Optional AXI Stream Parameters
data_packet.tuser = optional data
data_packet.tkeep = optional data 

#Send Transaction
data_in.b_transport(data_packet)

sample_transaction()

该阻塞 API 用于接收 AXI4‑Stream 包。
API 用法
使用创建的 receiver_obj 调用 API:
packet = receiver_obj.sample_transaction()
示例
#Sample the packets
data_packet = data_out.sample_transaction() # Will wait until data is received.
此处,data_out 是接收器对象,data_packet 是接收的包。

ipc_axis_master_nb_util (<sender_io_name>) 

例化发送器端口用于在 Python 脚本中执行非阻塞调用。
API 用法
使用 API 为发送器创建用户对象:
sender_obj = ipc_axis_master_nb_util (<sender_io_name>) 
参数
发送数据的外部端口的名称:
<sender_io_name>
示例
#Instantiating AXI Master Utilities 
data_in = ipc_axis_master_nb_util("DataIn")

ipc_axis_slave_nb_util(<receiver_io_name>)

用于例化接收器端口,以便在 Python 脚本中执行非阻塞调用

API 用法
使用 API 为接收器创建用户对象
receiver_obj = ipc_axis_slave_util (<receiver_io_name>)
参数
发送数据的外部端口的名称
receiver_io_name
示例
#Instantiating AXI Slave Utilities
data_out = ipc_axis_slave_nb_util("DataOut")

nb_transport(<packet>)

该非阻塞 API 用于发送 AXI4‑Stream 包。传输数据前,需按以下通过流量生成器以 Python 来格式化数据所述将其格式化为数据包。

API 用法
使用 sender_obj 调用 API:
sender_obj.nb_transport(<packet>) 
参数
AXI4‑Stream 包:
packet
AXI4‑Stream 包是由用户创建的,其中包含以下属性:
需发送的数据:
data
用于标记包结束的 TLAST 值。该值由用户根据应用进行选择。
tlast
如不指定该值,则 TLAST 默认值为 true,即,每个数据节拍均为一项传输事务。如果 tlast=false,则不驱动任何 TLAST。如果 tlast=true(由用户根据应用来指定),那么最后一个数据节拍 tlast=true
部分其他可选 AXI4‑Stream 包字段包括tuser 和。 tkeep
示例
#Create packet
data_packet = xtlm_ipc.axi_stream_packet()
data = generate_data() # custom user code to generate the desired data
data_packet.data_length = "DATA_LENGTH"
data_packet.data = data
data_packet.tlast = True #to mark end of packet. This is user's choice based on application
data_in.nb_transport(data_packet)

nb_sample_transaction()

用于接收 AXI 串流包的非阻塞 API。

API 用法
packet = receiver_obj.nb_sample_transaction() 
示例
#Sample the packets
# Will return data if it is available
# Will return empty list if nothing is available
# Will not wait or stop the execution until data is present to sample
data_packet = data_out.nb_sample_transaction()

此处,data_out 是接收器对象,data_packet 是接收的包。

no_of_available_txns

对于非阻塞调用,此 API 用于获取接收到的传输事务数。

API 用法
使用用户对象为接收器调用 API。
示例
data_out = ipc_axis_slave_nb_util("ReceiverName")
data = []

# Checks if any data is present to receive
if(slave.no_of_available_txns != 0) :
data = data_out.nb_sample_transaction() # fill data with respective received data

end_of_simulation()

结束并退出仿真。

API 用法
使用发送器对象调用 API
sender_obj.end_of_simulation()
示例
data_in.end_of_simulation()

建议搭配 sw_emu/hw_emu 使用。

disconnect()

用于断开套接字的连接(DataIn、DataOut 等)

API 用法
使用发送器或接收器对象来调用 API:
sender_obj.disconect()
receiver_obj.disconnect()
示例
data_in.disconnect() 
data_out.disconnect()

首选搭配 x86simulatoraiesimulator 一起使用。

通过流量生成器以 Python 来格式化数据

适用于 AI 引擎计算图的 Python API 中所述使用 AI 引擎 API 时,无需进行数据格式化。但为了使用高级 API 来发送或接收数据,这些数据需采用数据包形式,因此您需要先进行数据格式化,然后才能传输。

要对 AXI4‑Stream 传输事务进行仿真,AXI Traffic Generator 需将有效载荷数据分割为相应大小的突发。要发送 PLIO 宽度为 32 位(4 字节)的 128 字节数据,就需要 128 字节/4 字节 = 32 项 AXI4‑Stream 传输事务。字节阵列与 AXI 传输事务之间的转换可由 Python 来处理。

Python struct 库可提供相应机制以在 Python 数据类型与 C 语言数据类型之间进行转换。尤其是,struct.packstruct.unpack 函数可根据格式字符串实参对字节阵列进行打包和解包。下表显示了常用 C 语言数据类型和 PLIO 宽度的格式字符串。

如需了解更多信息,请访问:https://docs.python.org/3/library/struct.html

表 2. C 语言数据类型和 PLIO 宽度的格式字符串
数据类型 PLIO 宽度 Python 代码片段
cfloat PLIO32 不适用
PLIO64 rVec = np.real(data)

iVec = np.imag(data)

out2column = np.zeros((L,2)).astype(np.single)

out2column.tobytes()

formatString = "<"+str(len(byte_arry)//4)+"f"

PLIO128
cint16 PLIO32 rVec = np.real(data).astype(np.int16)

iVec = np.imag(data).astype(np.int16)

formatString = "<"+str(len(byte_arry)//2)+"h"

PLIO64
PLIO128
int8 PLIO32 intvec = np.real(data).astype(np.int8)

formatString = "<"+str(len(byte_arry)//1)+"b"

PLIO64
PLIO128
int32 PLIO32 intvec = np.real(data).astype(np.int32)

formatString = "<"+str(len(byte_arry)//4)+"i"

PLIO64
PLIO128