class_docstr: Components orchestration engine.
  
  Builds a graph of components and orchestrates their execution according to the execution graph.
class_docstr_tok: ["Components","orchestration","engine",".","Builds","a","graph","of","components","and","orchestrates","their","execution","according","to","the","execution","graph","."]
docstr: Runs the pipeline.
  
  Args:
      data: the inputs to give to the input components of the Pipeline.
      debug: whether to collect and return debug information.
  
  Returns:
      A dictionary with the outputs of the output components of the Pipeline.
  
  Raises:
      PipelineRuntimeError: if the any of the components fail or return unexpected output.
docstr_tok: ["Runs","the","pipeline",".","Args",":","data",":","the","inputs","to","give","to","the","input","components","of","the","Pipeline",".","debug",":","whether","to","collect","and","return","debug","information",".","Returns",":","A","dictionary","with","the","outputs","of","the","output","components","of","the","Pipeline",".","Raises",":","PipelineRuntimeError",":","if","the","any","of","the","components","fail","or","return","unexpected","output","."]
code: def run(
      self, data: Dict[str, Any], debug: bool = False
  ) -> Dict[str, Any]:  # pylint: disable=too-many-locals
      """
      Runs the pipeline.
  
      Args:
          data: the inputs to give to the input components of the Pipeline.
          debug: whether to collect and return debug information.
  
      Returns:
          A dictionary with the outputs of the output components of the Pipeline.
  
      Raises:
          PipelineRuntimeError: if the any of the components fail or return unexpected output.
      """
      self._clear_visits_count()
      data = validate_pipeline_input(self.graph, input_values=data)
      logger.info("Pipeline execution started.")
  
      self._debug = {}
      if debug:
          logger.info("Debug mode ON.")
          os.makedirs("debug", exist_ok=True)
  
      logger.debug(
          "Mandatory connections:\n%s",
          "\n".join(
              f" - {component}: {', '.join([str(s) for s in sockets])}"
              for component, sockets in self._mandatory_connections.items()
          ),
      )
  
      self.warm_up()
  
      # Prepare the inputs buffers and components queue
      components_queue: List[str] = []
      mandatory_values_buffer: Dict[Connection, Any] = {}
      optional_values_buffer: Dict[Connection, Any] = {}
      pipeline_output: Dict[str, Dict[str, Any]] = defaultdict(dict)
  
      for node_name, input_data in data.items():
          for socket_name, value in input_data.items():
              # Make a copy of the input value so components don't need to
              # take care of mutability.
              value = deepcopy(value)
              connection = Connection(
                  None,
                  None,
                  node_name,
                  self.graph.nodes[node_name]["input_sockets"][
                      socket_name
                  ],
              )
              self._add_value_to_buffers(
                  value,
                  connection,
                  components_queue,
                  mandatory_values_buffer,
                  optional_values_buffer,
              )
  
      # *** PIPELINE EXECUTION LOOP ***
      step = 0
      while components_queue:  # pylint: disable=too-many-nested-blocks
          step += 1
          if debug:
              self._record_pipeline_step(
                  step,
                  components_queue,
                  mandatory_values_buffer,
                  optional_values_buffer,
                  pipeline_output,
              )
  
          component_name = components_queue.pop(0)
          logger.debug(
              "> Queue at step %s: %s %s",
              step,
              component_name,
              components_queue,
          )
          self._check_max_loops(component_name)
  
          # **** RUN THE NODE ****
          if not self._ready_to_run(
              component_name, mandatory_values_buffer, components_queue
          ):
              continue
  
          inputs = {
              **self._extract_inputs_from_buffer(
                  component_name, mandatory_values_buffer
              ),
              **self._extract_inputs_from_buffer(
                  component_name, optional_values_buffer
              ),
          }
          outputs = self._run_component(
              name=component_name, inputs=dict(inputs)
          )
  
          # **** PROCESS THE OUTPUT ****
          for socket_name, value in outputs.items():
              targets = self._collect_targets(
                  component_name, socket_name
              )
              if not targets:
                  pipeline_output[component_name][socket_name] = value
              else:
                  for target in targets:
                      self._add_value_to_buffers(
                          value,
                          target,
                          components_queue,
                          mandatory_values_buffer,
                          optional_values_buffer,
                      )
  
      if debug:
          self._record_pipeline_step(
              step + 1,
              components_queue,
              mandatory_values_buffer,
              optional_values_buffer,
              pipeline_output,
          )
          os.makedirs(self._debug_path, exist_ok=True)
          with open(
              self._debug_path / "data.json", "w", encoding="utf-8"
          ) as datafile:
              json.dump(self._debug, datafile, indent=4, default=str)
          pipeline_output["_debug"] = self._debug  # type: ignore
  
      logger.info("Pipeline executed successfully.")
      return dict(pipeline_output)
  
imports: from typing import Optional, Any, Dict, List, Union, TypeVar, Type, Set
  import os
  import json
  import datetime
  import logging
  import importlib
  from pathlib import Path
  from copy import deepcopy
  from collections import defaultdict
  import networkx
  from haystack.core.component import component, Component, InputSocket, OutputSocket
  from haystack.core.errors import PipelineError, PipelineConnectError, PipelineMaxLoops, PipelineRuntimeError, PipelineValidationError
  from haystack.core.pipeline.descriptions import find_pipeline_outputs
  from haystack.core.pipeline.draw.draw import _draw, RenderingEngines
  from haystack.core.pipeline.validation import validate_pipeline_input, find_pipeline_inputs
  from haystack.core.component.connection import Connection, parse_connect_string
  from haystack.core.type_utils import _type_name
  from haystack.core.serialization import component_to_dict, component_from_dict
label_desc: Use image node_id 13 for calling the Pipeline obj's underlying member method code with example usage: obj.run(data, debug) and returns: dict