package.infer.session.session Class Reference

Collaboration diagram for package.infer.session.session:

Public Member Functions
None	__init__ (self, str path, str backends="cpu")
Union[List[np.ndarray], Tuple[List[np.ndarray], Dict[str, float]]]	infer (self, List[np.ndarray] inputs_array, *bool measure=False)
Public Member Functions inherited from package.common.basesession.BaseSession
	__getattr__ (self, name)
List[tensorinfo]	get_inputs_tensorinfo (self)
List[tensorinfo]	get_outputs_tensorinfo (self)
	set_inputs (self, size, inputs_array=[])

Data Fields
	outputs
Data Fields inherited from package.common.basesession.BaseSession
	session
	inputs
	outputs

Protected Member Functions
None	_update_inputs_tensorinfo (self, List[tensorinfo] new_infos)
	_time_block (self, Dict[str, float] metrics, str key, bool measure)
Protected Member Functions inherited from package.common.basesession.BaseSession
	_recreate_session (self, backend_session)
	_set_outputs (self, size)

Protected Attributes
	_prepared

Detailed Description

Class for inference using nnfw_session.

Definition at line 12 of file session.py.

Constructor & Destructor Documentation

__init__()

None package.infer.session.session.__init__	(		self,
		str	path,
		str	backends = "cpu"
	)

Initialize the inference session.

Args:
    path (str): Path to the model file or nnpackage directory.
    backends (str): Backends to use, default is "cpu".

Reimplemented from package.common.basesession.BaseSession.

Definition at line 16 of file session.py.

    def __init__(self, path: str, backends: str = "cpu") -> None:
        """
        Initialize the inference session.
 
        Args:
            path (str): Path to the model file or nnpackage directory.
            backends (str): Backends to use, default is "cpu".
        """
        super().__init__(infer.nnfw_session(path, backends))
        self._prepared: bool = False
 

References package.infer.session.session.__init__(), nnfw::cker::Conv._prepared, nnfw::cker::FusedBatchNorm._prepared, nnfw::cker::Reduce._prepared, nnfw::ruy::Conv._prepared, package.infer.session.session._prepared, and onert::backend::cpu::ops::DepthwiseConvolutionLayer._prepared.

Referenced by package.infer.session.session.__init__().

Member Function Documentation

_time_block()

package.infer.session.session._time_block (   self, Dict[str, float]  metrics, str  key, bool  measure  )

protected

Definition at line 167 of file session.py.

    def _time_block(self, metrics: Dict[str, float], key: str, measure: bool):
        if measure:
            start = time.perf_counter()
            yield
            metrics[key] = (time.perf_counter() - start) * 1000
        else:
            yield

Referenced by package.infer.session.session.infer().

_update_inputs_tensorinfo()

None package.infer.session.session._update_inputs_tensorinfo (   self, List[tensorinfo]  new_infos  )

protected

Update all input tensors' tensorinfo at once.

Args:
    new_infos (list[tensorinfo]): A list of updated tensorinfo objects for the inputs.

Raises:
    ValueError: If the number of new_infos does not match the session's input size,
                or if any tensorinfo contains a negative dimension.

    OnertError: If the underlying C-API call fails.

Definition at line 134 of file session.py.

    def _update_inputs_tensorinfo(self, new_infos: List[tensorinfo]) -> None:
        """
        Update all input tensors' tensorinfo at once.
 
        Args:
            new_infos (list[tensorinfo]): A list of updated tensorinfo objects for the inputs.
 
        Raises:
            ValueError: If the number of new_infos does not match the session's input size,
                        or if any tensorinfo contains a negative dimension.
 
            OnertError: If the underlying C-API call fails.
        """
        num_inputs: int = self.session.input_size()
        if len(new_infos) != num_inputs:
            raise ValueError(
                f"Expected {num_inputs} input tensorinfo(s), but got {len(new_infos)}.")
 
        for i, info in enumerate(new_infos):
            # Check for any negative dimension in the specified rank
            if any(d < 0 for d in info.dims[:info.rank]):
                raise ValueError(
                    f"Input tensorinfo at index {i} contains negative dimension(s): "
                    f"{info.dims[:info.rank]}")
            try:
                self.session.set_input_tensorinfo(i, info)
            except ValueError:
                # re-raise ValueError directly
                raise
            except Exception as e:
                raise OnertError(f"Failed to update input tensorinfo: {e}") from e
 

References validate_onnx2circle.OnnxRunner.session, onert::api::python::NNFW_SESSION.session, and package.common.basesession.BaseSession.session.

Referenced by package.infer.session.session.infer().

infer()

Union[List[np.ndarray], Tuple[List[np.ndarray], Dict[str, float]]] package.infer.session.session.infer	(		self,
		List[np.ndarray]	inputs_array,
		*bool	measure = False
	)

Run a complete inference cycle:
 - If the session has not been prepared or outputs have not been set, call prepare().
 - Automatically configure input buffers based on the provided numpy arrays.
 - Execute the inference session.
 - Return the output tensors with proper multi-dimensional shapes.

This method supports dynamic shape modification:
 - The input shapes can be adjusted dynamically.

Args:
    inputs_array (list[np.ndarray]): List of numpy arrays representing the input data.
    measure (bool): If True, measure prepare/io/run latencies (ms).

Returns:
    list[np.ndarray]: A list containing the output numpy arrays.
    OR
    (outputs, metrics): Tuple where metrics is a dict with keys
        'prepare_time_ms', 'io_time_ms', 'run_time_ms'

Definition at line 27 of file session.py.

    ) -> Union[List[np.ndarray], Tuple[List[np.ndarray], Dict[str, float]]]:
        """
        Run a complete inference cycle:
         - If the session has not been prepared or outputs have not been set, call prepare().
         - Automatically configure input buffers based on the provided numpy arrays.
         - Execute the inference session.
         - Return the output tensors with proper multi-dimensional shapes.
 
        This method supports dynamic shape modification:
         - The input shapes can be adjusted dynamically.
 
        Args:
            inputs_array (list[np.ndarray]): List of numpy arrays representing the input data.
            measure (bool): If True, measure prepare/io/run latencies (ms).
 
        Returns:
            list[np.ndarray]: A list containing the output numpy arrays.
            OR
            (outputs, metrics): Tuple where metrics is a dict with keys
                'prepare_time_ms', 'io_time_ms', 'run_time_ms'
        """
        metrics: Dict[str, float] = {}
 
        # Verify that the number of provided inputs matches the session's expected input count.
        expected_input_size: int = self.session.input_size()
        if len(inputs_array) != expected_input_size:
            raise ValueError(
                f"Expected {expected_input_size} input(s), but received {len(inputs_array)}."
            )
 
        # Check if the session is prepared. If not, call prepare() once.
        if not self._prepared:
            try:
                with self._time_block(metrics, 'prepare_time_ms', measure):
                    # On first call, fix any -1 dims to real input shapes and validate
                    original_infos = self.get_inputs_tensorinfo()
                    fixed_infos = []
                    for idx, info in enumerate(original_infos):
                        input_shape = inputs_array[idx].shape
                        new_dims = []
                        static_dim_changed = False
                        # only the first `info.rank` entries matter
                        for j, d in enumerate(info.dims[:info.rank]):
                            if d == -1:
                                # replace dynamic dim with actual incoming shape
                                new_dims.append(input_shape[j])
                            elif d == input_shape[j]:
                                # static dim must match the provided array
                                new_dims.append(d)
                            else:
                                static_dim_changed = True
 
                        if static_dim_changed:
                            warnings.warn(
                                f"infer() called with input {idx}'s shape={input_shape}, "
                                f"which differs from model's expected shape={tuple(info.dims)}. "
                                "Ensure this is intended.", UserWarning)
 
                        info.dims = new_dims
                        fixed_infos.append(info)
 
                    # Update tensorinfo to optimize using it
                    self._update_inputs_tensorinfo(fixed_infos)
 
                    self.session.set_prepare_config(
                        prepare_config.ENABLE_INTERNAL_OUTPUT_ALLOC)
                    self.session.prepare()
                    self._prepared = True
            except ValueError:
                raise
            except Exception as e:
                raise OnertError(f"Session preparation failed: {e}") from e
 
        # Configure input buffers using the current session's input size and provided data.
        try:
            with self._time_block(metrics, 'input_time_ms', measure):
                self.set_inputs(expected_input_size, inputs_array)
        except ValueError:
            raise
        except Exception as e:
            raise OnertError(f"Failed to bind inputs: {e}") from e
 
        # Execute the inference.
        try:
            with self._time_block(metrics, 'run_time_ms', measure):
                self.session.run()
        except ValueError:
            raise
        except Exception as e:
            raise OnertError(f"Inference execution failed: {e}") from e
 
        try:
            with self._time_block(metrics, 'output_time_ms', measure):
                self._set_outputs(self.session.output_size())
        except ValueError:
            raise
        except Exception as e:
            raise OnertError(f"Failed to bind outputs: {e}") from e
 
        # Return the output buffers.
        return (self.outputs, metrics) if measure else self.outputs
 

References nnfw::cker::Conv._prepared, nnfw::cker::FusedBatchNorm._prepared, nnfw::cker::Reduce._prepared, nnfw::ruy::Conv._prepared, package.infer.session.session._prepared, onert::backend::cpu::ops::DepthwiseConvolutionLayer._prepared, package.infer.session.session._time_block(), package.infer.session.session._update_inputs_tensorinfo(), package.common.basesession.BaseSession.get_inputs_tensorinfo(), validate_onnx2circle.OnnxRunner.session, onert::api::python::NNFW_SESSION.session, and package.common.basesession.BaseSession.session.

Field Documentation

_prepared

package.infer.session.session._prepared

protected

Definition at line 99 of file session.py.

Referenced by package.infer.session.session.__init__(), and package.infer.session.session.infer().

outputs

package.infer.session.session.outputs

Definition at line 132 of file session.py.

Referenced by package.common.basesession.BaseSession._set_outputs(), and validate_onnx2circle.OnnxRunner.get_outputs().

---

The documentation for this class was generated from the following file:

• runtime/onert/api/python/package/infer/session.py