code: def optimize_blockwise(graph, keys=()):
      """High level optimization of stacked Blockwise layers
  
      For operations that have multiple Blockwise operations one after the other, like
      ``x.T + 123`` we can fuse these into a single Blockwise operation.  This happens
      before any actual tasks are generated, and so can reduce overhead.
  
      This finds groups of Blockwise operations that can be safely fused, and then
      passes them to ``rewrite_blockwise`` for rewriting.
  
      Parameters
      ----------
      graph : HighLevelGraph
      keys : Iterable
          The keys of all outputs of all collections.
          Used to make sure that we don't fuse a layer needed by an output
  
      Returns
      -------
      HighLevelGraph
  
      See Also
      --------
      rewrite_blockwise
      """
      out = _optimize_blockwise(graph, keys=keys)
      while out.dependencies != graph.dependencies:
          graph = out
          out = _optimize_blockwise(graph, keys=keys)
      return out
  
imports: from __future__ import annotations
  import itertools
  import os
  from collections.abc import Hashable, Iterable, Mapping, Sequence
  from itertools import product
  from math import prod
  from typing import Any
  import tlz
  import dask
  from dask.base import clone_key, get_name_from_key, tokenize
  from dask.core import flatten, ishashable, keys_in_tasks, reverse_dict
  from dask.highlevelgraph import HighLevelGraph, Layer
  from dask.optimization import SubgraphCallable, fuse
  from dask.typing import Graph, Key
  from dask.utils import _deprecated, apply, ensure_dict, homogeneous_deepmap, stringify_collection_keys
label_desc: Use image node_id 9 for calling a global function with example usage: optimize_blockwise(graph, keys) and returns: out