class_docstr: Compiles a TFX pipeline or a component into a uDSL IR proto.
docstr: Compiles an individual TFX node into a PipelineNode proto.
  
  Args:
    tfx_node: A TFX node.
    pipeline_ctx: Resources needed to compile the node.
    deployment_config: Intermediate deployment config to set. Will include
      related specs for executors, drivers and platform specific configs.
    enable_cache: whether cache is enabled
  
  Raises:
    TypeError: When supplied tfx_node has values of invalid type.
  
  Returns:
    A PipelineNode proto that encodes information of the node.
code: def _compile_node(
      self,
      tfx_node: base_node.BaseNode,
      pipeline_ctx: compiler_context.PipelineContext,
      deployment_config: pipeline_pb2.IntermediateDeploymentConfig,
      enable_cache: bool,
  ) -> pipeline_pb2.PipelineNode:
      """Compiles an individual TFX node into a PipelineNode proto.
  
      Args:
        tfx_node: A TFX node.
        pipeline_ctx: Resources needed to compile the node.
        deployment_config: Intermediate deployment config to set. Will include
          related specs for executors, drivers and platform specific configs.
        enable_cache: whether cache is enabled
  
      Raises:
        TypeError: When supplied tfx_node has values of invalid type.
  
      Returns:
        A PipelineNode proto that encodes information of the node.
      """
      node = pipeline_pb2.PipelineNode()
  
      # Step 1: Node info
      node.node_info.type.name = tfx_node.type
      if (
          isinstance(tfx_node, base_component.BaseComponent)
          and tfx_node.type_annotation
      ):
          node.node_info.type.base_type = (
              tfx_node.type_annotation.MLMD_SYSTEM_BASE_TYPE
          )
      node.node_info.id = tfx_node.id
  
      # Step 2: Node Context
      _set_node_context(node, pipeline_ctx)
  
      # Step 3: Node inputs
      node_inputs_compiler.compile_node_inputs(
          pipeline_ctx, tfx_node, node.inputs
      )
  
      # Step 4: Node outputs
      if isinstance(
          tfx_node, base_component.BaseComponent
      ) or compiler_utils.is_importer(tfx_node):
          _set_node_outputs(node, tfx_node.outputs)
      pipeline_ctx.channels.update(
          _generate_input_spec_for_outputs(node, tfx_node.outputs)
      )
  
      # Step 5: Node parameters
      if not compiler_utils.is_resolver(tfx_node):
          _set_node_parameters(node, tfx_node)
  
      # Step 6: Executor spec and optional driver spec for components
      if isinstance(tfx_node, base_component.BaseComponent):
          executor_spec = tfx_node.executor_spec.encode(
              component_spec=tfx_node.spec
          )
          deployment_config.executor_specs[tfx_node.id].Pack(
              executor_spec
          )
  
          # TODO(b/163433174): Remove specialized logic once generalization of
          # driver spec is done.
          if tfx_node.driver_class != base_driver.BaseDriver:
              driver_class_path = _fully_qualified_name(
                  tfx_node.driver_class
              )
              driver_spec = (
                  executable_spec_pb2.PythonClassExecutableSpec()
              )
              driver_spec.class_path = driver_class_path
              deployment_config.custom_driver_specs[tfx_node.id].Pack(
                  driver_spec
              )
  
      # Step 7: Upstream/Downstream nodes
      upstreams = set(
          _find_runtime_upstream_node_ids(pipeline_ctx, tfx_node)
      )
      if _begin_node_is_upstream(tfx_node, pipeline_ctx.pipeline):
          upstreams.add(
              compiler_utils.pipeline_begin_node_id(
                  pipeline_ctx.pipeline
              )
          )
      # Sort node ids so that compiler generates consistent results.
      node.upstream_nodes.extend(sorted(upstreams))
  
      downstreams = set(
          _find_runtime_downstream_node_ids(pipeline_ctx, tfx_node)
      )
      if _end_node_is_downstream(tfx_node, pipeline_ctx.pipeline):
          downstreams.add(
              compiler_utils.pipeline_end_node_id(pipeline_ctx.pipeline)
          )
      # Sort node ids so that compiler generates consistent results.
      node.downstream_nodes.extend(sorted(downstreams))
  
      # Step 8: Node execution options and triggers.
      _set_node_execution_options(
          node, tfx_node, pipeline_ctx, enable_cache
      )
  
      # Step 9: Per-node platform config
      if isinstance(tfx_node, base_component.BaseComponent):
          tfx_component = cast(base_component.BaseComponent, tfx_node)
          if tfx_component.platform_config:
              deployment_config.node_level_platform_configs[
                  tfx_node.id
              ].Pack(tfx_component.platform_config)
  
      return node
  
imports: import itertools
  from typing import Any, Callable, Dict, Iterator, List, Optional, Tuple, Type, cast
  from tfx import types
  from tfx.dsl.compiler import compiler_context
  from tfx.dsl.compiler import compiler_utils
  from tfx.dsl.compiler import constants
  from tfx.dsl.compiler import node_inputs_compiler
  from tfx.dsl.components.base import base_component
  from tfx.dsl.components.base import base_driver
  from tfx.dsl.components.base import base_node
  from tfx.dsl.experimental.node_execution_options import utils
  from tfx.dsl.placeholder import placeholder
  from tfx.orchestration import data_types
  from tfx.orchestration import data_types_utils
  from tfx.orchestration import pipeline
  from tfx.proto.orchestration import executable_spec_pb2
  from tfx.proto.orchestration import pipeline_pb2
  from tfx.types import channel
  from tfx.types import channel_utils
  from tfx.utils import deprecation_utils
  from tfx.utils import name_utils
label_desc: Use image node_id 4 for calling the Compiler obj's underlying member method code with example usage: obj._compile_node(tfx_node, pipeline_ctx, deployment_config, enable_cache) and returns: node