code: def __init__(
      self,
      stage_id: str,
      task: Task,
      chunk_graph: ChunkGraph,
      subtask_graph: SubtaskGraph,
      bands: List[BandType],
      tile_context: Dict[TileableType, TileableType],
      scheduling_api: SchedulingAPI,
      meta_api: MetaAPI,
  ):
      self.stage_id = stage_id
      self.task = task
      self.chunk_graph = chunk_graph
      self.subtask_graph = subtask_graph
      self._bands = bands
      self._tile_context = tile_context
  
      # APIs
      self._scheduling_api = scheduling_api
      self._meta_api = meta_api
  
      # gen subtask_id to subtask
      self.subtask_id_to_subtask = {
          subtask.subtask_id: subtask for subtask in subtask_graph
      }
      self._subtask_to_bands: Dict[Subtask, BandType] = dict()
      self.subtask_snapshots: Dict[Subtask, SubtaskResult] = dict()
      self.subtask_results: Dict[Subtask, SubtaskResult] = dict()
      self._submitted_subtask_ids = set()
  
      # All subtask IDs whose input chunk reference count is reduced.
      self.decref_subtask = set()
  
      self._band_manager: Dict[BandType, mo.ActorRef] = dict()
  
      # result
      self.result = TaskResult(
          task.task_id,
          task.session_id,
          self.stage_id,
          status=TaskStatus.pending,
          start_time=time.time(),
      )
      # status
      self._done = asyncio.Event()
      self._cancelled = asyncio.Event()
  
      # add metrics
      self._stage_execution_time = Metrics.gauge(
          "mars.stage_execution_time_secs",
          "Time consuming in seconds to execute a stage",
          ("session_id", "task_id", "stage_id"),
      )
  
imports: import asyncio
  import itertools
  import logging
  import time
  from collections import defaultdict
  from typing import Dict, List
  from .....None import oscar
  from .....core import ChunkGraph, Chunk
  from .....core.operand import Fuse, Fetch
  from .....metrics import Metrics
  from .....utils import get_chunk_params
  from .....typing import BandType, TileableType
  from ....meta import MetaAPI, WorkerMetaAPI
  from ....scheduling import SchedulingAPI
  from ....subtask import Subtask, SubtaskGraph, SubtaskResult, SubtaskStatus
  from ....task.core import Task, TaskResult, TaskStatus
  from ..api import ExecutionChunkResult
label_desc: Use image node_id 1 to create a new TaskStageProcessor object  with example: obj = TaskStageProcessor(stage_id, task, chunk_graph, subtask_graph, bands, tile_context, scheduling_api, meta_api)