跳转到主要内容
SDK Version: 2.4.0

Deprecated API Migration Guide

The following InferenceEngine methods have been deprecated and will be removed in a future release. Please update your code to use the recommended alternatives.

Naming Convention Changes (PascalCase → snake_case)​

Deprecated MethodReplacementNotes
Run(input_feed_list, user_arg)run(input_data, user_args=user_arg)Parameter renamed to user_args (plural). Note: run_async, run_multi_input, run_async_multi_input retain the singular user_arg form
RunBatch(input_buffers, output_buffers, user_args)run(input_buffers, output_buffers=output_buffers, user_args=user_args)Use run() with batch-formatted inputs
RunAsync(input_feed_list, user_arg)run_async(input_data, user_arg=user_arg)Parameter name kept as singular user_arg (only run() uses plural user_args)
RunBenchMark(loop_cnt, input_feed_list)run_benchmark(num_loops, input_data)Parameter names changed
ValidateDevice(input_feed_list, device_id)validate_device(input_data, device_id)
RegisterCallBack(callback)register_callback(callback)
Wait(req_id)wait(job_id)Parameter name changed: req_id → job_id
is_PPU()is_ppu()

Method Renaming (Consistency Improvements)​

Deprecated MethodReplacementNotes
input_size()get_input_size()
output_size()get_output_size()
task_order()get_task_order()
latency()get_latency()
inference_time()get_npu_inference_time()
get_outputs()get_all_task_outputs()
bitmatch_mask(index)get_bitmatch_mask(index)
get_num_tails()get_num_tail_tasks()
run_batch(input_buffers, output_buffers, user_args)run(input_buffers, output_buffers=output_buffers, user_args=user_args)Consolidated into run()

Data Type Query Methods​

Deprecated MethodReplacementNotes
input_dtype()get_input_tensors_info()Access 'dtype' key from returned dict list
output_dtype()get_output_tensors_info()Access 'dtype' key from returned dict list
get_input_data_type()get_input_tensors_info()Access 'dtype' key from returned dict list
get_output_data_type()get_output_tensors_info()Access 'dtype' key from returned dict list

Migration Examples​

Before (deprecated):

engine = InferenceEngine("model.dxnn")

# Inference
outputs = engine.Run([input_buffer])

# Batch inference
batch_outputs = engine.RunBatch(input_buffers, output_buffers)

# Async inference
job_id = engine.RunAsync([input_buffer])
result = engine.Wait(job_id)

# Query info
size = engine.input_size()
dtypes = engine.input_dtype()
time = engine.inference_time()

After (recommended):

engine = InferenceEngine("model.dxnn")

# Inference
outputs = engine.run(input_buffer)

# Batch inference
batch_outputs = engine.run(input_buffers, output_buffers=output_buffers)

# Async inference
job_id = engine.run_async(input_buffer)
result = engine.wait(job_id)

# Query info
size = engine.get_input_size()
dtypes = [info['dtype'] for info in engine.get_input_tensors_info()]
time = engine.get_npu_inference_time()

This section describes the Python bindings of the DX-RT SDK. It provides a streamlined interface to the same core functionalities as the C++ SDK, making it ideal for rapid development, prototyping, and integration with Python-based AI workflows.

class dx_engine.InferenceEngine​

This class is the main Python wrapper for the DXRT Inference Engine. It provides an interface to load a compiled model and perform inference tasks, either synchronously or asynchronously, supporting both single and batch inference.

Constructor​

__init__(self, model_path: str, inference_option: Optional[InferenceOption] = None)

  • Description: Initializes the InferenceEngine by loading a compiled model from the specified file path.
  • Parameters:
    • model_path: str. Path to the compiled model file (e.g., *.dxnn).
    • inference_option: Optional[InferenceOption]. An InferenceOption object for configuration. If None, default options are used.
  • Raises: RuntimeError if the underlying C++ engine fails to initialize.
  • Example: 
    from dx_engine import InferenceEngine, InferenceOption

    # Load model from file path
    engine = InferenceEngine("model.dxnn")

    # Load model with custom options
    option = InferenceOption()
    option.devices = [0, 1]
    option.bound_option = InferenceOption.BOUND_OPTION.NPU_ALL
    engine2 = InferenceEngine("model.dxnn", option)

from_buffer(cls, memory_buffer: np.ndarray, inference_option: Optional[InferenceOption] = None) -> InferenceEngine

  • Description: Alternative class method constructor that creates an InferenceEngine from a memory buffer without requiring a file path. This is useful for embedded systems, network-based model distribution, or when models are loaded from custom sources (e.g., encrypted storage).
  • Parameters:
    • memory_buffer: np.ndarray. Pre-allocated memory buffer containing the compiled model data. Must be a C-contiguous numpy array (dtype=uint8).
    • inference_option: Optional[InferenceOption]. Configuration options for inference. If None, default options are used.
  • Returns: A new InferenceEngine instance loaded from the memory buffer.
  • Raises:
    • TypeError if memory_buffer is not a numpy array.
    • ValueError if memory_buffer is not C-contiguous.
    • RuntimeError if the C++ engine fails to initialize from the buffer.
  • Example: 
    import numpy as np
    from dx_engine import InferenceEngine, InferenceOption

    # Load model file into memory buffer
    with open('model.dxnn', 'rb') as f:
        buffer = np.frombuffer(f.read(), dtype=np.uint8)

    # Create engine from memory buffer
    engine = InferenceEngine.from_buffer(buffer)

    # With custom options
    option = InferenceOption()
    option.use_ort = True
    engine2 = InferenceEngine.from_buffer(buffer, option)
  • Use Cases:
    • Loading models from encrypted or compressed storage
    • Network-based model distribution
    • Embedded systems with models stored in ROM
    • Dynamic model loading without filesystem access

Member Functions​

dispose(self)

  • Signature: def dispose(self) -> None
  • Description: Explicitly releases the underlying C++ resources held by the inference engine. This is automatically called when using a with statement, so manual invocation is typically not required.

get_all_task_outputs(self)

  • Signature: def get_all_task_outputs(self) -> List[List[np.ndarray]]
  • Description: Retrieves the outputs of all internal tasks in their execution order. This is useful for debugging the intermediate steps of a model.
  • Returns: A list of lists, where each inner list contains the output np.ndarray objects for a single task.

get_bitmatch_mask(self, index: int = 0)

  • Signature: def get_bitmatch_mask(self, index: int = 0) -> np.ndarray
  • Description: Retrieves a bitmatch mask for a specific NPU task, which can be used for validation and debugging purposes.
  • Parameters:
    • index: int. The index of the NPU task.
  • Returns: A boolean np.ndarray representing the bitmatch mask.

get_compile_type(self)

  • Signature: def get_compile_type(self) -> str
  • Description: Returns the compilation type or strategy of the loaded model (e.g., "debug", "release").
  • Returns: The compilation type as a string.

get_input_size(self)

  • Signature: def get_input_size(self) -> int
  • Description: Gets the total expected size of all input tensors combined in bytes.
  • Returns: The total input size as an integer.

get_input_tensor_count(self)

  • Signature: def get_input_tensor_count(self) -> int
  • Description: Returns the number of input tensors required by the model.
  • Returns: The number of input tensors.

get_input_tensor_names(self)

  • Signature: def get_input_tensor_names(self) -> List[str]
  • Description: Returns the names of all input tensors in the order they should be provided.
  • Returns: A list of input tensor names.

get_input_tensor_sizes(self)

  • Signature: def get_input_tensor_sizes(self) -> List[int]
  • Description: Gets the individual sizes of each input tensor in bytes, in their correct order.
  • Returns: A list of integer sizes.

get_input_tensor_to_task_mapping(self)

  • Signature: def get_input_tensor_to_task_mapping(self) -> Dict[str, str]
  • Description: Returns the mapping from input tensor names to their target tasks within the model graph.
  • Returns: A dictionary mapping tensor names to task names.

get_input_tensors_info(self)

  • Signature: def get_input_tensors_info(self) -> List[Dict[str, Any]]
  • Description: Returns detailed information for each input tensor.
  • Returns: A list of dictionaries, where each dictionary contains keys: 'name' (str), 'shape' (List[int]), 'dtype' (np.dtype), and 'elem_size' (int).

get_latency(self)

  • Signature: def get_latency(self) -> int
  • Description: Returns the latency of the most recent inference in microseconds.
  • Returns: The latency value as an integer.

get_latency_count(self)

  • Signature: def get_latency_count(self) -> int
  • Description: Returns the total count of latency values collected since initialization.
  • Returns: The number of measurements.

get_latency_list(self)

  • Signature: def get_latency_list(self) -> List[int]
  • Description: Returns a list of recent latency measurements in microseconds.
  • Returns: A list of integers.

get_latency_mean(self)

  • Signature: def get_latency_mean(self) -> float
  • Description: Returns the mean (average) of all collected latency values.
  • Returns: The mean latency as a float.

get_latency_std(self)

  • Signature: def get_latency_std(self) -> float
  • Description: Returns the standard deviation of all collected latency values.
  • Returns: The standard deviation as a float.

get_model_version(self)

  • Signature: def get_model_version(self) -> str
  • Description: Returns the DXNN file format version of the loaded model.
  • Returns: The model version string.

get_npu_inference_time(self)

  • Signature: def get_npu_inference_time(self) -> int
  • Description: Returns the pure NPU processing time for the most recent inference in microseconds.
  • Returns: The NPU inference time as an integer.

get_npu_inference_time_count(self)

  • Signature: def get_npu_inference_time_count(self) -> int
  • Description: Returns the total count of NPU inference time values collected.
  • Returns: The number of measurements.

get_npu_inference_time_list(self)

  • Signature: def get_npu_inference_time_list(self) -> List[int]
  • Description: Returns a list of recent NPU inference time measurements in microseconds.
  • Returns: A list of integers.

get_npu_inference_time_mean(self)

  • Signature: def get_npu_inference_time_mean(self) -> float
  • Description: Returns the mean (average) of all collected NPU inference times.
  • Returns: The mean time as a float.

get_npu_inference_time_std(self)

  • Signature: def get_npu_inference_time_std(self) -> float
  • Description: Returns the standard deviation of all collected NPU inference times.
  • Returns: The standard deviation as a float.

get_num_tail_tasks(self)

  • Signature: def get_num_tail_tasks(self) -> int
  • Description: Returns the number of 'tail' tasks (tasks with no successors) in the model graph.
  • Returns: The number of tail tasks.

get_output_size(self)

  • Signature: def get_output_size(self) -> int
  • Description: Gets the total size of all output tensors combined in bytes.
  • Returns: The total output size as an integer.

get_output_tensor_count(self)

  • Signature: def get_output_tensor_count(self) -> int
  • Description: Returns the number of output tensors produced by the model.
  • Returns: The number of output tensors.

get_output_tensor_names(self)

  • Signature: def get_output_tensor_names(self) -> List[str]
  • Description: Returns the names of all output tensors in the order they are produced.
  • Returns: A list of output tensor names.

get_output_tensor_sizes(self)

  • Signature: def get_output_tensor_sizes(self) -> List[int]
  • Description: Gets the individual sizes of each output tensor in bytes, in their correct order.
  • Returns: A list of integer sizes.

get_output_tensors_info(self)

  • Signature: def get_output_tensors_info(self) -> List[Dict[str, Any]]
  • Description: Returns detailed information for each output tensor.
  • Returns: A list of dictionaries with keys: 'name' (str), 'shape' (List[int]), 'dtype' (np.dtype), and 'elem_size' (int).

get_task_order(self)

  • Signature: def get_task_order(self) -> np.ndarray
  • Description: Returns the execution order of tasks/subgraphs within the model.
  • Returns: A numpy array of strings representing the task order.

is_multi_input_model(self)

  • Signature: def is_multi_input_model(self) -> bool
  • Description: Checks if the loaded model requires multiple input tensors.
  • Returns: True if the model has multiple inputs, False otherwise.

is_ppu(self)

  • Signature: def is_ppu(self) -> bool
  • Description: Checks if the loaded model utilizes a Post-Processing Unit (PPU).
  • Returns: True if the model uses a PPU, False otherwise.

register_callback(self, callback: Optional[Callable[[List[np.ndarray], Any], int]])

  • Signature: def register_callback(self, callback: Optional[Callable[[List[np.ndarray], Any], int]]) -> None
  • Description: Registers a user-defined callback function to be executed upon completion of an asynchronous inference.
  • Parameters:
    • callback: A callable function or None to unregister. The callback receives the list of output arrays and the user argument.

run(self, input_data: Union[np.ndarray, List[np.ndarray], List[List[np.ndarray]]], output_buffers: Optional[Union[List[np.ndarray], List[List[np.ndarray]]]] = None, user_args: Optional[Union[Any, List[Any]]] = None)

  • Signature: def run(self, input_data, output_buffers=None, user_args=None) -> Union[List[np.ndarray], List[List[np.ndarray]]]
  • Description: Runs inference synchronously. This versatile method handles single-item, multi-input, and batch inference based on the format of input_data.
  • Parameters:
    • input_data: Input data in various formats (np.ndarray, List[np.ndarray], List[List[np.ndarray]]).
    • output_buffers: Optional pre-allocated buffers for the output.
    • user_args: Optional user-defined argument or list of arguments for batch mode.
  • Returns: The inference result(s). A List[np.ndarray] for single inference or a List[List[np.ndarray]] for batch inference.

run_async(self, input_data: Union[np.ndarray, List[np.ndarray]], user_arg: Any = None, output_buffer: Optional[Union[np.ndarray, List[np.ndarray]]] = None)

  • Signature: def run_async(self, input_data, user_arg=None, output_buffer=None) -> int
  • Description: Runs inference asynchronously for a single item. Batch processing is not supported with this method.
  • Parameters:
    • input_data: A single np.ndarray or a List[np.ndarray] for multi-input models.
    • user_arg: An optional user-defined argument to be passed to the callback.
    • output_buffer: An optional pre-allocated buffer for the output.
  • Returns: An integer job_id for this asynchronous operation.

run_async_multi_input(self, input_tensors: Dict[str, np.ndarray], user_arg: Any = None, output_buffer: Optional[List[np.ndarray]] = None)

  • Signature: def run_async_multi_input(self, input_tensors, user_arg=None, output_buffer=None) -> int
  • Description: A convenience method to run asynchronous inference on a multi-input model using a dictionary of named tensors.
  • Parameters:
    • input_tensors: A dictionary mapping input tensor names to np.ndarray data.
    • user_arg: An optional user-defined argument.
    • output_buffer: An optional list of pre-allocated output arrays.
  • Returns: An integer job_id.

run_benchmark(self, num_loops: int, input_data: Optional[List[np.ndarray]] = None)

  • Signature: def run_benchmark(self, num_loops: int, input_data: Optional[List[np.ndarray]] = None) -> float
  • Description: Runs a performance benchmark for a specified number of loops.
  • Parameters:
    • num_loops: The number of inference iterations to run.
    • input_data: An optional list of np.ndarray to use as input for the benchmark.
  • Returns: The average frames per second (FPS) as a float.

run_multi_input(self, input_tensors: Dict[str, np.ndarray], output_buffers: Optional[List[np.ndarray]] = None, user_arg: Any = None)

  • Signature: def run_multi_input(self, input_tensors, output_buffers=None, user_arg=None) -> List[np.ndarray]
  • Description: A convenience method to run synchronous inference on a multi-input model using a dictionary of named tensors.
  • Parameters:
    • input_tensors: A dictionary mapping input tensor names to np.ndarray data.
    • output_buffers: An optional list of pre-allocated output arrays.
    • user_arg: An optional user-defined argument.
  • Returns: A list of np.ndarray objects containing the output.

wait(self, job_id: int)

  • Signature: def wait(self, job_id: int) -> List[np.ndarray]
  • Description: Waits for an asynchronous job (identified by job_id) to complete and retrieves its output.
  • Parameters:
    • job_id: The integer job ID returned from a run_async call.
  • Returns: A list of np.ndarray objects containing the output from the completed job.

class dx_engine.InferenceOption​

This class provides a Pythonic interface to configure inference options such as device selection and core binding. It wraps the C++ InferenceOption struct.

Constructor​

__init__(self)

  • Signature: def __init__(self) -> None
  • Description: Initializes a new InferenceOption object with default values from the C++ backend.

Properties​

bound_option

  • Description: Gets or sets the NPU core binding strategy.
  • Type: InferenceOption.BOUND_OPTION (Enum).

devices

  • Description: Gets or sets the list of device IDs to be used for inference. An empty list means all available devices will be used.
  • Type: List[int].

use_ort

  • Description: Gets or sets whether to use the ONNX Runtime for executing CPU-based tasks in the model graph.
  • Type: bool.

buffer_count

  • Description: Gets or sets the number of internal buffers allocated for inference. Higher values can improve throughput in pipelined inference scenarios by allowing more concurrent operations, but consume more memory. Default is 6 (delegate to the library default). When set explicitly, valid range is 1-100.
  • Type: int.
  • Example: 
    from dx_engine import InferenceEngine, InferenceOption

    option = InferenceOption()
    option.buffer_count = 8  # Allocate 8 buffers for higher throughput
    engine = InferenceEngine("model.dxnn", option)

Member Functions​

get_bound_option(self)

  • Signature: def get_bound_option(self) -> BOUND_OPTION
  • Description: Returns the current NPU core binding option.
  • Returns: An InferenceOption.BOUND_OPTION enum member.

get_devices(self)

  • Signature: def get_devices(self) -> List[int]
  • Description: Returns the list of device IDs targeted for inference.
  • Returns: A list of integers.

get_use_ort(self)

  • Signature: def get_use_ort(self) -> bool
  • Description: Returns whether ONNX Runtime usage is enabled.
  • Returns: A boolean value.

get_buffer_count(self)

  • Signature: def get_buffer_count(self) -> int
  • Description: Returns the current buffer count configuration.
  • Returns: An integer representing the number of buffers.

set_bound_option(self, boundOption: BOUND_OPTION)

  • Signature: def set_bound_option(self, boundOption: BOUND_OPTION)
  • Description: Sets the NPU core binding option.
  • Parameters:
    • boundOption: An InferenceOption.BOUND_OPTION enum member.

set_devices(self, devices: List[int])

  • Signature: def set_devices(self, devices: List[int])
  • Description: Sets the list of device IDs to be used for inference.
  • Parameters:
    • devices: A list of integers representing device IDs.

set_use_ort(self, use_ort: bool)

  • Signature: def set_use_ort(self, use_ort: bool)
  • Description: Enables or disables the use of ONNX Runtime for CPU tasks.
  • Parameters:
    • use_ort: A boolean value.

set_buffer_count(self, buffer_count: int)

  • Signature: def set_buffer_count(self, buffer_count: int)
  • Description: Sets the number of internal buffers for inference.
  • Parameters:
    • buffer_count: An integer value specifying the number of buffers (valid range: 1-100).

Nested Classes​

class BOUND_OPTION(Enum)

  • Description: An enumeration defining how NPU cores are utilized.
  • Members: NPU_ALL, NPU_0, NPU_1, NPU_2, NPU_01, NPU_12, NPU_02.

class dx_engine.Configuration​

Provides access to the global DXRT configuration singleton, allowing for system-wide settings like enabling the profiler.

Constructor​

__init__(self)

  • Signature: def __init__(self)
  • Description: Initializes the Configuration object by getting a reference to the underlying C++ singleton instance.

Member Functions​

get_attribute(self, item: ITEM, attrib: ATTRIBUTE)

  • Signature: def get_attribute(self, item: ITEM, attrib: ATTRIBUTE) -> str
  • Description: Retrieves the value of a specific attribute for a configuration item.
  • Parameters:
    • item: The configuration category (e.g., Configuration.ITEM.PROFILER).
    • attrib: The attribute to retrieve (e.g., Configuration.ATTRIBUTE.PROFILER_SHOW_DATA).
  • Returns: The attribute value as a string.

get_driver_version(self)

  • Signature: def get_driver_version(self) -> str
  • Description: Returns the version of the installed device driver.
  • Returns: The driver version string.

get_enable(self, item: ITEM)

  • Signature: def get_enable(self, item: ITEM) -> bool
  • Description: Checks if a specific configuration item is enabled.
  • Parameters:
    • item: The configuration category to check.
  • Returns: True if enabled, False otherwise.

get_pcie_driver_version(self)

  • Signature: def get_pcie_driver_version(self) -> str
  • Description: Returns the version of the installed PCIe driver.
  • Returns: The PCIe driver version string.

get_version(self)

  • Signature: def get_version(self) -> str
  • Description: Returns the version of the DXRT library.
  • Returns: The library version string.

load_config_file(self, file_name: str)

  • Signature: def load_config_file(self, file_name: str)
  • Description: Loads configuration settings from a specified file.
  • Parameters:
    • file_name: The path to the configuration file.

set_attribute(self, item: ITEM, attrib: ATTRIBUTE, value: str)

  • Signature: def set_attribute(self, item: ITEM, attrib: ATTRIBUTE, value: str)
  • Description: Sets a string value for a specific attribute of a configuration item (e.g., setting PROFILER_SAVE_DATA to "ON").
  • Parameters:
    • item: The configuration category.
    • attrib: The attribute to set.
    • value: The string value to assign.

set_enable(self, item: ITEM, enabled: bool)

  • Signature: def set_enable(self, item: ITEM, enabled: bool)
  • Description: Enables or disables a global configuration item, such as PROFILER.
  • Parameters:
    • item: The configuration category.
    • enabled: A boolean value to enable (True) or disable (False) the item.

Nested Classes​

class ITEM

  • Description: An enumeration-like class defining configuration categories.
  • Members: DEBUG, PROFILER, SERVICE, DYNAMIC_CPU_THREAD, TASK_FLOW, SHOW_THROTTLING, SHOW_PROFILE, SHOW_MODEL_INFO, CUSTOM_INTRA_OP_THREADS, CUSTOM_INTER_OP_THREADS.

class ATTRIBUTE

  • Description: An enumeration-like class defining attributes for configuration items.
  • Members: PROFILER_SHOW_DATA, PROFILER_SAVE_DATA, CUSTOM_INTRA_OP_THREADS_NUM, CUSTOM_INTER_OP_THREADS_NUM.

class dx_engine.DeviceStatus​

Provides an interface to query real-time status and static information about hardware devices.

Class Methods​

get_current_status(cls, deviceId: int)

  • Signature: def get_current_status(cls, deviceId: int) -> object
  • Description: Retrieves the status for the device with the specified ID. The status includes key metrics such as core utilization and temperature. Monitoring data is refreshed every 1 second.
  • Parameters:
    • deviceId: The integer ID of the device to query.
  • Returns: An instance of DeviceStatus.
  • Raises: RuntimeError if the specified device ID is invalid or does not exist.

get_device_count(cls)

  • Signature: def get_device_count(cls) -> int
  • Description: Returns the total number of hardware devices detected by the system. This includes devices that are initialized and ready for operation.
  • Returns: The number of devices as an integer.

Instance Methods​

get_id(self)

  • Signature: def get_id(self) -> int
  • Description: Returns the unique ID of the device associated with this DeviceStatus instance.
  • Returns: The device ID as an integer.

get_npu_clock(self, ch: int)

  • Signature: def get_npu_clock(self, ch: int) -> int
  • Description: Returns the current clock frequency of a specific NPU core. The clock frequency may change dynamically depending on performance scaling settings.
  • Parameters:
    • ch: The integer index of the NPU core.
  • Returns: The clock speed in MHz.

get_npu_voltage(self, ch: int)

  • Signature: def get_npu_voltage(self, ch: int) -> int
  • Description: Returns the current voltage of a specific NPU core. The voltage level can vary depending on power management settings and workload.
  • Parameters:
    • ch: The integer index of the NPU core.
  • Returns: The voltage in millivolts (mV).

get_temperature(self, ch: int)

  • Signature: def get_temperature(self, ch: int) -> int
  • Description: Returns the current temperature of a specific NPU core. Monitoring the temperature is crucial for ensuring the NPU operates within safe thermal limits.
  • Parameters:
    • ch: The integer index of the NPU core.
  • Returns: The temperature in degrees Celsius. Returns INT16_MIN when the channel index is out of range.

get_core_utilization(self, core_id: int)

  • Signature: def get_core_utilization(self, core_id: int) -> float
  • Description: Retrieves the utilization of the specified NPU core.
  • Parameters:
    • core_id: The NPU core index (0 to 2).
  • Returns: Utilization percentage (0.0 ~ 100.0). Returns -1.0 if core_id is out of range.

get_memory_used(self)

  • Signature: def get_memory_used(self) -> int
  • Description: Retrieves the amount of NPU DRAM currently in use.
  • Returns: DRAM usage in bytes.

get_memory_free(self)

  • Signature: def get_memory_free(self) -> int
  • Description: Retrieves the amount of free NPU DRAM.
  • Returns: Free DRAM in bytes.

class dx_engine.RuntimeEventDispatcher​

A singleton class that provides a centralized event dispatching mechanism for runtime events such as device errors, warnings, and notifications. It supports custom event handlers and automatic logging of events with different severity levels.

Constructor​

__init__(self)

  • Description: Obtains the singleton instance of the underlying C++ RuntimeEventDispatcher. Instantiating the Python class is the standard entry point.
  • Example: 
    from dx_engine import RuntimeEventDispatcher
    dispatcher = RuntimeEventDispatcher()

Nested Classes​

class LEVEL(IntEnum)

  • Description: Event severity levels for categorizing runtime events.
  • Members:
    • INFO (1): Informational messages for normal operation events.
    • WARNING (2): Warning messages for potential issues that don't stop execution.
    • ERROR (3): Error messages for recoverable failures.
    • CRITICAL (4): Critical errors that may cause system instability.

class TYPE(IntEnum)

  • Description: Event type categories for classifying the source of events.
  • Members:
    • DEVICE_CORE (1000): Events related to NPU core operations.
    • DEVICE_STATUS (1001): Device status change events.
    • DEVICE_IO (1002): Input/Output operation events.
    • DEVICE_MEMORY (1003): Memory management events.
    • UNKNOWN (1004): Unknown or unclassified event types.

class CODE(IntEnum)

  • Description: Specific event codes for identifying the exact nature of events.
  • Members:
    • WRITE_INPUT (2000): Input data write operation event.
    • READ_OUTPUT (2001): Output data read operation event.
    • MEMORY_OVERFLOW (2002): Memory overflow or capacity exceeded.
    • MEMORY_ALLOCATION (2003): Memory allocation failure or issue.
    • DEVICE_EVENT (2004): General device event notification.
    • RECOVERY_OCCURRED (2005): Device recovery action taken.
    • TIMEOUT_OCCURRED (2006): Operation timeout event.
    • THROTTLING_NOTICE (2007): Device throttling notification.
    • THROTTLING_EMERGENCY (2008): Device throttling emergency notification.
    • UNKNOWN (2009): Unknown or unclassified event code.

Instance Methods​

dispatch_event(self, level: LEVEL, type: TYPE, code: CODE, event_message: str)

  • Signature: def dispatch_event(self, level: LEVEL, type: TYPE, code: CODE, event_message: str) -> None
  • Description: Dispatches a runtime event with the specified parameters. The event is logged and any registered custom event handler is invoked. Events are filtered based on the current level threshold set via set_current_level.
  • Parameters:
    • level: Severity level of the event (INFO, WARNING, ERROR, CRITICAL).
    • type: Category of the event (DEVICE_CORE, DEVICE_IO, etc.).
    • code: Specific event code identifying the exact event.
    • event_message: Descriptive string providing event details.
  • Raises: ValueError if event_message is not a string.

register_event_handler(self, handler: Callable[[int, int, int, str, str], None])

  • Signature: def register_event_handler(self, handler: Callable[[int, int, int, str, str], None]) -> None
  • Description: Registers a user-defined callback function that will be invoked for each dispatched event. Only one handler can be registered at a time; subsequent calls will replace the previous handler. The handler is invoked synchronously but with minimal lock holding time to avoid blocking.
  • Parameters:
    • handler: A callable with signature handler(level, type, code, message, timestamp) -> None.
  • Raises: ValueError if handler is not callable.
  • Example: 
    from dx_engine import RuntimeEventDispatcher

    def my_handler(level, type, code, message, timestamp):
        print(f"[{timestamp}] Level {level}: {message}")

    dispatcher = RuntimeEventDispatcher()
    dispatcher.register_event_handler(my_handler)

set_current_level(self, level: LEVEL)

  • Signature: def set_current_level(self, level: LEVEL) -> None
  • Description: Sets the minimum event level threshold. Events below this level may be filtered out by custom handlers.
  • Parameters:
    • level: Minimum severity level for events to be processed.

get_current_level(self)

  • Signature: def get_current_level(self) -> LEVEL
  • Description: Gets the current minimum event level threshold.
  • Returns: The current minimum event severity level as a LEVEL enum value.

Standalone Functions​

dx_engine.inference_engine.parse_model(model_path: str)

  • Signature: def parse_model(model_path: str) -> str
  • Description: Module-level function (not a class method) that parses a .dxnn model file using the C++ backend and returns a string containing information about the model's structure and properties. It is defined at the top level of dx_engine.inference_engine and is not re-exported from the dx_engine package root, so it must be imported from its fully qualified module path.
  • Parameters:
    • model_path: The path to the compiled model file.
  • Returns: A string with model information.
  • Raises: TypeError if model_path is not a str.
  • Import: 
    from dx_engine.inference_engine import parse_model
  • Example: 
    from dx_engine.inference_engine import parse_model

    info = parse_model("model.dxnn")
    print(info)
NOTE

A separate function dx_engine.utils.parse_model(model_path, options=None) -> int also exists with a different signature and an int return code (0 on success, -1 on failure). The function documented here (returning str) is the recommended one for retrieving model information.

Last updated: