#########################################################
# Copyright (C) 2022 SiMa Technologies, Inc.
#
# This material is SiMa proprietary and confidential.
#
# This material may not be copied or distributed without
# the express prior written permission of SiMa.
#
# All rights reserved.
#########################################################
# Code owner: Ljubomir Papuga
#########################################################
"""
This is the pre-release API for AFE. It supports importing models,
loading and storing AFE's internal format, quantizing, executing, and
simulating.
"""
import numpy as np
import os
from typing import Dict, Tuple, List, Optional
import subprocess
import csv
import tempfile
from sima_utils.data.data_generator import DataGenerator
from afe.core.utils import convert_data_generator_to_iterable as _convert_data_generator_to_iterable
from afe.ir.defines import Status
from afe.ir.net import AwesomeNet as _AwesomeNet
import afe.core.quantize_networks as _quantize_networks
import afe.core.compile_networks as _compile_networks
import afe.load.loader as _loader
from afe.ir.tensor_type import ScalarType
from sima_utils.common import Platform
from afe.apis.defines import gen1_target
#######################################
# Load API
#######################################
[docs]
def load_onnx_model(model_path: str,
shape_dict: Dict[str, Tuple[int, ...]],
dtype_dict: Dict[str, ScalarType],
layout: str = 'NCHW',
is_quantized: bool = False
) -> _AwesomeNet:
"""
Loads an ONNX model into an _AwesomeNet format.
:param model_path: Path to a .onnx file containing the onnx model
:param shape_dict: Dictionary of input names to input shapes eg. {'input', (1,224,224,3)}
:param dtype_dict: Dictionary of input names to input types eg. {'input', 'float32'}
:param layout: Input data layout. Default is 'NCHW'
:param is_quantized: Whether the model is pre-quantized. Default is False
:return: An _AwesomeNet
"""
return _loader.load_onnx_model(model_path, shape_dict, dtype_dict, layout,
convert_layout=True, is_quantized=is_quantized)
[docs]
def load_pytorch_model(model_path: str,
input_names: List[str],
input_shapes: List[Tuple[int, ...]],
input_dtypes: Optional[List[ScalarType]] = None,
layout: str = 'NCHW',
is_quantized: bool = False
) -> _AwesomeNet:
"""
Loads a PyTorch model into an _AwesomeNet format.
:param model_path: Path to a PyTorch file (.pt) that contains the entire model
:param input_names: List of input names. eg ['input']
:param input_shapes: List of input shapes corresponding to input names. eg. [(1, 224, 224, 3)]
:param input_dtypes: List of input datatypes corresponding to input names. eg ['float32']
:param layout: Input data layout. Default is 'NCHW'
:param is_quantized: Whether the model is pre-quantized. Default is False
:return: An _AwesomeNet
"""
return _loader.load_pytorch_model(model_path, input_names, input_shapes, input_dtypes,
layout, convert_layout=True, is_quantized=is_quantized)
[docs]
def load_tensorflow_model(model_path: str,
shape_dict: Dict[str, Tuple[int, ...]],
output_names: List[str],
layout: str = 'NHWC',
is_quantized: bool = False
) -> _AwesomeNet:
"""
Loads a TensorFlow model into an _AwesomeNet format
:param model_path: Path to a .pb TensorFlow model
:param shape_dict: Dictionary of input names to input shapes eg. {'input', (1,224,224,3)}
:param output_names: List of output names of the network eg. ['output1', 'output2']
:param layout: any variation of the characters NHWC representing Batch Size, Height, Width, and Channels
:param is_quantized: Whether the model is pre-quantized. Default is False
:return: An _AwesomeNet
"""
return _loader.load_tensorflow_model(model_path, shape_dict, output_names, layout,
convert_layout=True, is_quantized=is_quantized)
[docs]
def load_tensorflow2_model(model_path: str,
shape_dict: Dict[str, Tuple[int, ...]],
output_names: List[str],
layout: str = 'NHWC',
is_quantized: bool = False
) -> _AwesomeNet:
"""
Loads a TensorFlow model into an _AwesomeNet format
:param model_path: Path to a SavedModel tensorflow model.
:param shape_dict: Dictionary of input names to input shapes eg. {'input', (1,224,224,3)}
:param output_names: List of output names of the network eg. ['output1', 'output2']
:param layout: any variation of the characters NHWC representing Batch Size, Height, Width, and Channels
:param is_quantized: Whether the model is pre-quantized. Default is False
:return: An _AwesomeNet
"""
return _loader.load_tensorflow2_model(model_path, shape_dict, output_names, layout,
convert_layout=True, is_quantized=is_quantized)
[docs]
def load_tflite_model(model_path: str,
shape_dict: Dict[str, Tuple[int, ...]],
dtype_dict: Dict[str, ScalarType],
layout: str = 'NHWC',
is_quantized: bool = False
) -> _AwesomeNet:
"""
Loads a TensorFlow Lite model into an _AwesomeNet format.
:param model_path: Path to a .tflite file containing the TensorFlow Lite model
:param shape_dict: Dictionary of input names to input shapes eg. {'input', (1,224,224,3)}
:param dtype_dict: Dictionary of input names to input types eg. {'input', 'float32'}
:param layout: Input data layout. Default is 'NHWC'
:param is_quantized: Whether the model is pre-quantized. Default is False
:return: An _AwesomeNet
"""
return _loader.load_tflite_model(model_path, shape_dict, dtype_dict, layout,
convert_layout=True, is_quantized=is_quantized)
[docs]
def load_keras_model(model_path: str,
shape_dict: Dict[str, Tuple[int, ...]],
layout: str = 'NCHW',
is_quantized: bool = False
) -> _AwesomeNet:
"""
Loads a Keras model into an _AwesomeNet format.
:param model_path: Path to a .h5 file containing the keras model
:param shape_dict: Dictionary of input names to input shapes eg. {'input', (1,224,224,3)}
:param layout: Input data layout. Default is 'NCHW'
:param is_quantized: Whether the model is pre-quantized. Default is False
:return: An _AwesomeNet
"""
return _loader.load_keras_model(model_path, shape_dict, layout,
convert_layout=True, is_quantized=is_quantized)
[docs]
def load_caffe_model(prototxt_file_path: str,
caffemodel_file_path: str,
shape_dict: Dict[str, Tuple[int, ...]],
dtype_dict: Dict[str, ScalarType],
layout: str = 'NCHW',
is_quantized: bool = False
) -> _AwesomeNet:
"""
Loads a caffe model into an _AwesomeNet format.
:param prototxt_file_path: filepath to the caffe .prototxt file
:param caffemodel_file_path: filepath to the caffe .caffemodel file
:param shape_dict: Dictionary of input names to input shapes eg. {'input', (1,224,224,3)}
:param dtype_dict: Dictionary of input names to input types eg. {'input', 'float32'}
:param layout: Input data layout. Default is 'NCHW'
:param is_quantized: Whether the model is pre-quantized. Default is False
:return: An _AwesomeNet
"""
return _loader.load_caffe_model(prototxt_file_path, caffemodel_file_path, shape_dict, dtype_dict,
layout, convert_layout=True, is_quantized=is_quantized)
[docs]
def load_caffe2_model(init_net_file_path: str,
predict_net_file_path: str,
shape_dict: Dict[str, Tuple[int, ...]],
dtype_dict: Dict[str, ScalarType],
layout: str = 'NCHW',
is_quantized: bool = False
) -> _AwesomeNet:
"""
Loads a caffe2 model into an _AwesomeNet format.
:param init_net_file_path: filepath to the caffe2 .pb init_net file
:param predict_net_file_path: filepath to the caffe2 .pb predict_net file
:param shape_dict: Dictionary of input names to input shapes eg. {'input', (1,224,224,3)}
:param dtype_dict: Dictionary of input names to input types eg. {'input', 'float32'}
:param layout: Input data layout. Default is 'NCHW'
:param is_quantized: Whether the model is pre-quantized. Default is False
:return: An _AwesomeNet
"""
return _loader.load_caffe2_model(init_net_file_path, predict_net_file_path, shape_dict, dtype_dict,
layout, convert_layout=True, is_quantized=is_quantized)
#######################################
# Quantize API
#######################################
[docs]
def quantize_net(net: _AwesomeNet,
input_generator: DataGenerator,
asymmetry: bool,
per_channel: bool,
max_calibration_samples: int):
"""
Quantizes a network.
:param net: an _AwesomeNet
:param input_generator: a DataGenerator we use to feed inputs into the _AwesomeNet during calibration
:param asymmetry: If True this function performs asymmetric quantization. Otherwise, it performs symmetric
quantization
:param per_channel: If True this function performs per_channel quantization.
:param max_calibration_samples: Maximum number of samples we use for calibration
"""
from afe.core.utils import wrap_parameters_to_model_configs, wrap_parameters_to_optimization_configs
model_configs = wrap_parameters_to_model_configs(name=net.name)
opt_configs = wrap_parameters_to_optimization_configs(asymmetry=asymmetry, per_channel=per_channel)
# Truncate the input to the given limit
input_generator = _convert_data_generator_to_iterable(input_generator, length_limit=max_calibration_samples)
_quantize_networks.quantize_network(net, model_configs, opt_configs, input_generator)
#######################################
# Execute API
#######################################
[docs]
def execute_net(net: _AwesomeNet,
inputs: Dict[str, np.ndarray],
dequantize_outputs: bool = True) -> List[np.ndarray]:
"""
Executes a network.
:param net: an _AwesomeNet
:param inputs: Dictionary of placeholder node names (str) to the input data
:param dequantize_outputs: Whether to dequantize the output data to floating point
:return The result of executing the output node
"""
from afe.ir.execute import execute_node, execute_node_quant
assert dequantize_outputs == True, "This option cannot be modified"
node_callable = execute_node_quant if net.status == Status.SIMA_QUANTIZED else execute_node
return net.run(inputs, node_callable=node_callable)
#######################################
# Compile API
#######################################
[docs]
def compile_net(net: _AwesomeNet,
output_elf_path: str,
compress: bool = True):
"""
Compile a network using Product Compiler.
:param net: an _AwesomeNet.
:param output_elf_path: Path in which elf files should be created.
:param compress: If True mlc file is compressed before generating .elf file.
"""
_ = _compile_networks.compile_net_to_elf(net, output_elf_path, compress=compress,
# enable_large_tensors MUST be False when called from user-facing APIs
enable_large_tensors=False)
#######################################
# ISIM API
#######################################
[docs]
def simulate_isim_module(elf_file: str,
output_path: str,
output_kpi_file_name: Optional[str] = None,
output_trace_file_name: Optional[str] = None,
generate_csv_kpi: Optional[bool] = True,
override_existing: Optional[bool] = False,
target: Optional[Platform] = None):
"""
Wrapper to take elf file and generate summary KPI file, and trace file.
:param elf_file: Path to the location of the elf file to simulate in ISIM.
:param output_path: Location of where to store all output files. If location does not exist, it will be created.
:param output_kpi_file_name: Name for output kpi file. Default: elf_file name without '.elf' extension and
finishing with '_kpis.txt' postfix.
:param output_trace_file_name: Name for output trace file. Default: elf_file name without '.elf' extension and
finishing with '_trace.mpack' postfix.
:param generate_csv_kpi: Whether to generate a CSV file with the KPIs in addition to the txt.
:param override_existing: Whether to override existing kpis and trace files from a previous run.
:param target: [Optional] platform target.
"""
assert elf_file.endswith('.elf'), "elf file should contain the extension \'.elf\'"
elf_file_name = os.path.splitext(os.path.basename(elf_file))[0] # elf file name without extension
if output_kpi_file_name is None:
output_kpi_file_name = os.path.join(output_path, elf_file_name + "_kpis.txt")
if target and (target == Platform.GEN1):
# mla-isim supports --mpack
if output_trace_file_name is None:
output_trace_file_name = os.path.join(output_path, elf_file_name + "_trace.mpack")
output_kpi_csv_path = os.path.join(output_path, elf_file_name + "_kpis.csv")
# Generate output directory if it does not exist, delete existing files if already exists (override_flag)
if not os.path.exists(output_path):
os.makedirs(output_path)
else:
isim_paths_error_msg = "When running ISIM: Output file: {} already exists. To override any output files, " \
"use override_existing=True"
if not override_existing:
if os.path.exists(output_kpi_file_name):
raise FileExistsError(isim_paths_error_msg.format(output_kpi_file_name))
if (target == Platform.GEN1) and os.path.exists(output_trace_file_name):
raise FileExistsError(isim_paths_error_msg.format(output_trace_file_name))
if os.path.exists(output_kpi_file_name):
raise FileExistsError(isim_paths_error_msg.format(output_kpi_file_name))
isim_net(elf_file=elf_file, output_kpi_path=output_kpi_file_name, output_trace_path=output_trace_file_name,
output_kpi_csv_path=output_kpi_csv_path, csv_format=generate_csv_kpi, target=target)
[docs]
def isim_net(elf_file: str,
output_kpi_path: str,
output_trace_path: Optional[str] = None,
output_kpi_csv_path: Optional[str] = None,
csv_format: Optional[bool] = False,
target: Optional[Platform] = None):
"""
Take elf file and generate summary KPI file, and trace file.
:param elf_file: a elf_file
:param output_kpi_path: Path in which kpi files should be created
:param output_trace_path: Optional Path in which trace files should be created. Only mla-isim supports --mpack.
:param output_kpi_csv_path: Path in which kpi files in csv format should be created
:param csv_format: Generate the Summary KPI in CVS format
:param target: [Optional] platform target.
"""
if not os.path.exists(elf_file):
raise Exception("{} does not exist".format(elf_file))
if (target == Platform.GEN1):
assert output_trace_path.endswith('.mpack'), "output_trace_path should be in binary MessagePack format"
# Generate the Summary KPI, and the trace file in binary MessagePack format, for visualizer.
command = ["mla-isim", elf_file, "--stats={output_kpi_path}".format(output_kpi_path=output_kpi_path), \
"--mpack={output_trace_path}".format(output_trace_path=output_trace_path)]
else:
# Generate the Summary KPI, and the trace file in binary MessagePack format, for visualizer.
command = ["mla-msim", elf_file, "--stats={output_kpi_path}".format(output_kpi_path=output_kpi_path)]
p_command = subprocess.run(command)
if p_command.returncode != 0:
raise Exception("Error occurred when generating the trace file, and summary KPI")
if csv_format:
try:
import yaml
except Exception as e:
raise Exception("Could not load the yamlmodule (pypi package pyyaml)")
with open(output_kpi_path, "r") as f:
isim_stats_dict = yaml.load(f, Loader=yaml.FullLoader)
with open(output_kpi_csv_path, 'w') as csv_file:
writer = csv.writer(csv_file)
for key, value in isim_stats_dict.items():
writer.writerow([key, value])