code: def enumerate_implicit_dependencies(
      components: Collection[base_node.BaseNode],
      registry: dsl_context_registry.DslContextRegistry,
      pipeline: Optional[Pipeline] = None,
  ) -> Iterator[Tuple[base_node.BaseNode, base_node.BaseNode]]:
      """Enumerate component dependencies arising from data deps between them.
  
      Args:
        components: Components to consider.
        registry: DslContextRegistry to use for looking up conditional predicates.
        pipeline: Pipeline object if calling from the context of one.
  
      Yields:
        Pairs of the form (upstream_component, component). If a component has no
        upstream components within `components` then it will not be present as the
        first element of any tuple in the output. A warning is generated if an
        `upstream_component` of some node in `components` is not part of the
        supplied pipeline's components.
  
      Raises:
        RuntimeError: When duplicate components are detected.
      """
      node_by_id = {}
  
      # Fills in producer map.
      for component in components:
          # Checks every node has an unique id.
          if component.id in node_by_id:
              raise RuntimeError(
                  f"Duplicated node_id {component.id} for component type"
                  f"{component.type}. Try setting a different node_id using "
                  "`.with_id()`."
              )
          if pipeline and component.id == pipeline.id:
              raise RuntimeError(
                  f"node id {component.id} is the same as its enclosing pipeline id."
                  "Try setting a different node_id using `.with_id()`."
              )
          node_by_id[component.id] = component
  
      # Deduce upstream nodes based on producer map.
      for component in components:
          channels = list(component.inputs.values())
          for exec_property in component.exec_properties.values():
              if isinstance(exec_property, ph.Placeholder):
                  channels.extend(
                      channel_utils.get_dependent_channels(
                          exec_property
                      )
                  )
          for predicate in conditional.get_predicates(
              component, registry
          ):
              channels.extend(
                  channel_utils.get_dependent_channels(predicate)
              )
  
          pipeline_component_ids = (
              set((component.id for component in pipeline.components))
              if pipeline
              else set()
          )
          for input_channel in channels:
              for (
                  upstream_node_id
              ) in input_channel.get_data_dependent_node_ids():
                  if pipeline and upstream_node_id == pipeline.id:
                      # If a component's input channel depends on the (self) pipeline,
                      # it means that component consumes pipeline-level inputs. No need to
                      # add upstream node here. Pipeline-level inputs will be handled
                      # during compilation.
                      continue
                  upstream_node = node_by_id.get(upstream_node_id)
                  if upstream_node:
                      yield (upstream_node, component)
                  elif (
                      pipeline
                      and upstream_node_id not in pipeline_component_ids
                  ):
                      warnings.warn(
                          f"Node {component.id} depends on the output of node"
                          f" {upstream_node_id}, but {upstream_node_id} is not included in"
                          " the components of pipeline. Did you forget to add it?"
                      )
  
imports: import copy
  import enum
  from typing import Any, Collection, Dict, Iterable, Iterator, List, Optional, Tuple, Union, cast
  import warnings
  from tfx.dsl.compiler import constants
  from tfx.dsl.components.base import base_node
  from tfx.dsl.components.base import executor_spec
  from tfx.dsl.context_managers import dsl_context_registry
  from tfx.dsl.experimental.conditionals import conditional
  from tfx.dsl.placeholder import placeholder
  from tfx.orchestration import data_types
  from tfx.orchestration import metadata
  from tfx.types import channel
  from tfx.types import channel_utils
  from tfx.utils import doc_controls
  from tfx.utils import topsort
  from google.protobuf import message
label_desc: Use image node_id 2 for calling a global function with example usage: enumerate_implicit_dependencies(components, registry, pipeline) without return types