afe.apis.model

Classes

Model

Module Contents

class afe.apis.model.Model(net: afe.ir.net.AwesomeNet, fp32_net: afe.ir.net.AwesomeNet | None = None)

execute(inputs: afe.apis.defines.InputValues, *, fast_mode: bool = False, log_level: int | None = logging.NOTSET, keep_layer_outputs: list[afe.ir.defines.NodeName] | str | None = None, output_file_path: str | None = None) → List[numpy.ndarray]

Run input data through the quantized model.

Parameters:

• inputs – Dictionary of placeholder node names (str) to the input data.

• fast_mode – If True, use a fast implementation of operators. If False, use an implementation that exactly matches execution on the MLA.

• log_level – Logging level.

• keep_layer_outputs – List of quantized model layer output names that should be saved. Each element of a list must have a valid name, according to the model layer outputs. Iff ‘all’, all intermediate results are saved.

• output_file_path – Location where the layer outputs should be saved. If defined, keep_layer_outputs argument must also be provided by the user.

Returns: Outputs of quantized model.

Also, saves the requested intermediate results inside output_file_path location.

save(model_name: str, output_directory: str = '', *, log_level: int | None = logging.NOTSET) → None

Save the quantized model and its floating-point counterpart (if available) to the specified directory.

Defaults to the current working directory if no output directory is provided.

Parameters:

• model_name (str) – Name for the saved quantized model file with a .sima extension.

• output_directory (str, optional) – Directory to save the model files. Defaults to the current working directory.

• log_level (Optional[int], optional) – Logging level for the operation. Defaults to logging.NOTSET.

Raises:

UserFacingException – If an error occurs during the save process.

Example

>>> model = Model(quantized_net, fp32_net)
>>> model.save("my_model", output_directory="models/")

static load(model_name: str, network_directory: str = '', *, log_level: int | None = logging.NOTSET) → Model

compile(output_path: str, batch_size: int = 1, compress: bool = True, log_level: int | None = logging.NOTSET, tessellate_parameters: afe.backends.mpk.interface.TessellateParameters | None = None, l2_caching_mode: afe.backends.mpk.interface.L2CachingMode = L2CachingMode.NONE, **kwargs) → None

Compile the quantized model into a .tar.gz package for deployment in an MPK package.

The compiled package includes the binary model and a JSON structure file, saved in output_path as <model_name>_mpk.tar.gz. Batch size can be specified, though compiler optimizations may adjust it for optimal performance. The first dimension of input tensors must represent batch size.

Parameters:

• output_path (str) – Directory to save the compiled package. Created if it doesn’t exist.

• batch_size (int, optional) – Batch size for compilation. Defaults to 1.

• compress (bool, optional) – Enable DRAM data compression for the .lm file. Defaults to True.

• log_level (Optional[int], optional) – Logging level. Defaults to logging.NOTSET.

• tessellate_parameters (Optional[TessellateParameters], optional) – Internal use for MLA tessellation parameters.

• l2_caching_mode (L2CachingMode, optional) – Internal use for N2A compiler’s L2 caching. Defaults to L2CachingMode.NONE.

• **kwargs –

  Additional internal options, including:

  • retained_temporary_directory_name (str): Path to retain intermediate files.

  • use_power_limits (bool): Enable power limits during compilation.

  • max_mla_power (float): Set maximum MLA power consumption.

  • layer_norm_use_fp32_intermediates (bool): Use FP32 intermediates for layer normalization.

  • rms_norm_use_fp32_intermediates (bool): Use FP32 intermediates for RMS normalization.

Raises:

UserFacingException – If compilation fails due to invalid parameters or errors.

Example:

>>> model = Model(quantized_net)
>>> model.compile(output_path="compiled_models/", batch_size=4, compress=True)

static create_auxiliary_network(transforms: List[afe.apis.transform.Transform], input_types: Dict[afe.ir.defines.InputName, afe.ir.tensor_type.TensorType], *, target: sima_utils.common.Platform = gen1_target, log_level: int | None = logging.NOTSET) → Model

static compose(nets: List[Model], combined_model_name: str = 'main', log_level: int | None = logging.NOTSET) → Model

evaluate(evaluation_data: Iterable[Tuple[afe.apis.defines.InputValues, afe.apis.compilation_job_base.GroundTruth]], criterion: afe.apis.statistic.Statistic[Tuple[List[numpy.ndarray], afe.apis.compilation_job_base.GroundTruth], str], *, fast_mode: bool = False, log_level: int | None = logging.NOTSET) → str

Evaluate the model using the provided evaluation data and criterion.

This method runs the model on the given dataset and computes an aggregate result using the specified criterion. It supports a fast execution mode for quicker evaluations and customizable logging levels for diagnostic purposes.

Parameters:

• evaluation_data – An iterable of tuples where each tuple contains input values and the corresponding ground truth for evaluation.

• criterion – A statistical function used to compute the evaluation metric based on the model’s outputs and ground truth.

• fast_mode – Optional; if set to True, the evaluation will execute in a faster but potentially less thorough manner. Defaults to False.

• log_level – Optional; specifies the logging level for evaluation. Defaults to logging.NOTSET.

Returns:

A string representing the final result of the evaluation based on the criterion.

Raises:

Exception – If an error occurs during model execution, it is sanitized and re-raised.

analyze_quantization_error(evaluation_data: Iterable[afe.apis.defines.InputValues], error_metric: afe.core.graph_analyzer.utils.Metric, *, local_feed: bool, log_level: int | None = logging.NOTSET)

get_performance_metrics(output_kpi_path: str, *, log_level: int | None = logging.NOTSET)

generate_elf_and_reference_files(input_data: Iterable[afe.apis.defines.InputValues], output_dir: str, *, batch_size: int = 1, compress: bool = True, tessellate_parameters: afe.backends.mpk.interface.TessellateParameters | None = None, log_level: int | None = logging.NOTSET, l2_caching_mode: afe.backends.mpk.interface.L2CachingMode = L2CachingMode.NONE) → None

execute_in_accelerator_mode(input_data: Iterable[afe.apis.defines.InputValues], devkit: str, *, username: str = cp.DEFAULT_USERNAME, password: str = '', batch_size: int = 1, compress: bool = True, tessellate_parameters: afe.backends.mpk.interface.TessellateParameters | None = None, log_level: int | None = logging.NOTSET, l2_caching_mode: afe.backends.mpk.interface.L2CachingMode = L2CachingMode.NONE) → List[numpy.ndarray]