#Flexible模式

Flexible模式下请使用 hbm_infer.hbm_rpc_session_flexible 模块。

#接口及参数

1. 全局方法：init_server

   def init_server(
       host: str,
       username: str = "root",
       password: Optional[str] = None,
       ssh_port: int = 22,
       remote_root: str = "/map/hbm_infer/",
   ) -> HbmRpcServer:

   构造 HbmRpcServer 对象。

   • 参数

   参数	说明
   host	开发板ip地址。
   username	板端用户名，默认值为 "root" 。
   password	登录开发板的密码，默认为 None ，代表无需使用密码。
   ssh_port	SSH目标端口号，默认值为 22 。
   remote_root	板端暂存文件的根目录路径，默认值为 "/map/hbm_infer/" 。

   • 返回值

   HbmRpcServer 对象实例。

2. 全局方法：deinit_server

   def deinit_server(hbm_rpc_server: HbmRpcServer) -> None:

   清理板端server文件。

   • 参数

   参数	说明
   hbm_rpc_server	HbmRpcServer 对象实例。

3. 全局方法：init_hbm

   def init_hbm(
       local_hbm_path: Union[str, List[str]],
       hbm_rpc_server: HbmRpcServer,
   ) -> HbmHandle:

   构造 HbmHandle 对象。

   • 参数

   参数	说明
   local_hbm_path	本地HBM文件路径。
   hbm_rpc_server	HbmRpcServer 对象，可通过调用 init_server 接口获得。

   • 返回值

   HbmHandle 对象实例。

4. 全局方法：deinit_hbm

   def deinit_hbm(hbm_handle: HbmHandle) -> None:

   清理板端HBM文件。

   • 参数

   参数	说明
   hbm_handle	HbmHandle 对象实例。

5. HbmRpcSession成员方法：__init__

   def __init__(
       self,
       hbm_handle: HbmHandle,
       hbm_rpc_server: HbmRpcServer,
       frame_timeout: int = 90,
       server_timeout: int = 5,
       with_profile: bool = False,
       debug: bool = False,
       compress_option: str = "NONE",
       core_id: Union[int, List[int]] = -1,
       remote_environment: Dict[str, Any] = {},
   ) -> None:

   初始化HbmRpcSession对象。

   • 参数

   参数	说明
   hbm_handle	HbmHandle 对象，可通过调用 init_hbm 接口获得。
   hbm_rpc_server	HbmRpcServer 对象，可通过调用 init_server 接口获得。
   frame_timeout	gRPC 通信的单帧超时时间，以秒为单位，默认值为 90 。
   server_timeout	服务端超时时间，以分钟为单位，默认值为 5 ，超时后服务端自动退出并清理 除日志外 的其他文件。
   with_profile	是否开启推理各阶段时间统计，默认值为 False 。
   debug	开启调试模式，保留更多日志信息，默认值为 False 。
   compress_option	开启 gRPC 压缩功能，可选值有 "IN" 、 "INOUT" 及 "NONE" ，分别代表请求数据帧启用压缩、请求和响应数据帧都启用压缩、不启用压缩，默认值为 "NONE" 。
   core_id	指定推理使用的BPU核心ID，0代表CORE_0，1代表CORE_1，...，-1代表CORE_ANY，默认值为-1。可通过列表指定多个核心。
   remote_environment	配置板端环境变量，字典类型，键为环境变量名称，值为环境变量取值，默认值为空字典。

6. HbmRpcSession成员方法：get_model_names

   def get_model_names(self) -> List[str]:

   获取当前Session中模型名称列表。

   • 返回值

   模型名称列表。

7. HbmRpcSession成员方法：get_input_info

   def get_input_info(self, model_name: Optional[str] = None) -> Dict[str, Dict]:

   获取模型输入信息。

   • 参数

   参数	说明
   model_name	模型名称。对于多模型Session，必须指定 model_name 。

   • 返回值

   字典形式的模型输入信息描述，具体格式可参考如下示例：

   {
       "input_name0": {
           "valid_shape": [1, 3, 224, 224],
           "tensor_type": "DATA_TYPE_S8",
           "quanti_type": "QUANTI_TYPE_SCALE",
           "quantizeAxis": 0,
           "scale_data": [0.006861070170998573],
           "zero_point_data": [0]
       },
       ...
   }

8. HbmRpcSession成员方法：get_output_info

   def get_output_info(self, model_name: Optional[str] = None) -> Dict[str, Dict]:

   获取模型输出信息。

   • 参数

   参数	说明
   model_name	模型名称。对于多模型Session，必须指定 model_name 。

   • 返回值

   字典形式的模型输出信息描述，格式与 get_input_info 返回值一致。

9. HbmRpcSession成员方法：show_input_output_info

   def show_input_output_info(self, model_name: Optional[str] = None) -> None:

   打印模型输入输出信息。

   • 参数

   参数	说明
   model_name	模型名称。对于多模型Session，必须指定 model_name 。

10. HbmRpcSession成员方法：__call__

    def __call__(
        self,
        data: Dict[str, Union[np.ndarray, torch.Tensor, HTensor]],
        output_config: Optional[Dict[str, Dict]] = None,
        model_name: Optional[str] = None,
    ) -> Dict[str, Union[np.ndarray, torch.Tensor, HTensor]]:

    执行模型推理。

    • 参数

    参数	说明
    data	模型输入，字典类型。 key 为输入张量名称， value 为输入张量，支持 torch.Tensor 、 numpy.ndarray 及 HTensor 三种格式。 需要注意： 输入数据需要与模型输入信息保持一致（包括名称、数量、形状以及数据元素类型等）。 输入数据不能混合使用 torch.Tensor 和 numpy.ndarray 。 当输入为 torch.Tensor 时，张量 device 需要一致。
    output_config	详细介绍可参考传输优化章节内容。
    model_name	模型名称。对于多模型Session，必须指定 model_name 。

    • 返回值

    模型输出，字典类型。 key 为输出张量名称， value 为输出张量，其类型与模型输入保持一致。

11. HbmRpcSession成员方法：close_server

    def close_server(self) -> None:

    关闭服务端，清理服务端资源。

12. HbmRpcSession成员方法：get_profile

    def get_profile(self, model_name: Optional[str] = None) -> Dict[str, Dict]:

    获取推理各阶段耗时统计信息，需要 with_profile 参数为 True 。

    • 参数

    参数	说明
    model_name	模型名称。对于多模型Session，必须指定 model_name 。

    • 返回值

    推理各阶段耗时统计信息，字典类型，其参考格式如下：

    {
        // 帧总延时（ms）
        "frame_duration": {
            "avg": 6,
            "min": 6,
            "max": 6,
        },
        // gRPC发送输入帧到返回输出帧的总延时（ms）
        "sd2rv_duration": {
            "avg": 5,
            "min": 5,
            "max": 5,
        },
        // 网络通信耗时（ms）
        "commu_duration": {
            "avg": 4,
            "min": 4,
            "max": 4,
        },
        // 板端总耗时（ms）
        "board_duration": {
            "avg": 1,
            "min": 1,
            "max": 1,
        },
        // 板端纯推理耗时（ms）
        "infer_duration": {
            "avg": 0.5,
            "min": 0.5,
            "max": 0.5,
        },
        // 板端前处理耗时（ms）
        "prepr_duration": {
            "avg": 0.3,
            "min": 0.3,
            "max": 0.3,
        },
        // 板端后处理耗时（ms）
        "pospr_duration": {
            "avg": 0.2,
            "min": 0.2,
            "max": 0.2,
        },
    }

13. HbmRpcSession成员方法：get_profile_last_frame

    def get_profile_last_frame(self, model_name: Optional[str] = None) -> Dict[str, Dict]:

    获取最近一帧推理各阶段耗时统计信息，需要 with_profile 参数为 True 。

    • 参数

    参数	说明
    model_name	模型名称。对于多模型Session，必须指定 model_name 。

    • 返回值

    最近一帧推理各阶段耗时统计信息，字典类型，其参考格式如下：

    {
        // 帧总延时（ms）
        "frame_duration": 12,
        // gRPC发送输入帧到返回输出帧的总延时（ms）
        "sd2rv_duration": 10,
        // 网络通信耗时（ms）
        "commu_duration": 6,
        // 板端总耗时（ms）
        "board_duration": 4,
        // 板端纯推理耗时（ms）
        "infer_duration": 2,
        // 板端前处理耗时（ms）
        "prepr_duration": 0.5,
        // 板端后处理耗时（ms）
        "pospr_duration": 0.5,
    }

#使用示例

import torch
import multiprocessing as mp
from hbm_infer.hbm_rpc_session_flexible import HbmRpcSession, init_server, deinit_server, init_hbm, deinit_hbm

def single_session_entry(rpc_server, hbm_handle, run_epoch):
    # 创建session
    sess = HbmRpcSession(
        hbm_rpc_server=rpc_server,
        hbm_handle=hbm_handle
    )
    # 打印模型输入输出信息
    sess.show_input_output_info()
    # 准备输入数据
    input_data = {
        'img': torch.ones((1, 3, 224, 224), dtype=torch.int8)
    }
    # 执行推理并返回结果
    for i in range(run_epoch):
        output_data = sess(input_data)
        print([output_data[k].shape for k in output_data])
    # 关闭server
    sess.close_server()

def run_hbm_infer(num_process=8, run_epoch=20):
    # 创建server
    rpc_server = init_server(
        host=<available_ip>
    )
    # 加载HBM文件
    hbm_handle = init_hbm(
        hbm_rpc_server=rpc_server,
        local_hbm_path=<local_hbm_path>
    )
    # 多进程推理
    processes = list()
    for i in range(num_process):
        p = mp.Process(target=single_session_entry, args=(rpc_server, hbm_handle, run_epoch))
        processes.append(p)
        p.start()
    for p in processes:
        p.join()
    # 清理板端server文件
    deinit_server(rpc_server)
    # 清理板端HBM文件
    deinit_hbm(hbm_handle)

if __name__ == "__main__":
    run_hbm_infer()