code: def analyze(
      self,
      chunk_graph: ChunkGraph,
      chunk_to_subtasks: Dict[ChunkType, Subtask],
      available_bands: Dict[BandType, Resource],
      stage_id: str,
      op_to_bands: Dict[str, BandType] = None,
      shuffle_fetch_type: ShuffleFetchType = None,
  ) -> SubtaskGraph:
      checked_chunks = set()
      for tileable in self.tileable_graph:
          try:
              tiled = self.get_tiled(tileable)
              self._check_shuffle_reduce_chunks(
                  tiled.chunks, checked_chunks
              )
          except KeyError:
              pass
  
      # check if duplicated operand keys exist
      if self._check_duplicated_operand_keys and len(
          {c.key for c in chunk_graph}
      ) < len(
          chunk_graph
      ):  # pragma: no cover
          raise AssertionError("Duplicated operands exist")
      # record shapes generated in tile
      for n in chunk_graph:
          self._raw_chunk_shapes[n.key] = getattr(n, "shape", None)
      task = self._task
      analyzer = GraphAnalyzer(
          chunk_graph,
          available_bands,
          task,
          self._config,
          chunk_to_subtasks,
          shuffle_fetch_type=shuffle_fetch_type,
          map_reduce_id_to_infos=self.map_reduce_id_to_infos,
      )
      subtask_graph = analyzer.gen_subtask_graph()
      results = set(
          analyzer._chunk_to_copied[c]
          for c in chunk_graph.results
          if not isinstance(c.op, Fetch)
      )
      for subtask in subtask_graph:
          if subtask.extra_config is None:
              subtask.extra_config = dict()
          if all(c not in results for c in subtask.chunk_graph.results):
              subtask.extra_config["check_all"] = False
          else:
              subtask.extra_config["check_keys"] = [
                  c.key
                  for c in subtask.chunk_graph.results
                  if c in results
              ]
          proxy_chunks = [
              c
              for c in subtask.chunk_graph
              if isinstance(c.op, ShuffleProxy)
          ]
          if proxy_chunks:
              assert len(proxy_chunks) == 1, proxy_chunks
              proxy_chunk_key = proxy_chunks[0].key
              proxy_chunk = next(
                  c for c in chunk_graph if c.key == proxy_chunk_key
              )
              reducer_chunks = chunk_graph.successors(proxy_chunk)
              n_reducers_list = [
                  c.op.n_reducers for c in reducer_chunks
              ]
              n_reducers = n_reducers_list[0]
              reducer_ordinals = [
                  c.op.reducer_ordinal for c in reducer_chunks
              ]
              assert set(reducer_ordinals).issubset(
                  list(range(n_reducers))
              ), (
                  reducer_ordinals,
                  n_reducers,
              )
              assert len(set(n_reducers_list)) == 1, n_reducers_list
              mapper_chunks = chunk_graph.predecessors(proxy_chunk)
              assert proxy_chunk.op.n_mappers == len(mapper_chunks), (
                  proxy_chunk.op.n_mappers,
                  mapper_chunks,
              )
              # If some reducer data are not used by downstream, then it won't be included in the chunk graph.
              assert proxy_chunk.op.n_reducers >= n_reducers, (
                  proxy_chunk.op.n_reducers,
                  n_reducers,
              )
      return subtask_graph
  
imports: import itertools
  from collections import defaultdict
  from functools import partial
  from typing import Callable, Dict, List
  import numpy
  from .....core import TileableType, ChunkGraph, OBJECT_TYPE, enter_mode, register, unregister
  from .....core.operand import Fetch, ShuffleProxy
  from .....core.operand.shuffle import ShuffleFetchType
  from .....resource import Resource
  from .....tests.core import _check_args, ObjectCheckMixin
  from .....typing import BandType, ChunkType
  from ....subtask import Subtask, SubtaskGraph
  from ...analyzer import GraphAnalyzer
  from ..preprocessor import CancellableTiler, TaskPreprocessor
label_desc: Use image node_id 7 for calling the CheckedTaskPreprocessor obj's underlying member method code with example usage: obj.analyze(chunk_graph, chunk_to_subtasks, available_bands, stage_id, op_to_bands, shuffle_fetch_type) and returns: subtask_graph