code: def rewrite_blockwise(inputs):
      """Rewrite a stack of Blockwise expressions into a single blockwise expression
  
      Given a set of Blockwise layers, combine them into a single layer.  The provided
      layers are expected to fit well together.  That job is handled by
      ``optimize_blockwise``
  
      Parameters
      ----------
      inputs : list[Blockwise]
  
      Returns
      -------
      blockwise: Blockwise
  
      See Also
      --------
      optimize_blockwise
      """
      if len(inputs) == 1:
          # Fast path: if there's only one input we can just use it as-is.
          return inputs[0]
  
      fused_annotations = _fuse_annotations(
          *[i.annotations for i in inputs if i.annotations]
      )
      inputs = {inp.output: inp for inp in inputs}
      dependencies = {
          inp.output: {
              d for d, v in inp.indices if v is not None and d in inputs
          }
          for inp in inputs.values()
      }
      dependents = reverse_dict(dependencies)
  
      new_index_iter = (
          c + (str(d) if d else "")  # A, B, ... A1, B1, ...
          for d in itertools.count()
          for c in "ABCDEFGHIJKLMNOPQRSTUVWXYZ"
      )
  
      [root] = [k for k, v in dependents.items() if not v]
  
      # Our final results.  These will change during fusion below
      indices = list(inputs[root].indices)
      new_axes = inputs[root].new_axes
      concatenate = inputs[root].concatenate
      dsk = dict(inputs[root].dsk)
  
      changed = True
      while changed:
          changed = False
          for i, (dep, ind) in enumerate(indices):
              if ind is None:
                  continue
              if dep not in inputs:
                  continue
  
              changed = True
  
              # Change dep name to avoid fusing the same dep
              # (in different iteration orders) into a single
              # subgraph key/dependency
              # (see: https://github.com/dask/dask/issues/8535)
              local_dep = dep if dep == root else _unique_dep(dep, ind)
  
              # Replace _n with dep name in existing tasks
              # (inc, _0) -> (inc, 'b')
              dsk = {
                  k: subs(v, {blockwise_token(i): local_dep})
                  for k, v in dsk.items()
              }
  
              # Remove current input from input indices
              # [('a', 'i'), ('b', 'i')] -> [('a', 'i')]
              _, current_dep_indices = indices.pop(i)
              sub = {
                  blockwise_token(i): blockwise_token(i - 1)
                  for i in range(i + 1, len(indices) + 1)
              }
              dsk = subs(dsk, sub)
  
              # Change new input_indices to match give index from current computation
              # [('c', j')] -> [('c', 'i')]
              new_indices = inputs[dep].indices
              sub = dict(
                  zip(inputs[dep].output_indices, current_dep_indices)
              )
              contracted = {
                  x
                  for _, j in new_indices
                  if j is not None
                  for x in j
                  if x not in inputs[dep].output_indices
              }
              extra = dict(zip(contracted, new_index_iter))
              sub.update(extra)
              new_indices = [
                  (x, index_subs(j, sub)) for x, j in new_indices
              ]
  
              # Update new_axes
              for k, v in inputs[dep].new_axes.items():
                  new_axes[sub[k]] = v
  
              # Bump new inputs up in list
              sub = {}
              # Map from (id(key), inds or None) -> index in indices. Used to deduplicate indices.
              index_map = {
                  (id(k), inds): n
                  for n, (k, inds) in enumerate(indices)
              }
              for ii, index in enumerate(new_indices):
                  id_key = (id(index[0]), index[1])
                  if id_key in index_map:  # use old inputs if available
                      sub[blockwise_token(ii)] = blockwise_token(
                          index_map[id_key]
                      )
                  else:
                      index_map[id_key] = len(indices)
                      sub[blockwise_token(ii)] = blockwise_token(
                          len(indices)
                      )
                      indices.append(index)
              new_dsk = subs(inputs[dep].dsk, sub)
  
              # Change new_dsk key to match local_dep
              if dep != local_dep and dep in new_dsk:
                  new_dsk[local_dep] = new_dsk.pop(dep)
  
              # indices.extend(new_indices)
              dsk.update(new_dsk)
  
      # De-duplicate indices like [(a, ij), (b, i), (a, ij)] -> [(a, ij), (b, i)]
      # Make sure that we map everything else appropriately as we remove inputs
      new_indices = []
      seen = {}
      sub = {}  # like {_0: _0, _1: _0, _2: _1}
      for i, x in enumerate(indices):
          if x[1] is not None and x in seen:
              sub[i] = seen[x]
          else:
              if x[1] is not None:
                  seen[x] = len(new_indices)
              sub[i] = len(new_indices)
              new_indices.append(x)
  
      sub = {
          blockwise_token(k): blockwise_token(v) for k, v in sub.items()
      }
      dsk = {
          k: subs(v, sub) for k, v in dsk.items() if k not in sub.keys()
      }
  
      indices_check = {k for k, v in indices if v is not None}
      numblocks = toolz.merge(
          [inp.numblocks for inp in inputs.values()]
      )
      numblocks = {
          k: v
          for k, v in numblocks.items()
          if v is None or k in indices_check
      }
  
      # Update IO-dependency information
      io_deps = {}
      for v in inputs.values():
          io_deps.update(v.io_deps)
  
      return Blockwise(
          root,
          inputs[root].output_indices,
          dsk,
          new_indices,
          numblocks=numblocks,
          new_axes=new_axes,
          concatenate=concatenate,
          annotations=fused_annotations,
          io_deps=io_deps,
      )
  
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 14 for calling a global function with example usage: rewrite_blockwise(inputs) and returns: Blockwise, inputs