parameters: (self, pipe_tuple, op_name, inputs, kwargs)
return_statement: return res
docstring: Run pipeline wrapper of a single operator.
docstring_summary: Run pipeline wrapper of a single operator.
docstring_tokens: Run pipeline wrapper of a single operator .
function: def _run_subpipeline(self, pipe_tuple, op_name, inputs, kwargs):
          """Run pipeline wrapper of a single operator."""
          self._debug_on = False
          pipe, inputs_es, kwargs_es = pipe_tuple
  
          if len(inputs_es) != len(inputs):
              raise RuntimeError(f"Trying to use operator '{op_name}' with different number of inputs than when"
                                 f" it was built. {len(inputs_es)} != {len(inputs)}")
          if len(kwargs_es) != len(kwargs.items()):
              raise RuntimeError(f"Trying to use operator '{op_name}' with different number of keyward arguments"
                                 " than when it was built.")
  
          def unexpected_argument_msg(to_external_source):
              return f"recognized as {'batch' if to_external_source else 'constant'} but when built value" \
                  f" in its place was recognized as {'constant' if to_external_source else 'batch'}"
  
          for i, input in enumerate(inputs):
              to_external_source, _, data = _PipelineDebug._classify_data(input)
              if to_external_source != inputs_es[i]:
                  raise RuntimeError(
                      f"In operator '{op_name}' input {input} {unexpected_argument_msg(to_external_source)}.")
              if to_external_source:
                  pipe.feed_input(f'{self._es_input_name}{i}', data)
          for key, value in kwargs.items():
              to_external_source, _, data = _PipelineDebug._classify_data(value)
              if to_external_source != kwargs_es[key]:
                  raise RuntimeError(
                      f"In operator '{op_name}' argument '{key}' {unexpected_argument_msg(to_external_source)}.")
              if to_external_source:
                  pipe.feed_input(f'{self._es_kwarg_name}{key}', data)
  
          res = pipe.run()
          res = tuple([DataNodeDebug(tensor, **data_node.__dict__)
                      for tensor, data_node in zip(res, pipe._graph_out)])
  
          self._debug_on = True
  
          if len(res) == 1:
              return res[0]
  
          return res
function_tokens: def _run_subpipeline ( self , pipe_tuple , op_name , inputs , kwargs ) : self . _debug_on = False pipe , inputs_es , kwargs_es = pipe_tuple if len ( inputs_es ) != len ( inputs ) : raise RuntimeError ( f"Trying to use operator '{op_name}' with different number of inputs than when" f" it was built. {len(inputs_es)} != {len(inputs)}" ) if len ( kwargs_es ) != len ( kwargs . items ( ) ) : raise RuntimeError ( f"Trying to use operator '{op_name}' with different number of keyward arguments" " than when it was built." ) def unexpected_argument_msg ( to_external_source ) : return f"recognized as {'batch' if to_external_source else 'constant'} but when built value" f" in its place was recognized as {'constant' if to_external_source else 'batch'}" for i , input in enumerate ( inputs ) : to_external_source , _ , data = _PipelineDebug . _classify_data ( input ) if to_external_source != inputs_es [ i ] : raise RuntimeError ( f"In operator '{op_name}' input {input} {unexpected_argument_msg(to_external_source)}." ) if to_external_source : pipe . feed_input ( f'{self._es_input_name}{i}' , data ) for key , value in kwargs . items ( ) : to_external_source , _ , data = _PipelineDebug . _classify_data ( value ) if to_external_source != kwargs_es [ key ] : raise RuntimeError ( f"In operator '{op_name}' argument '{key}' {unexpected_argument_msg(to_external_source)}." ) if to_external_source : pipe . feed_input ( f'{self._es_kwarg_name}{key}' , data ) res = pipe . run ( ) res = tuple ( [ DataNodeDebug ( tensor , * * data_node . __dict__ ) for tensor , data_node in zip ( res , pipe . _graph_out ) ] ) self . _debug_on = True if len ( res ) == 1 : return res [ 0 ] return res