code: def blockwise(
      func,
      output,
      output_indices,
      *arrind_pairs,
      numblocks=None,
      concatenate=None,
      new_axes=None,
      dependencies=(),
      **kwargs,
  ):
      """Create a Blockwise symbolic mutable mapping
  
      This is like the ``make_blockwise_graph`` function, but rather than construct a
      dict, it returns a symbolic Blockwise object.
  
      ``*arrind_pairs`` is similar to those in `make_blockwise_graph`, but in addition to
      allowing for collections it can accept BlockwiseDep instances, which allows for lazy
      evaluation of arguments to ``func`` which might be different for different
      chunks/partitions.
  
      See Also
      --------
      make_blockwise_graph
      Blockwise
      """
      new_axes = new_axes or {}
  
      arrind_pairs = list(arrind_pairs)
  
      # Transform indices to canonical elements
      # We use terms like _0, and _1 rather than provided index elements
      unique_indices = {
          i for ii in arrind_pairs[1::2] if ii is not None for i in ii
      } | set(output_indices)
      sub = {
          k: blockwise_token(i, ".")
          for i, k in enumerate(sorted(unique_indices))
      }
      output_indices = index_subs(tuple(output_indices), sub)
      a_pairs_list = []
      for a in arrind_pairs[1::2]:
          if a is not None:
              val = tuple(a)
          else:
              val = a
          a_pairs_list.append(index_subs(val, sub))
  
      arrind_pairs[1::2] = a_pairs_list
      new_axes = {
          index_subs((k,), sub)[0]: v for k, v in new_axes.items()
      }
  
      # Unpack dask values in non-array arguments
      inputs = []
      inputs_indices = []
      for name, index in toolz.partition(2, arrind_pairs):
          inputs.append(name)
          inputs_indices.append(index)
  
      # Unpack delayed objects in kwargs
      new_keys = {n for c in dependencies for n in c.__dask_layers__()}
      if kwargs:
          # replace keys in kwargs with _0 tokens
          new_tokens = tuple(
              blockwise_token(i)
              for i in range(len(inputs), len(inputs) + len(new_keys))
          )
          sub = dict(zip(new_keys, new_tokens))
          inputs.extend(new_keys)
          inputs_indices.extend((None,) * len(new_keys))
          kwargs = subs(kwargs, sub)
  
      indices = [(k, v) for k, v in zip(inputs, inputs_indices)]
      keys = map(blockwise_token, range(len(inputs)))
  
      # Construct local graph
      if not kwargs:
          subgraph = {output: (func,) + tuple(keys)}
      else:
          _keys = list(keys)
          if new_keys:
              _keys = _keys[: -len(new_keys)]
          kwargs2 = (dict, list(map(list, kwargs.items())))
          subgraph = {output: (apply, func, _keys, kwargs2)}
  
      # Construct final output
      subgraph = Blockwise(
          output,
          output_indices,
          subgraph,
          indices,
          numblocks=numblocks,
          concatenate=concatenate,
          new_axes=new_axes,
      )
      return subgraph
  
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 4 for calling a global function with example usage: blockwise(func, output, output_indices) and returns: subgraph