class_docstr: Compiles a TFX pipeline or a component into a uDSL IR proto.
docstr: Compiles a tfx pipeline into uDSL proto.
  
  Args:
    tfx_pipeline: A TFX pipeline.
    parent_pipeline_ctx: Optional PipelineContext that includes info for
      the immediate parent pipeline. This is mainly used by a pipeline begin
      node get info for artifacts from its parent pipeline.
  
  Returns:
    A Pipeline proto that encodes all necessary information of the pipeline.
code: def compile(
      self,
      tfx_pipeline: pipeline.Pipeline,
      parent_pipeline_ctx: Optional[
          compiler_context.PipelineContext
      ] = None,
  ) -> pipeline_pb2.Pipeline:
      """Compiles a tfx pipeline into uDSL proto.
  
      Args:
        tfx_pipeline: A TFX pipeline.
        parent_pipeline_ctx: Optional PipelineContext that includes info for
          the immediate parent pipeline. This is mainly used by a pipeline begin
          node get info for artifacts from its parent pipeline.
  
      Returns:
        A Pipeline proto that encodes all necessary information of the pipeline.
      """
      # Prepare pipeline compiler context.
      pipeline_ctx = compiler_context.PipelineContext(
          tfx_pipeline, parent_pipeline_ctx
      )
  
      tfx_pipeline.finalize()
      _validate_pipeline(tfx_pipeline, pipeline_ctx.parent_pipelines)
  
      pipeline_pb = pipeline_pb2.Pipeline()
      pipeline_pb.pipeline_info.id = (
          pipeline_ctx.pipeline_info.pipeline_name
      )
      pipeline_pb.execution_mode = pipeline_ctx.execution_mode
  
      if isinstance(
          pipeline_ctx.pipeline_info.pipeline_root,
          placeholder.Placeholder,
      ):
          pipeline_pb.runtime_spec.pipeline_root.placeholder.CopyFrom(
              pipeline_ctx.pipeline_info.pipeline_root.encode()
          )
      else:
          # Unless an inner pipeline specified its own pipeline root, it inherits
          # from its closest parent pipeline that has a pipelie root defined.
          pipeline_root_str = ""
          if tfx_pipeline.pipeline_info.pipeline_root:
              pipeline_root_str = (
                  tfx_pipeline.pipeline_info.pipeline_root
              )
          else:
              for parent_pipeline in reversed(
                  pipeline_ctx.parent_pipelines
              ):
                  if parent_pipeline.pipeline_info.pipeline_root:
                      pipeline_root_str = (
                          parent_pipeline.pipeline_info.pipeline_root
                      )
                      break
          compiler_utils.set_runtime_parameter_pb(
              pipeline_pb.runtime_spec.pipeline_root.runtime_parameter,
              constants.PIPELINE_ROOT_PARAMETER_NAME,
              str,
              pipeline_root_str,
          )
  
      if (
          pipeline_pb.execution_mode
          == pipeline_pb2.Pipeline.ExecutionMode.SYNC
      ):
          # TODO(kennethyang): Miragte all pipeline run ids to structural runtime
          # parameter. Currently only subpipelines use structural runtime
          # parameter for IR textproto compatibility.
          if not pipeline_ctx.is_root:
              compiler_utils.set_structural_runtime_parameter_pb(
                  pipeline_pb.runtime_spec.pipeline_run_id.structural_runtime_parameter,
                  [
                      f"{pipeline_pb.pipeline_info.id}_",
                      (constants.PIPELINE_RUN_ID_PARAMETER_NAME, str),
                  ],
              )
          else:
              compiler_utils.set_runtime_parameter_pb(
                  pipeline_pb.runtime_spec.pipeline_run_id.runtime_parameter,
                  constants.PIPELINE_RUN_ID_PARAMETER_NAME,
                  str,
              )
  
      deployment_config = pipeline_pb2.IntermediateDeploymentConfig()
      if tfx_pipeline.metadata_connection_config:
          deployment_config.metadata_connection_config.Pack(
              tfx_pipeline.metadata_connection_config
          )
  
      # Inner pipelines of a composable pipeline, or a outmost pipeline with
      # pipeline-level inputs have pipeline begin nodes.
      if (
          not pipeline_ctx.is_root or tfx_pipeline._inputs
      ):  # pylint: disable=protected-access
          pipeline_begin_node_pb = self._compile_pipeline_begin_node(
              tfx_pipeline, pipeline_ctx
          )
          pipeline_or_node = pipeline_pb.PipelineOrNode()
          pipeline_or_node.pipeline_node.CopyFrom(
              pipeline_begin_node_pb
          )
          pipeline_pb.nodes.append(pipeline_or_node)
  
      for node in tfx_pipeline.components:
          if isinstance(node, pipeline.Pipeline):
              pipeline_node_pb = self.compile(node, pipeline_ctx)
              pipeline_or_node = pipeline_pb.PipelineOrNode()
              pipeline_or_node.sub_pipeline.CopyFrom(pipeline_node_pb)
              pipeline_pb.nodes.append(pipeline_or_node)
          else:
              node_pb = self._compile_node(
                  node,
                  pipeline_ctx,
                  deployment_config,
                  tfx_pipeline.enable_cache,
              )
              pipeline_or_node = pipeline_pb.PipelineOrNode()
              pipeline_or_node.pipeline_node.CopyFrom(node_pb)
              pipeline_pb.nodes.append(pipeline_or_node)
  
      # Inner pipelines of a composable pipeline, or a outmost pipeline with
      # pipeline-level outputs have pipeline end nodes.
      if (
          not pipeline_ctx.is_root or tfx_pipeline._outputs
      ):  # pylint: disable=protected-access
          pipeline_end_node_pb = self._compile_pipeline_end_node(
              tfx_pipeline, pipeline_ctx
          )
          pipeline_or_node = pipeline_pb.PipelineOrNode()
          pipeline_or_node.pipeline_node.CopyFrom(pipeline_end_node_pb)
          pipeline_pb.nodes.append(pipeline_or_node)
  
      if tfx_pipeline.platform_config:
          deployment_config.pipeline_level_platform_config.Pack(
              tfx_pipeline.platform_config
          )
      pipeline_pb.deployment_config.Pack(deployment_config)
      return pipeline_pb
  
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 5 for calling the Compiler obj's underlying member method code with example usage: obj.compile(tfx_pipeline, parent_pipeline_ctx) and returns: pipeline_pb