# Copyright 2022 Huawei Technologies Co., Ltd
#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
# http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.
# ============================================================================
"""
Model API.
"""
import os
from enum import Enum
from ._checkparam import check_isinstance
from .context import Context
from .lib import _c_lite_wrapper
from .tensor import Tensor
__all__ = ['ModelType', 'Model', 'RunnerConfig', 'ModelParallelRunner']
class ModelType(Enum):
"""
The MoedelType is used to define the model type.
"""
MINDIR = 0
MINDIR_LITE = 4
model_type_py_cxx_map = {
ModelType.MINDIR: _c_lite_wrapper.ModelType.kMindIR,
ModelType.MINDIR_LITE: _c_lite_wrapper.ModelType.kMindIR_Lite,
}
model_type_cxx_py_map = {
_c_lite_wrapper.ModelType.kMindIR: ModelType.MINDIR,
_c_lite_wrapper.ModelType.kMindIR_Lite: ModelType.MINDIR_LITE,
}
[docs]class Model:
"""
The Model class is used to define a MindSpore model, facilitating computational graph management.
Examples:
>>> import mindspore_lite as mslite
>>> model = mslite.Model()
>>> print(model)
model_path: .
"""
def __init__(self):
self._model = _c_lite_wrapper.ModelBind()
self.model_path_ = ""
def __str__(self):
res = f"model_path: {self.model_path_}."
return res
[docs] def build_from_file(self, model_path, model_type, context):
"""
Load and build a model from file.
Args:
model_path (str): Define the model path, include model file.
model_type (ModelType): Define The type of model file. Options: ModelType::MINDIR | ModelType::MINDIR_LITE.
- ModelType::MINDIR: An intermediate representation of the MindSpore model.
The recommended model file suffix is ".mindir".
- ModelType::MINDIR_LITE: An intermediate representation of the MindSpore Lite model.
The recommended model file suffix is ".ms".
context (Context): Define the context used to store options during execution.
Raises:
TypeError: `model_path` is not a str.
TypeError: `model_type` is not a ModelType.
TypeError: `context` is not a Context.
RuntimeError: `model_path` does not exist.
RuntimeError: build from file failed.
Examples:
>>> import mindspore_lite as mslite
>>> model = mslite.Model()
>>> context = mslite.Context()
>>> context.append_device_info(mslite.CPUDeviceInfo())
>>> model.build_from_file("mobilenetv2.ms", mslite.ModelType.MINDIR_LITE, context)
>>> print(model)
model_path: mobilenetv2.ms.
"""
check_isinstance("model_path", model_path, str)
check_isinstance("model_type", model_type, ModelType)
check_isinstance("context", context, Context)
if not os.path.exists(model_path):
raise RuntimeError(f"build_from_file failed, model_path does not exist!")
self.model_path_ = model_path
model_type_ = _c_lite_wrapper.ModelType.kMindIR_Lite
if model_type is ModelType.MINDIR:
model_type_ = _c_lite_wrapper.ModelType.kMindIR
ret = self._model.build_from_file(self.model_path_, model_type_, context._context)
if not ret.IsOk():
raise RuntimeError(f"build_from_file failed! Error is {ret.ToString()}")
[docs] def resize(self, inputs, dims):
"""
Resizes the shapes of inputs.
Args:
inputs (list[Tensor]): A list that includes all input tensors in order.
dims (list[list[int]]): A list that includes the new shapes of inputs, should be consistent with inputs.
Raises:
TypeError: `inputs` is not a list.
TypeError: `inputs` is a list, but the elements are not Tensor.
TypeError: `dims` is not a list.
TypeError: `dims` is a list, but the elements are not list.
TypeError: `dims` is a list, the elements are list, but the element's elements are not int.
ValueError: The size of `inputs` is not equal to the size of `dims` .
RuntimeError: resize input failed.
Examples:
>>> import mindspore_lite as mslite
>>> model = mslite.Model()
>>> context = mslite.Context()
>>> context.append_device_info(mslite.CPUDeviceInfo())
>>> model.build_from_file("mobilenetv2.ms", mslite.ModelType.MINDIR_LITE, context)
>>> inputs = model.get_inputs()
>>> print("Before resize, the first input shape: ", inputs[0].get_shape())
Before resize, the first input shape: [1, 224, 224, 3]
>>> model.resize(inputs, [[1, 112, 112, 3]])
>>> print("After resize, the first input shape: ", inputs[0].get_shape())
After resize, the first input shape: [1, 112, 112, 3]
"""
if not isinstance(inputs, list):
raise TypeError("inputs must be list, but got {}.".format(type(inputs)))
_inputs = []
if not isinstance(dims, list):
raise TypeError("dims must be list, but got {}.".format(type(dims)))
for i, element in enumerate(inputs):
if not isinstance(element, Tensor):
raise TypeError(f"inputs element must be Tensor, but got "
f"{type(element)} at index {i}.")
for i, element in enumerate(dims):
if not isinstance(element, list):
raise TypeError(f"dims element must be list, but got "
f"{type(element)} at index {i}.")
for j, dim in enumerate(element):
if not isinstance(dim, int):
raise TypeError(f"dims element's element must be int, but got "
f"{type(dim)} at {i}th dims element's {j}th element.")
if len(inputs) != len(dims):
raise ValueError(f"inputs' size does not match dims's size, but got "
f"inputs: {len(inputs)} and dims: {len(dims)}.")
for i, element in enumerate(inputs):
_inputs.append(element._tensor)
ret = self._model.resize(_inputs, dims)
if not ret.IsOk():
raise RuntimeError(f"resize failed! Error is {ret.ToString()}")
[docs] def predict(self, inputs, outputs):
"""
Inference model.
Args:
inputs (list[Tensor]): A list that includes all input tensors in order.
outputs (list[Tensor]): The model outputs are filled in the container in sequence.
Raises:
TypeError: `inputs` is not a list.
TypeError: `inputs` is a list, but the elements are not Tensor.
TypeError: `outputs` is not a list.
TypeError: `outputs` is a list, but the elements are not Tensor.
RuntimeError: predict model failed.
Examples:
>>> # 1. predict which indata is from file
>>> import mindspore_lite as mslite
>>> import numpy as np
>>> model = mslite.Model()
>>> context = mslite.Context()
>>> context.append_device_info(mslite.CPUDeviceInfo())
>>> model.build_from_file("mobilenetv2.ms", mslite.ModelType.MINDIR_LITE, context)
>>> inputs = model.get_inputs()
>>> outputs = model.get_outputs()
>>> in_data = np.fromfile("input.bin", dtype=np.float32)
>>> inputs[0].set_data_from_numpy(in_data)
>>> model.predict(inputs, outputs)
>>> for output in outputs:
... data = output.get_data_to_numpy()
... print("outputs: ", data)
...
outputs: [[1.02271215e-05 9.92699006e-06 1.69684317e-05 ... 6.69087376e-06
2.16263197e-06 1.24009384e-04]]
>>> # 2. predict which indata is numpy array
>>> import mindspore_lite as mslite
>>> import numpy as np
>>> model = mslite.Model()
>>> context = mslite.Context()
>>> context.append_device_info(mslite.CPUDeviceInfo())
>>> model.build_from_file("mobilenetv2.ms", mslite.ModelType.MINDIR_LITE, context)
>>> inputs = model.get_inputs()
>>> outputs = model.get_outputs()
>>> for input in inputs:
... in_data = np.arange(1 * 224 * 224 * 3, dtype=np.float32).reshape((1, 224, 224, 3))
... input.set_data_from_numpy(in_data)
...
>>> model.predict(inputs, outputs)
>>> for output in outputs:
... data = output.get_data_to_numpy()
... print("outputs: ", data)
...
outputs: [[0.00035889 0.00065501 0.00052925 ... 0.00018388 0.00148316 0.00116824]]
>>> # 3. predict which indata is new mslite tensor with numpy array
>>> import mindspore_lite as mslite
>>> import numpy as np
>>> model = mslite.Model()
>>> context = mslite.Context()
>>> context.append_device_info(mslite.CPUDeviceInfo())
>>> model.build_from_file("mobilenetv2.ms", mslite.ModelType.MINDIR_LITE, context)
>>> inputs = model.get_inputs()
>>> outputs = model.get_outputs()
>>> input_tensors = []
>>> for input in inputs:
... input_tensor = mslite.Tensor()
... input_tensor.set_data_type(input.get_data_type())
... input_tensor.set_shape(input.get_shape())
... input_tensor.set_format(input.get_format())
... input_tensor.set_tensor_name(input.get_tensor_name())
... in_data = np.arange(1 * 224 * 224 * 3, dtype=np.float32).reshape((1, 224, 224, 3))
... input_tensor.set_data_from_numpy(in_data)
... input_tensors.append(input_tensor)
...
>>> model.predict(input_tensors, outputs)
>>> for output in outputs:
... data = output.get_data_to_numpy()
... print("outputs: ", data)
...
outputs: [[0.00035889 0.00065501 0.00052925 ... 0.00018388 0.00148316 0.00116824]]
"""
if not isinstance(inputs, list):
raise TypeError("inputs must be list, but got {}.".format(type(inputs)))
if not isinstance(outputs, list):
raise TypeError("outputs must be list, but got {}.".format(type(outputs)))
_inputs = []
for i, element in enumerate(inputs):
if not isinstance(element, Tensor):
raise TypeError(f"inputs element must be Tensor, but got "
f"{type(element)} at index {i}.")
_inputs.append(element._tensor)
_outputs = []
for i, element in enumerate(outputs):
if not isinstance(element, Tensor):
raise TypeError(f"outputs element must be Tensor, but got "
f"{type(element)} at index {i}.")
_outputs.append(element._tensor)
ret = self._model.predict(_inputs, _outputs, None, None)
if not ret.IsOk():
raise RuntimeError(f"predict failed! Error is {ret.ToString()}")
[docs] def get_outputs(self):
"""
Obtains all output tensors of the model.
Returns:
list[Tensor], the outputs tensor list of the model.
Examples:
>>> import mindspore_lite as mslite
>>> model = mslite.Model()
>>> context = mslite.Context()
>>> context.append_device_info(mslite.CPUDeviceInfo())
>>> model.build_from_file("mobilenetv2.ms", mslite.ModelType.MINDIR_LITE, context)
>>> outputs = model.get_outputs()
"""
outputs = []
for _tensor in self._model.get_outputs():
outputs.append(Tensor(_tensor))
return outputs
[docs] def get_output_by_tensor_name(self, tensor_name):
"""
Obtains the output tensor of the model by name.
Args:
tensor_name (str): tensor name.
Returns:
Tensor, the output tensor of the tensor name.
Raises:
TypeError: `tensor_name` is not a str.
RuntimeError: get output by tensor name failed.
Examples:
>>> import mindspore_lite as mslite
>>> model = mslite.Model()
>>> context = mslite.Context()
>>> context.append_device_info(mslite.CPUDeviceInfo())
>>> model.build_from_file("mobilenetv2.ms", mslite.ModelType.MINDIR_LITE, context)
>>> output_tensor = model.get_output_by_tensor_name("Softmax-65")
>>> print(output_tensor)
tensor_name: Softmax-65,
data_type: DataType.FLOAT32,
shape: [1, 1001],
format: Format.NHWC,
element_num: 1001,
data_size: 4004.
"""
check_isinstance("tensor_name", tensor_name, str)
_tensor = self._model.get_output_by_tensor_name(tensor_name)
if _tensor.is_null():
raise RuntimeError(f"get_output_by_tensor_name failed!")
return Tensor(_tensor)
[docs]class RunnerConfig:
"""
RunnerConfig Class defines runner config of one or more servables.
The class can be used to make model parallel runner which corresponds to the service provided by a model.
The client sends inference tasks and receives inference results through server.
Args:
context (Context, optional): Define the context used to store options during execution. Default: None.
workers_num (int, optional): the num of workers. Default: None.
config_info (dict{str, dict{str, str}}, optional): Nested map for passing model weight paths.
e.g. {"weight": {"weight_path": "/home/user/weight.cfg"}}. Default: None.
key currently supports ["weight"];
value is in dict format, key of it currently supports ["weight_path"],
value of it is the path of weight, e.g. "/home/user/weight.cfg".
Raises:
TypeError: `context` is neither a Context nor None.
TypeError: `workers_num` is neither an int nor None.
TypeError: `config_info` is neither a dict nor None.
TypeError: `config_info` is a dict, but the key is not str.
TypeError: `config_info` is a dict, the key is str, but the value is not dict.
TypeError: `config_info` is a dict, the key is str, the value is dict, but the key of value is not str.
TypeError: `config_info` is a dict, the key is str, the value is dict, the key of the value is str, but
the value of the value is not str.
ValueError: `workers_num` is an int, but it is less than 0.
Examples:
>>> # only for serving inference
>>> import mindspore_lite as mslite
>>> context = mslite.Context()
>>> context.append_device_info(mslite.CPUDeviceInfo())
>>> config_info = {"weight": {"weight_path": "path of model weight"}}
>>> runner_config = mslite.RunnerConfig(context=context, workers_num=0, config_info=config_info)
>>> print(runner_config)
workers num: 0,
config info: weight: weight_path path of model weight
,
context: thread num: 0, bind mode: 1.
"""
def __init__(self, context=None, workers_num=None, config_info=None, config_path=None):
if context is not None:
check_isinstance("context", context, Context)
if workers_num is not None:
check_isinstance("workers_num", workers_num, int)
if workers_num < 0:
raise ValueError(f"RunnerConfig's init failed! workers_num must be positive.")
if config_info is not None:
check_isinstance("config_info", config_info, dict)
for k, v in config_info.items():
check_isinstance("config_info_key", k, str)
check_isinstance("config_info_value", v, dict)
for v_k, v_v in v.items():
check_isinstance("config_info_value_key", v_k, str)
check_isinstance("config_info_value_value", v_v, str)
self._runner_config = _c_lite_wrapper.RunnerConfigBind()
if context is not None:
self._runner_config.set_context(context._context)
if workers_num is not None:
self._runner_config.set_workers_num(workers_num)
if config_info is not None:
for k, v in config_info.items():
self._runner_config.set_config_info(k, v)
if config_path is not None:
if not os.path.exists(config_path):
raise ValueError(f"RunnerConfig's init failed, config_path does not exist!")
check_isinstance("config_path", config_path, str)
self._runner_config.set_config_path(config_path)
def __str__(self):
res = f"workers num: {self._runner_config.get_workers_num()},\n" \
f"config info: {self._runner_config.get_config_info_string()},\n" \
f"context: {self._runner_config.get_context_info()},\n" \
f"config path: {self._runner_config.get_config_path()}."
return res
[docs]class ModelParallelRunner:
"""
The ModelParallelRunner class is used to define a MindSpore ModelParallelRunner, facilitating Model management.
Examples:
>>> # only for serving inference
>>> import mindspore_lite as mslite
>>> model_parallel_runner = mslite.ModelParallelRunner()
>>> print(model_parallel_runner)
model_path: .
"""
def __init__(self):
self._model = _c_lite_wrapper.ModelParallelRunnerBind()
self.model_path_ = ""
def __str__(self):
return f"model_path: {self.model_path_}."
[docs] def init(self, model_path, runner_config=None):
"""
build a model parallel runner from model path so that it can run on a device.
Args:
model_path (str): Define the model path.
runner_config (RunnerConfig, optional): Define the config used to store options during model pool init.
Default: None.
Raises:
TypeError: `model_path` is not a str.
TypeError: `runner_config` is neither a RunnerConfig nor None.
RuntimeError: `model_path` does not exist.
RuntimeError: ModelParallelRunner's init failed.
Examples:
>>> import mindspore_lite as mslite
>>> context = mslite.Context()
>>> context.append_device_info(mslite.CPUDeviceInfo())
>>> runner_config = mslite.RunnerConfig(context=context, workers_num=4)
>>> model_parallel_runner = mslite.ModelParallelRunner()
>>> model_parallel_runner.init(model_path="mobilenetv2.ms", runner_config=runner_config)
>>> print(model_parallel_runner)
model_path: mobilenetv2.ms.
"""
check_isinstance("model_path", model_path, str)
if not os.path.exists(model_path):
raise RuntimeError(f"ModelParallelRunner's init failed, model_path does not exist!")
self.model_path_ = model_path
if runner_config is not None:
check_isinstance("runner_config", runner_config, RunnerConfig)
ret = self._model.init(self.model_path_, runner_config._runner_config)
else:
ret = self._model.init(self.model_path_, None)
if not ret.IsOk():
raise RuntimeError(f"ModelParallelRunner's init failed! Error is {ret.ToString()}")
[docs] def predict(self, inputs, outputs):
"""
Inference ModelParallelRunner.
Args:
inputs (list[Tensor]): A list that includes all input tensors in order.
outputs (list[Tensor]): The model outputs are filled in the container in sequence.
Raises:
TypeError: `inputs` is not a list.
TypeError: `inputs` is a list, but the elements are not Tensor.
TypeError: `outputs` is not a list.
TypeError: `outputs` is a list, but the elements are not Tensor.
RuntimeError: predict model failed.
Examples:
>>> import mindspore_lite as mslite
>>> import numpy as np
>>> context = mslite.Context()
>>> context.append_device_info(mslite.CPUDeviceInfo())
>>> runner_config = mslite.RunnerConfig(context=context, workers_num=4)
>>> model_parallel_runner = mslite.ModelParallelRunner()
>>> model_parallel_runner.init(model_path="mobilenetv2.ms", runner_config=runner_config)
>>> inputs = model_parallel_runner.get_inputs()
>>> in_data = np.fromfile("input.bin", dtype=np.float32)
>>> inputs[0].set_data_from_numpy(in_data)
>>> outputs = model_parallel_runner.get_outputs()
>>> model_parallel_runner.predict(inputs, outputs)
>>> for output in outputs:
... data = output.get_data_to_numpy()
... print("outputs: ", data)
...
outputs: [[1.02271215e-05 9.92699006e-06 1.69684317e-05 ... 6.69087376e-06
2.16263197e-06 1.24009384e-04]]
"""
if not isinstance(inputs, list):
raise TypeError("inputs must be list, but got {}.".format(type(inputs)))
if not isinstance(outputs, list):
raise TypeError("outputs must be list, but got {}.".format(type(outputs)))
_inputs = []
for i, element in enumerate(inputs):
if not isinstance(element, Tensor):
raise TypeError(f"inputs element must be Tensor, but got "
f"{type(element)} at index {i}.")
_inputs.append(element._tensor)
_outputs = []
for i, element in enumerate(outputs):
if not isinstance(element, Tensor):
raise TypeError(f"outputs element must be Tensor, but got "
f"{type(element)} at index {i}.")
_outputs.append(element._tensor)
predict_output = self._model.predict(_inputs, _outputs, None, None)
if not isinstance(predict_output, list) or len(predict_output) == 0:
raise RuntimeError(f"predict failed!")
outputs.clear()
for _, element in enumerate(predict_output):
outputs.append(Tensor(element))
[docs] def get_outputs(self):
"""
Obtains all output tensors of the model.
Returns:
list[Tensor], the outputs tensor list of the model.
Examples:
>>> import mindspore_lite as mslite
>>> context = mslite.Context()
>>> context.append_device_info(mslite.CPUDeviceInfo())
>>> runner_config = mslite.RunnerConfig(context=context, workers_num=4)
>>> model_parallel_runner = mslite.ModelParallelRunner()
>>> model_parallel_runner.init(model_path="mobilenetv2.ms", runner_config=runner_config)
>>> outputs = model_parallel_runner.get_outputs()
"""
outputs = []
for _tensor in self._model.get_outputs():
outputs.append(Tensor(_tensor))
return outputs