class_docstr: A task scheduler for subpipeline.
code: def schedule(self) -> task_scheduler.TaskSchedulerResult:
      if self._cancel.is_set() or self._get_pipeline_view() is not None:
          logging.info(
              "Cancel was set OR pipeline view was not none, skipping start,"
              " cancel.is_set(): %s, view exists: %s",
              self._cancel.is_set(),
              self._get_pipeline_view() is not None,
          )
      else:
          try:
              # Only create a begin node execution if we need to start the pipeline.
              # If we don't need to start the pipeline this likely means the pipeline
              # was already started so the execution should already exist.
              self._put_begin_node_execution()
              logging.info(
                  "[Subpipeline Task Scheduler]: start subpipeline."
              )
              pipeline_ops.initiate_pipeline_start(
                  self.mlmd_handle, self._sub_pipeline, None, None
              )
          except status_lib.StatusNotOkError as e:
              return task_scheduler.TaskSchedulerResult(
                  status=e.status()
              )
  
      while not self._cancel.wait(_POLLING_INTERVAL_SECS.value):
          view = self._get_pipeline_view()
          if view:
              if execution_lib.is_execution_successful(view.execution):
                  return task_scheduler.TaskSchedulerResult(
                      status=status_lib.Status(code=status_lib.Code.OK)
                  )
              if execution_lib.is_execution_failed(view.execution):
                  return task_scheduler.TaskSchedulerResult(
                      status=status_lib.Status(
                          code=status_lib.Code.ABORTED,
                          message="Subpipeline execution is failed.",
                      )
                  )
              if execution_lib.is_execution_canceled(view.execution):
                  return task_scheduler.TaskSchedulerResult(
                      status=status_lib.Status(
                          code=status_lib.Code.CANCELLED,
                          message="Subpipeline execution is cancelled.",
                      )
                  )
          else:
              return task_scheduler.TaskSchedulerResult(
                  status=status_lib.Status(
                      code=status_lib.Code.INTERNAL,
                      message=(
                          "Failed to find the subpipeline run with run id: "
                          f"{self._pipeline_run_id}."
                      ),
                  )
              )
  
      view = self._get_pipeline_view()
      if view and execution_lib.is_execution_active(view.execution):
          logging.info(
              "[Subpipeline Task Scheduler]: stopping subpipeline %s",
              self._pipeline_uid,
          )
          pipeline_ops.stop_pipeline(
              self.mlmd_handle, self._pipeline_uid
          )
          logging.info(
              "[Subpipeline Task Scheduler]: subpipeline stopped %s",
              self._pipeline_uid,
          )
      return task_scheduler.TaskSchedulerResult(
          status=status_lib.Status(code=status_lib.Code.CANCELLED)
      )
  
imports: import copy
  import threading
  from typing import Callable, Optional
  from absl import flags
  from absl import logging
  from tfx.orchestration import metadata
  from tfx.orchestration.experimental.core import pipeline_ops
  from tfx.orchestration.experimental.core import pipeline_state
  from tfx.orchestration.experimental.core import task
  from tfx.orchestration.experimental.core import task_scheduler
  from tfx.orchestration.portable.mlmd import context_lib
  from tfx.orchestration.portable.mlmd import execution_lib
  from tfx.proto.orchestration import pipeline_pb2
  from tfx.utils import status
  from ml_metadata.proto import metadata_store_pb2
label_desc: Use image node_id 4 for calling the SubPipelineTaskScheduler obj's underlying member method code with example usage: obj.schedule() and returns: task_scheduler, task_scheduler, task_scheduler, task_scheduler, task_scheduler, task_scheduler