#########################################################
# 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 development API for AFE. It supports importing models,
loading and storing AFE's internal format, quantizing, executing, and
simulating.
"""
import copy
from typing import List, Dict, Optional, Tuple
from afe import __version__
from afe.apis.defines import gen1_target
from afe.apis.transform import Transform
import afe.ir.build_node as _build_node
from afe.ir.defines import InputName, NodeName, Status, TupleValue, data_value_elements, get_expected_tensor_value
from afe.ir.net import AwesomeNet, inline_awesomenet_subgraphs, Renaming, rename_awesomenet_nodes
from afe.ir.node import AwesomeNode, node_is_tuple
from afe.ir.tensor_type import TensorType
from sima_utils.common import Platform
from sima_utils.data.data_generator import DataGenerator, get_dummy_data_generator
from ev_transforms.transforms import resize
from afe.core.utils import length_hinted, convert_data_generator_to_iterable
def _create_nodes_for_transform(transform: Transform,
transform_idx: int,
input_type: TensorType,
input_node_name: NodeName,
net_name: str) -> List[AwesomeNode]:
"""
Creates list of AwesomeNodes for one Transform. The list must contain at least one node,
representing the placeholder node holding the transform input.
:param transform: Transform. The Transform for which the nodes are created.
:param transform_idx: int. The Transform's index in the transforms list. Represents the
index of the input to which the transform is applied.
:param input_type: TensorType. The tensor type of the input to the Transform.
:param input_node_name: NodeName. The name of the input node for the Transform.
:param net_name: str. The network name.
:return: List[AwesomeNode]. The list of created AwesomeNodes.
"""
node_list: List[AwesomeNode] = list()
node_list.append(_build_node.create_placeholder_node(node_name=input_node_name, input_type=input_type))
node_name = f"{net_name}/{transform_idx}"
out_type, nodes = transform.extract_ir(input_type, input_node_name, node_name)
node_list.extend(nodes)
return node_list
def _create_node_dict_from_transforms(transforms: List[Transform],
input_types: Dict[InputName, TensorType],
input_node_names: List[NodeName],
net_name: str) -> Tuple[Dict[NodeName, AwesomeNode], List[NodeName]]:
"""
Creates dictionary of AwesomeNodes, generating one list of AwesomeNodes for each transform.
Also, returns the list of output node names, corresponding to the names of the last nodes
in the generated node list for each transform.
:param transforms: List[Transform]. The list of tensor transformations, each applied to a
single input.
:param input_types: Dict[InputName, TensorType]. Dictionary containing (name, type)
pairs for each input.
:param input_node_names: List[NodeName]. List of node names for each input.
:param net_name: str. The name of the AwesomeNet which will be produced using the resulting
AwesomeNodes.
:return: Tuple[Dict[NodeName, AwesomeNode], List[NodeName]]. The dictionary of AwesomeNodes
and the list of names representing the outputting AwesomeNodes.
"""
node_dict: Dict[NodeName, AwesomeNode] = dict()
output_node_names: List[NodeName] = list()
for idx, (transform, input_type, input_node_name) in \
enumerate(zip(transforms, input_types.values(), input_node_names)):
node_list = _create_nodes_for_transform(transform, idx, input_type, input_node_name, net_name)
assert len(node_list) > 0, f"Created empty node list for transform {transform.name}"
node_dict.update({node.name: node for node in node_list})
output_node_names.append(node_list[-1].name)
return node_dict, output_node_names
[docs]
def create_auxiliary_processing_network(transforms: List[Transform],
input_types: Dict[InputName, TensorType], *,
input_node_names: Optional[List[NodeName]] = None,
net_name: str = "aux_net",
status: Status = Status.RELAY,
target: Platform = gen1_target) \
-> AwesomeNet:
"""
Creates an AwesomeNet from the list of Transforms.
:param transforms: The list of Transforms. Each transform in the list correspond to one input.
:param input_types: The list of input types. Each input correspond to one Transform.
:param input_node_names: If set, determines the input names of the resulting AwesomeNet.
:param net_name: The name of the resulting AwesomeNet. Default it "aux_net".
:param status: The status of the created AwesomeNet. Default is Status.RELAY.
:param target: A target platform that a model is compiled for.
:return: AwesomeNet containing nodes corresponding to the transforms list.
"""
# Each transform in the list should be associated with a single input.
assert len(transforms) == len(input_types)
# Generate input_node_names list if it's not provided by user.
if input_node_names is None:
input_node_names = [f"{net_name}/placeholder_{idx}" for idx in range(len(transforms))]
assert len(transforms) == len(input_node_names)
node_dict, output_node_names = _create_node_dict_from_transforms(transforms, input_types,
input_node_names, net_name)
if len(output_node_names) > 1:
# Create tuple node.
output_nodes = [node_dict[node_name] for node_name in output_node_names]
tuple_node = _build_node.create_tuple_output_node(output_nodes, "aux_net")
output_node_name = tuple_node.name
node_dict.update({output_node_name: tuple_node})
else:
assert len(output_node_names) == 1, "Got empty list of output_node_names."
output_node_name = output_node_names[-1]
return AwesomeNet(name=net_name,
nodes=node_dict,
input_node_names=input_node_names if input_node_names is not None else ["input"],
output_node_name=output_node_name,
_status=status,
_target=target)
[docs]
def compose_awesomenets(nets: List[AwesomeNet], status: Status = Status.RELAY,
combined_model_name: str = 'main') -> AwesomeNet:
"""
Creates an AwesomeNet form the list of AwesomeNets. Each AwesomeNet in the list
of input AwesomeNets becomes the subnet of the resulting AwesomeNet.
:param nets: List[AwesomeNet]. The list of input AwesomeNet which are to be composed
into a single AwesomeNet.
:param status: Parameter setting the status of composed network. Default value is Status.RELAY
:param combined_model_name: Combined model name.
:return: The AwesomeNet consisting of the input AwesomeNets.
"""
# Copy the original AwesomeNets, so we don't mutate them.
nets_mutated = [copy.deepcopy(net) for net in nets]
nodes: List[AwesomeNode] = list()
# Create placeholder nodes.
input_node_names = nets[0].input_node_names
input_nodes = [nets[0].nodes[n] for n in input_node_names]
input_node_types = [get_expected_tensor_value(n.get_type().output) for n in input_nodes]
placeholder_nodes = [_build_node.create_placeholder_node(input_node_name, input_node_type)
for (input_node_name, input_node_type) in zip(input_node_names, input_node_types)]
nodes.extend(placeholder_nodes)
# Rename all AwesomeNodes to avoid duplicate names.
for idx, net in enumerate(nets_mutated):
renaming = Renaming({node_name: f"subgraph_{idx}_{node_name}" for node_name in net.nodes})
rename_awesomenet_nodes(renaming, net)
net.topological_sort()
untuple_input: bool = False
for idx, net in enumerate(nets_mutated):
# Un-tuple the output from previous AwesomeNet, if needed.
if untuple_input:
tuple_node = nodes[-1]
tuple_node_output_type = tuple_node.get_type().output
assert isinstance(tuple_node_output_type, TupleValue)
tuple_node_output_types = data_value_elements(tuple_node_output_type)
tuple_get_item_nodes = _build_node.create_tuple_get_item_nodes(input_node_names[0],
tuple_node_output_types,
prefix=f"subnet_{idx-1}")
nodes.extend(tuple_get_item_nodes)
input_node_names = [n.name for n in tuple_get_item_nodes]
output_node = net.nodes[net.output_node_name]
node = AwesomeNode(name=f"subnet_{idx}", input_names=net.input_node_names,
input_node_names=input_node_names,
ir=net)
nodes.append(node)
input_node_names = [node.name]
untuple_input = node_is_tuple(output_node)
targets = set([net.target for net in nets_mutated])
if len(targets) > 1:
raise ValueError("Cannot compose networks built for different targets!")
composed_net = AwesomeNet(name=combined_model_name, nodes={n.name: n for n in nodes},
input_node_names=nets[0].input_node_names,
output_node_name=nodes[-1].name, _status=status, _target=targets.pop(),
_fp_input_range=nets[0].fp_input_range)
inline_awesomenet_subgraphs(composed_net, lambda x: True)
return composed_net
[docs]
def get_model_sdk_version() -> str:
return __version__