#Flexible Mode

Please use the hbm_infer.hbm_rpc_session_flexible module for Flexible Mode.

#API Parameter Descriptions

1. Global Method: init_server

   def init_server(
       host:         str,
       username:     str = "root",
       password:     str = "",
       remote_root:  str = "/map/hbm_infer/",
   ) -> HbmRpcServer:

   Constructs an HbmRpcServer object.

   • Parameters

   PARAMETER	DESCRIPTIONS
   host	IP address of the development board.
   username	Board-side username.
   remote_root	Root directory for temporary files on the board.
   password	Login password for the development board.

   • Returns

   An instance of the HbmRpcServer object.

2. Global Method: deinit_server

   def deinit_server(hbm_rpc_server: HbmRpcServer) -> None:

   Clean up the server files on the board.

   • Parameters

   PARAMETER	DESCRIPTIONS
   hbm_rpc_server	Instance of the HbmRpcServer object.

3. Global Method: init_hbm

   def init_hbm(
       local_hbm_path: str,
       hbm_rpc_server: HbmRpcServer,
   ) -> HbmHandle:

   Constructs an HbmHandle object.

   • Parameters

   PARAMETER	DESCRIPTIONS
   local_hbm_path	Local path to the HBM file.
   hbm_rpc_server	Instance of the HbmRpcServer object.

   • Returns

   An instance of the HbmHandle object.

4. Global Method: deinit_hbm

   def deinit_hbm(hbm_handle: HbmHandle) -> None:

   Clean up the board-side HBM files.

   • Parameters

   PARAMETER	DESCRIPTIONS
   hbm_handle	Instance of the HbmHandle object.

5. HbmRpcSession Member Method: __init__

   def __init__(
       self,
       hbm_handle:       HbmHandle,
       hbm_rpc_server:   HbmRpcServer,
       frame_timeout:    Optional[int]   = 90,
       server_timeout:   Optional[int]   = 30,
       with_profile:     Optional[bool]  = False,
       debug:            Optional[bool]  = False,
       compress_option:  Optional[str]   = "NONE",
   ) -> None:

   Initializes an HbmRpcSession object.

   • Parameters

   PARAMETER	DESCRIPTIONS
   hbm_handle	Instance of the HbmHandle object.
   hbm_rpc_server	Instance of the HbmRpcServer object.
   frame_timeout	Per-frame timeout for gRPC communication in seconds.
   server_timeout	Server timeout in minutes. Server auto-terminates and cleans non-log files after timeout.
   with_profile	Whether to enable time statistics for each stage of inference. The default value is False .
   debug	Enable debug mode retains more logs.
   compress_option	Enable the gRPC compression feature. Optional values are "IN" , "INOUT" , and "NONE" , which indicate enabling compression for request data frames , enabling compression for both request and response data frames , and disabling compression , respectively.

6. HbmRpcSession Member Method: get_input_info

   def get_input_info(self) -> dict:

   Get model input information.

   • Returns

   A dictionary describing the model input information. For specific format details, refer to the example below:

   {
       "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]
       },
       ...
   }

7. HbmRpcSession Member Method: get_output_info

   def get_output_info(self) -> dict:

   Get model output information.

   • Returns

   Returns a dictionary describing the model output information , with a format consistent with the return value of get_input_info .

8. HbmRpcSession Member Method: show_input_output_info

   def show_input_output_info(self) -> None:

   Print model input and output information.

9. HbmRpcSession Member Method: __call__

   def __call__(self, data) -> (dict | list):

   Perform model inference.

   • Parameters

   PARAMETER	DESCRIPTIONS
   data	Model input data (dictionary type). The value supports two formats: torch.Tensor and numpy.ndarray . Note: The input data must align with the model's input specifications (including name, quantity, shape, and data element type) .

   • Returns

   Model output, of dictionary type. The key is the name of the output tensor, and the value is the output tensor, which has the same type as the model input.

10. HbmRpcSession Member Method: close_server

    def close_server(self) -> None:

    Shut down the server and clean up server-side resources.

11. HbmRpcSession Member Method: get_profile

    def get_profile(self) -> dict:

    To obtain the time statistics for each stage of inference, the with_profile parameter must be set to True , otherwise an error will be raised. The return value is of dictionary type, and its reference format is as follows:

    {
        // Total frame latency (ms)
        "frame_duration": {
            "avg": 6,
            "min": 6,
            "max": 6,
        },
        // Latency from sending a gRPC request to receiving a response (ms)
        "sd2rv_duration": {
            "avg": 5,
            "min": 5,
            "max": 5,
        },
        // Network communication latency (ms)
        "commu_duration": {
            "avg": 4,
            "min": 4,
            "max": 4,
        },
        // Board-side total latency (ms)
        "board_duration": {
            "avg": 1,
            "min": 1,
            "max": 1,
        },
        // Board-side infer latency (ms)
        "infer_duration": {
            "avg": 0.5,
            "min": 0.5,
            "max": 0.5,
        },
        // Board-side preprocess latency (ms)
        "prepr_duration": {
            "avg": 0.3,
            "min": 0.3,
            "max": 0.3,
        },
        // Board-side postprocess latency (ms)
        "pospr_duration": {
            "avg": 0.2,
            "min": 0.2,
            "max": 0.2,
        },
    }

12. HbmRpcSession Member Method: get_profile_last_frame

    def get_profile_last_frame(self) -> dict:

    To obtain the time statistics for each stage of the most recent frame inference, the with_profile parameter must be set to True , otherwise an error will be raised. The return value is of dictionary type, and its reference format is as follows:

    {
        // Total frame latency (ms)
        "frame_duration": 12,
        // Latency from sending a gRPC request to receiving a response (ms)
        "sd2rv_duration": 10,
        // Network communication latency (ms)
        "commu_duration": 6,
        // Board-side total latency (ms)
        "board_duration": 4,
        // Board-side infer latency (ms)
        "infer_duration": 2,
        // Board-side preprocess latency (ms)
        "prepr_duration": 0.5,
        // Board-side postprocess latency (ms)
        "pospr_duration": 0.5,
    }

#Usage Example

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):
    # Create session
    sess = HbmRpcSession(
        hbm_rpc_server=rpc_server,
        hbm_handle=hbm_handle
    )
    # Print model input/output information
    sess.show_input_output_info()
    # Prepare input data
    input_data = {
        'img': torch.ones((1, 3, 224, 224), dtype=torch.int8)
    }
    # Execute inference and return results
    for i in range(run_epoch):
        output_data = sess(input_data)
        print([output_data[k].shape for k in output_data])
    # Close server
    sess.close_server()

def run_hbm_infer(num_process=8, run_epoch=20):
    # Initialize server
    rpc_server = init_server(
        host=<available_ip>
    )
    # Load HBM file
    hbm_handle = init_hbm(
        hbm_rpc_server=rpc_server,
        local_hbm_path=<local_hbm_path>
    )
    # Multi-process inference
    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()
    # Clean up board-side server files
    deinit_server(rpc_server)
    # Clean up board-side HBM files
    deinit_hbm(hbm_handle)

if __name__ == "__main__":
    run_hbm_infer()