TrainingExtensions/torch/src/python/aimet_torch/meta/connectedgraph.py
def parse_callmethod_node(self, node: torch._C.Node,
                              trace: Union[torch.jit.TopLevelTracedModule, torch.jit.TracedModule],
                              node_name_to_module: Dict[str, torch.nn.Module],
                              node_name_to_subgraph_model: Dict[str, Tuple[torch.jit.TracedModule, torch._C.Node]],
                              ir_nodes_list: List[IrNode],
                              inputs_map: Dict[torch._C.TensorType, torch._C.TensorType],
                              output_map: Dict[torch._C.TensorType, torch._C.TensorType]):
        # pylint: disable=too-many-locals
        """
        The call method node signifies invocation of the forward method, this method extracts an IrNode representation
        of the module. Typically the node has the following construct:
            %output_N : Tensor = prim::CallMethod[name="forward"](%output_L, %output_M)
        :param node: trace graph node i.e. 'CallMethod' node
        :param trace: trace of model or submodule
        :param node_name_to_module: dictionary of module indexed by output_name referenced in the sub-graph
        :param node_name_to_subgraph_model: dictionary of torch graph nodes index of output_name that have not been
            resolved
        :param ir_nodes_list: List of IrNodes created from traversing the trace graph
        :param inputs_map: Dictionary mapping low recursion level inputs to higher level equivalent inputs
        :param output_map: Dictionary mapping high recursion level outputs to lower level equivalent outputs
        """
        inputs = [inp for inp in node.inputs()]
        # 1st input is a reference on which the call method is being invoked.
        input_name: str = inputs[0].debugName()
        outputs = [output for output in node.outputs()]
        if input_name in node_name_to_subgraph_model:
            subgraph_model, getattr_node_info = node_name_to_subgraph_model[input_name]
            trace_levels = [getattr_node_info.node_name]
            # If node_input (input to the current GetAttr node) is None, we are at the topmost level, and can call
            # trace.<current node name> to get the trace for the subgraph. Otherwise, compile a list of node names to
            # call into by following the subgraph_input entries up the chain.
            while getattr_node_info.node_input is not None:
                _, getattr_node_info = node_name_to_subgraph_model[getattr_node_info.node_input]
                # Later on trace levels will be processed with most recent level first, so insert each one at the front.
                trace_levels.insert(0, getattr_node_info.node_name)
            subgraph_trace = trace
            for level in trace_levels:
                subgraph_trace = getattr(subgraph_trace, level)

            submodule_outputs = self._parse_trace_graph(subgraph_trace, subgraph_model, ir_nodes_list,
                                                        higher_level_inputs=inputs[1:], inputs_map=inputs_map,
                                                        output_map=output_map)
            # Current node is a subgraph. The outputs of the subgraph at this level will correspond to the outputs of
            # the subgraph in the inner recursion level. Update output_map to contain this new mapping.
            assert len(submodule_outputs) == len(outputs)
            for idx, output in enumerate(outputs):
                if submodule_outputs[idx] in output_map:
                    output_map[output] = output_map[submodule_outputs[idx]]
                else:
                    output_map[output] = submodule_outputs[idx]

                # Go through each ir_node and examine the outputs. For any output that was returned from an inner
                # recursion level, replace it with the corresponding output from this level.
                for ir_node in ir_nodes_list:
                    for ir_node_idx, ir_node_output in enumerate(ir_node.outputs):
                        if ir_node_output == submodule_outputs[idx]:
                            ir_node.outputs[ir_node_idx] = outputs[idx]

        elif input_name in node_name_to_module and is_leaf_module(node_name_to_module[input_name]):
            # the graph is fully represented by a directional graph of leaf torch modules so the recursion is
            # stopped at this level. torch.nn.Identity are being ignored because in graph node representation
            # the torch.nn.Identity op generate no output and are not part of inputs for downstream op
            if not isinstance(node_name_to_module[input_name], torch.nn.Identity):
                op_type = self.get_op_type(type(node_name_to_module[input_name]))
                node_inputs = [inp for inp in node.inputs()]
                ir_node = IrNode(node_type=op_type,
                                 inputs=[inputs_map.get(inp, inp) for inp in node_inputs[1:]],
                                 outputs=outputs,
                                 module=node_name_to_module[input_name])
                ir_nodes_list.append(ir_node)

        The call method node signifies invocation of the forward method, this method extracts an IrNode representation
        of the module. Typically the node has the following construct:
            %output_N : Tensor = prim::CallMethod[name="forward"](%output_L, %output_M)
        :param node: trace graph node i.e. 'CallMethod' node
        :param trace: trace of model or submodule
        :param node_name_to_module: dictionary of module indexed by output_name referenced in the sub-graph
        :param node_name_to_subgraph_model: dictionary of torch graph nodes index of output_name that have not been
            resolved
        :param ir_nodes_list: List of IrNodes created from traversing the trace graph
        :param inputs_map: Dictionary mapping low recursion level inputs to higher level equivalent inputs
        :param output_map: Dictionary mapping high recursion level outputs to lower level equivalent outputs
        
The call method node signifies invocation of the forward method, this method extracts an IrNode representation
of the module.
The call method node signifies invocation of the forward method this method extracts an IrNode representation of the module .
def parse_callmethod_node(self, node: torch._C.Node,
                              trace: Union[torch.jit.TopLevelTracedModule, torch.jit.TracedModule],
                              node_name_to_module: Dict[str, torch.nn.Module],
                              node_name_to_subgraph_model: Dict[str, Tuple[torch.jit.TracedModule, torch._C.Node]],
                              ir_nodes_list: List[IrNode],
                              inputs_map: Dict[torch._C.TensorType, torch._C.TensorType],
                              output_map: Dict[torch._C.TensorType, torch._C.TensorType]):
        inputs = [inp for inp in node.inputs()]
        input_name: str = inputs[0].debugName()
        outputs = [output for output in node.outputs()]
        if input_name in node_name_to_subgraph_model:
            subgraph_model, getattr_node_info = node_name_to_subgraph_model[input_name]
            trace_levels = [getattr_node_info.node_name]
            while getattr_node_info.node_input is not None:
                _, getattr_node_info = node_name_to_subgraph_model[getattr_node_info.node_input]
                trace_levels.insert(0, getattr_node_info.node_name)
            subgraph_trace = trace
            for level in trace_levels:
                subgraph_trace = getattr(subgraph_trace, level)
            submodule_outputs = self._parse_trace_graph(subgraph_trace, subgraph_model, ir_nodes_list,
                                                        higher_level_inputs=inputs[1:], inputs_map=inputs_map,
                                                        output_map=output_map)
            assert len(submodule_outputs) == len(outputs)
            for idx, output in enumerate(outputs):
                if submodule_outputs[idx] in output_map:
                    output_map[output] = output_map[submodule_outputs[idx]]
                else:
                    output_map[output] = submodule_outputs[idx]
                for ir_node in ir_nodes_list:
                    for ir_node_idx, ir_node_output in enumerate(ir_node.outputs):
                        if ir_node_output == submodule_outputs[idx]:
                            ir_node.outputs[ir_node_idx] = outputs[idx]
        elif input_name in node_name_to_module and is_leaf_module(node_name_to_module[input_name]):
            if not isinstance(node_name_to_module[input_name], torch.nn.Identity):
                op_type = self.get_op_type(type(node_name_to_module[input_name]))
                node_inputs = [inp for inp in node.inputs()]
                ir_node = IrNode(node_type=op_type,
                                 inputs=[inputs_map.get(inp, inp) for inp in node_inputs[1:]],
                                 outputs=outputs,
                                 module=node_name_to_module[input_name])
                ir_nodes_list.append(ir_node)
def parse_callmethod_node ( self , node : torch . _C . Node , trace : Union [ torch . jit . TopLevelTracedModule , torch . jit . TracedModule ] , node_name_to_module : Dict [ str , torch . nn . Module ] , node_name_to_subgraph_model : Dict [ str , Tuple [ torch . jit . TracedModule , torch . _C . Node ] ] , ir_nodes_list : List [ IrNode ] , inputs_map : Dict [ torch . _C . TensorType , torch . _C . TensorType ] , output_map : Dict [ torch . _C . TensorType , torch . _C . TensorType ] ) : inputs = [ inp for inp in node . inputs ( ) ] input_name : str = inputs [ 0 ] . debugName ( ) outputs = [ output for output in node . outputs ( ) ] if input_name in node_name_to_subgraph_model : subgraph_model , getattr_node_info = node_name_to_subgraph_model [ input_name ] trace_levels = [ getattr_node_info . node_name ] while getattr_node_info . node_input is not None : _ , getattr_node_info = node_name_to_subgraph_model [ getattr_node_info . node_input ] trace_levels . insert ( 0 , getattr_node_info . node_name ) subgraph_trace = trace for level in trace_levels : subgraph_trace = getattr ( subgraph_trace , level ) submodule_outputs = self . _parse_trace_graph ( subgraph_trace , subgraph_model , ir_nodes_list , higher_level_inputs = inputs [ 1 : ] , inputs_map = inputs_map , output_map = output_map ) assert len ( submodule_outputs ) == len ( outputs ) for idx , output in enumerate ( outputs ) : if submodule_outputs [ idx ] in output_map : output_map [ output ] = output_map [ submodule_outputs [ idx ] ] else : output_map [ output ] = submodule_outputs [ idx ] for ir_node in ir_nodes_list : for ir_node_idx , ir_node_output in enumerate ( ir_node . outputs ) : if ir_node_output == submodule_outputs [ idx ] : ir_node . outputs [ ir_node_idx ] = outputs [ idx ] elif input_name in node_name_to_module and is_leaf_module ( node_name_to_module [ input_name ] ) : if not isinstance ( node_name_to_module [ input_name ] , torch . nn . Identity ) : op_type = self . get_op_type ( type ( node_name_to_module [ input_name ] ) ) node_inputs = [ inp for inp in node . inputs ( ) ] ir_node = IrNode ( node_type = op_type , inputs = [ inputs_map . get ( inp , inp ) for inp in node_inputs [ 1 : ] ] , outputs = outputs , module = node_name_to_module [ input_name ] ) ir_nodes_list . append ( ir_node )
The call method node signifies invocation of the forward method, this method extracts an IrNode representation
of the module.
The call method node signifies invocation of the forward method this method extracts an IrNode representation of the module .
# pylint: disable=too-many-locals """
        The call method node signifies invocation of the forward method, this method extracts an IrNode representation
        of the module. Typically the node has the following construct:
            %output_N : Tensor = prim::CallMethod[name="forward"](%output_L, %output_M)
        :param node: trace graph node i.e. 'CallMethod' node
        :param trace: trace of model or submodule
        :param node_name_to_module: dictionary of module indexed by output_name referenced in the sub-graph
        :param node_name_to_subgraph_model: dictionary of torch graph nodes index of output_name that have not been
            resolved
        :param ir_nodes_list: List of IrNodes created from traversing the trace graph
        :param inputs_map: Dictionary mapping low recursion level inputs to higher level equivalent inputs
        :param output_map: Dictionary mapping high recursion level outputs to lower level equivalent outputs
        """ # 1st input is a reference on which the call method is being invoked. # If node_input (input to the current GetAttr node) is None, we are at the topmost level, and can call # trace.<current node name> to get the trace for the subgraph. Otherwise, compile a list of node names to # call into by following the subgraph_input entries up the chain. # Later on trace levels will be processed with most recent level first, so insert each one at the front. # Current node is a subgraph. The outputs of the subgraph at this level will correspond to the outputs of # the subgraph in the inner recursion level. Update output_map to contain this new mapping. # Go through each ir_node and examine the outputs. For any output that was returned from an inner # recursion level, replace it with the corresponding output from this level. # the graph is fully represented by a directional graph of leaf torch modules so the recursion is # stopped at this level. torch.nn.Identity are being ignored because in graph node representation # the torch.nn.Identity op generate no output and are not part of inputs for downstream op
self
node
torch._C.Node
trace graph node i.e. 'CallMethod' node
trace graph node i . e . ' CallMethod ' node
trace
Union[torch.jit.TopLevelTracedModule, torch.jit.TracedModule]
trace of model or submodule
trace of model or submodule
node_name_to_module
Dict[str, torch.nn.Module]
dictionary of module indexed by output_name referenced in the sub-graph
dictionary of module indexed by output_name referenced in the sub - graph
node_name_to_subgraph_model
Dict[str, Tuple[torch.jit.TracedModule, torch._C.Node]]
dictionary of torch graph nodes index of output_name that have not been
resolved
dictionary of torch graph nodes index of output_name that have not been resolved
ir_nodes_list
List[IrNode]
List of IrNodes created from traversing the trace graph
List of IrNodes created from traversing the trace graph
inputs_map
Dict[torch._C.TensorType, torch._C.TensorType]
Dictionary mapping low recursion level inputs to higher level equivalent inputs
Dictionary mapping low recursion level inputs to higher level equivalent inputs
output_map
Dict[torch._C.TensorType, torch._C.TensorType]
Dictionary mapping high recursion level outputs to lower level equivalent outputs
Dictionary mapping high recursion level outputs to lower level equivalent outputs