code: def test_blockwise_cull(flat):
      da = pytest.importorskip("dask.array")
      np = pytest.importorskip("numpy")
      if flat:
          # Simple "flat" mapping between input and
          # outut indices
          x = (
              da.from_array(np.arange(40).reshape((4, 10)), (2, 4))
              + 100
          )
      else:
          # Complex mapping between input and output
          # indices (outer product and transpose)
          x = da.from_array(np.arange(10).reshape((10,)), (4,))
          y = da.from_array(np.arange(10).reshape((10,)), (4,))
          x = da.outer(x, y).transpose()
  
      # Check that blockwise culling results in correct
      # output keys and that full graph is not materialized
      dsk = x.__dask_graph__()
      select = (1, 1)  # Select a single chunk
      keys = {(x._name, *select)}
      dsk_cull = dsk.cull(keys)
      for name, layer in dsk_cull.layers.items():
          if not isinstance(layer, dask.blockwise.Blockwise):
              # The original layer shouldn't be Blockwise if the new one isn't
              assert not isinstance(
                  dsk.layers[name], dask.blockwise.Blockwise
              )
              continue
          assert isinstance(dsk.layers[name], dask.blockwise.Blockwise)
          assert not layer.is_materialized()
          out_keys = layer.get_output_keys()
          assert out_keys == {(layer.output, *select)}
          assert not layer.is_materialized()
  
imports: from __future__ import annotations
  import os
  import threading
  import xml.etree.ElementTree
  from collections.abc import Set
  from concurrent.futures import ThreadPoolExecutor
  import pytest
  import dask
  from dask.blockwise import Blockwise, blockwise_token
  from dask.highlevelgraph import HighLevelGraph, Layer, MaterializedLayer, to_graphviz
  from dask.utils_test import inc
label_desc: Use image node_id 15 for calling a global function with example usage: test_blockwise_cull(flat) without return types