We are currently solving the following issue within our repository. Here is the issue text:
--- BEGIN ISSUE ---
IndexError with util.apply_parallel (RGB images)
I don't really understand the origin of this error. I could only reproduce with RGB images, not with 2D arrays. Does `apply_parallel` only work for 2D images?

## Way to reproduce
```python
from skimage import util, segmentation, data
img = data.astronaut()
_ = util.apply_parallel(segmentation.slic, img, chunks=None)
```

Output
```
---------------------------------------------------------------------------
IndexError                                Traceback (most recent call last)
<ipython-input-3-63f67e7b9ccd> in <module>
----> 1 _ = util.apply_parallel(segmentation.slic, img, chunks=None)

~/code/scikit-image/skimage/util/apply_parallel.py in apply_parallel(function, array, chunks, depth, mode, extra_arguments, extra_keywords, compute)
    143     res = darr.map_overlap(wrapped_func, depth, boundary=mode)
    144     if compute:
--> 145         res = res.compute()
    146 
    147     return res

~/.local/lib/python3.7/site-packages/dask/base.py in compute(self, **kwargs)
    164         dask.base.compute
    165         """
--> 166         (result,) = compute(self, traverse=False, **kwargs)
    167         return result
    168 

~/.local/lib/python3.7/site-packages/dask/base.py in compute(*args, **kwargs)
    443 
    444     results = schedule(dsk, keys, **kwargs)
--> 445     return repack([f(r, *a) for r, (f, a) in zip(results, postcomputes)])
    446 
    447 

~/.local/lib/python3.7/site-packages/dask/base.py in <listcomp>(.0)
    443 
    444     results = schedule(dsk, keys, **kwargs)
--> 445     return repack([f(r, *a) for r, (f, a) in zip(results, postcomputes)])
    446 
    447 

~/.local/lib/python3.7/site-packages/dask/array/core.py in finalize(results)
    984     while isinstance(results2, (tuple, list)):
    985         if len(results2) > 1:
--> 986             return concatenate3(results)
    987         else:
    988             results2 = results2[0]

~/.local/lib/python3.7/site-packages/dask/array/core.py in concatenate3(arrays)
   4383     if not ndim:
   4384         return arrays
-> 4385     chunks = chunks_from_arrays(arrays)
   4386     shape = tuple(map(sum, chunks))
   4387 

~/.local/lib/python3.7/site-packages/dask/array/core.py in chunks_from_arrays(arrays)
   4152 
   4153     while isinstance(arrays, (list, tuple)):
-> 4154         result.append(tuple([shape(deepfirst(a))[dim] for a in arrays]))
   4155         arrays = arrays[0]
   4156         dim += 1

~/.local/lib/python3.7/site-packages/dask/array/core.py in <listcomp>(.0)
   4152 
   4153     while isinstance(arrays, (list, tuple)):
-> 4154         result.append(tuple([shape(deepfirst(a))[dim] for a in arrays]))
   4155         arrays = arrays[0]
   4156         dim += 1

IndexError: tuple index out of range
```

## Version information
```python
# Paste the output of the following python commands
from __future__ import print_function
import sys; print(sys.version)
import platform; print(platform.platform())
import skimage; print("scikit-image version: {}".format(skimage.__version__))
import numpy; print("numpy version: {}".format(numpy.__version__))
```

```python
# your output here
3.7.3 (default, Oct  7 2019, 12:56:13) 
[GCC 8.3.0]
Linux-5.0.0-38-generic-x86_64-with-Ubuntu-19.04-disco
scikit-image version: 0.18.dev0
numpy version: 1.16.4
```


--- END ISSUE ---

Below are some code segments, each from a relevant file. One or more of these files may contain bugs.
--- BEGIN FILES ---
Path: `skimage/util/apply_parallel.py`
Content:
```
1 __all__ = ['apply_parallel']
2 
3 
4 def _get_chunks(shape, ncpu):
5     """Split the array into equal sized chunks based on the number of
6     available processors. The last chunk in each dimension absorbs the
7     remainder array elements if the number of CPUs does not divide evenly into
8     the number of array elements.
9 
10     Examples
11     --------
12     >>> _get_chunks((4, 4), 4)
13     ((2, 2), (2, 2))
14     >>> _get_chunks((4, 4), 2)
15     ((2, 2), (4,))
16     >>> _get_chunks((5, 5), 2)
17     ((2, 3), (5,))
18     >>> _get_chunks((2, 4), 2)
19     ((1, 1), (4,))
20     """
21     # since apply_parallel is in the critical import path, we lazy import
22     # math just when we need it.
23     from math import ceil
24 
25     chunks = []
26     nchunks_per_dim = int(ceil(ncpu ** (1./len(shape))))
27 
28     used_chunks = 1
29     for i in shape:
30         if used_chunks < ncpu:
31             regular_chunk = i // nchunks_per_dim
32             remainder_chunk = regular_chunk + (i % nchunks_per_dim)
33 
34             if regular_chunk == 0:
35                 chunk_lens = (remainder_chunk,)
36             else:
37                 chunk_lens = ((regular_chunk,) * (nchunks_per_dim - 1) +
38                               (remainder_chunk,))
39         else:
40             chunk_lens = (i,)
41 
42         chunks.append(chunk_lens)
43         used_chunks *= nchunks_per_dim
44     return tuple(chunks)
45 
46 
47 def _ensure_dask_array(array, chunks=None):
48     import dask.array as da
49     if isinstance(array, da.Array):
50         return array
51 
52     return da.from_array(array, chunks=chunks)
53 
54 
55 def apply_parallel(function, array, chunks=None, depth=0, mode=None,
56                    extra_arguments=(), extra_keywords={}, *, compute=None):
57     """Map a function in parallel across an array.
58 
59     Split an array into possibly overlapping chunks of a given depth and
60     boundary type, call the given function in parallel on the chunks, combine
61     the chunks and return the resulting array.
62 
63     Parameters
64     ----------
65     function : function
66         Function to be mapped which takes an array as an argument.
67     array : numpy array or dask array
68         Array which the function will be applied to.
69     chunks : int, tuple, or tuple of tuples, optional
70         A single integer is interpreted as the length of one side of a square
71         chunk that should be tiled across the array.  One tuple of length
72         ``array.ndim`` represents the shape of a chunk, and it is tiled across
73         the array.  A list of tuples of length ``ndim``, where each sub-tuple
74         is a sequence of chunk sizes along the corresponding dimension. If
75         None, the array is broken up into chunks based on the number of
76         available cpus. More information about chunks is in the documentation
77         `here <https://dask.pydata.org/en/latest/array-design.html>`_.
78     depth : int, optional
79         Integer equal to the depth of the added boundary cells. Defaults to
80         zero.
81     mode : {'reflect', 'symmetric', 'periodic', 'wrap', 'nearest', 'edge'}, optional
82         type of external boundary padding.
83     extra_arguments : tuple, optional
84         Tuple of arguments to be passed to the function.
85     extra_keywords : dictionary, optional
86         Dictionary of keyword arguments to be passed to the function.
87     compute : bool, optional
88         If ``True``, compute eagerly returning a NumPy Array.
89         If ``False``, compute lazily returning a Dask Array.
90         If ``None`` (default), compute based on array type provided
91         (eagerly for NumPy Arrays and lazily for Dask Arrays).
92 
93     Returns
94     -------
95     out : ndarray or dask Array
96         Returns the result of the applying the operation.
97         Type is dependent on the ``compute`` argument.
98 
99     Notes
100     -----
101     Numpy edge modes 'symmetric', 'wrap', and 'edge' are converted to the
102     equivalent ``dask`` boundary modes 'reflect', 'periodic' and 'nearest',
103     respectively.
104     Setting ``compute=False`` can be useful for chaining later operations.
105     For example region selection to preview a result or storing large data
106     to disk instead of loading in memory.
107 
108     """
109     try:
110         # Importing dask takes time. since apply_parallel is on the
111         # minimum import path of skimage, we lazy attempt to import dask
112         import dask.array as da
113     except ImportError:
114         raise RuntimeError("Could not import 'dask'.  Please install "
115                            "using 'pip install dask'")
116 
117     if compute is None:
118         compute = not isinstance(array, da.Array)
119 
120     if chunks is None:
121         shape = array.shape
122         try:
123             # since apply_parallel is in the critical import path, we lazy
124             # import multiprocessing just when we need it.
125             from multiprocessing import cpu_count
126             ncpu = cpu_count()
127         except NotImplementedError:
128             ncpu = 4
129         chunks = _get_chunks(shape, ncpu)
130 
131     if mode == 'wrap':
132         mode = 'periodic'
133     elif mode == 'symmetric':
134         mode = 'reflect'
135     elif mode == 'edge':
136         mode = 'nearest'
137 
138     def wrapped_func(arr):
139         return function(arr, *extra_arguments, **extra_keywords)
140 
141     darr = _ensure_dask_array(array, chunks=chunks)
142 
143     res = darr.map_overlap(wrapped_func, depth, boundary=mode)
144     if compute:
145         res = res.compute()
146 
147     return res
148
```
--- END FILES ---

Please first localize the bug based on the issue statement, and then generate a patch according to the `git diff` format fenced by three backticks.

Here is an example:

```diff
diff --git a/examples/server_async.py b/examples/server_async.py
--- a/examples/server_async.py
+++ b/examples/server_async.py
@@ -313,4 +313,4 @@
 
 
 if __name__ == "__main__":
-    asyncio.run(run_async_server("."), debug=True)
+    asyncio.run(run_async_server(), debug=True)
diff --git a/examples/server_sync.py b/examples/server_sync.py
--- a/examples/server_sync.py
+++ b/examples/server_sync.py
@@ -313,5 +313,5 @@
 
 
 if __name__ == "__main__":
-    server = run_sync_server(".")
+    server = run_sync_server()
     server.shutdown()

```

diff --git a/skimage/util/apply_parallel.py b/skimage/util/apply_parallel.py
--- a/skimage/util/apply_parallel.py
+++ b/skimage/util/apply_parallel.py
@@ -1,3 +1,6 @@
+import numpy
+
+
 __all__ = ['apply_parallel']
 
 
@@ -53,7 +56,8 @@
 
 
 def apply_parallel(function, array, chunks=None, depth=0, mode=None,
-                   extra_arguments=(), extra_keywords={}, *, compute=None):
+                   extra_arguments=(), extra_keywords={}, *, dtype=None,
+                   multichannel=False, compute=None):
     """Map a function in parallel across an array.
 
     Split an array into possibly overlapping chunks of a given depth and
@@ -84,6 +88,25 @@
         Tuple of arguments to be passed to the function.
     extra_keywords : dictionary, optional
         Dictionary of keyword arguments to be passed to the function.
+    dtype : data-type or None, optional
+        The data-type of the `function` output. If None, Dask will attempt to
+        infer this by calling the function on data of shape ``(1,) * ndim``.
+        For functions expecting RGB or multichannel data this may be
+        problematic. In such cases, the user should manually specify this dtype
+        argument instead.
+
+        .. versionadded:: 0.18
+           ``dtype`` was added in 0.18.
+    multichannel : bool, optional
+        If `chunks` is None and `multichannel` is True, this function will keep
+        only a single chunk along the channels axis. When `depth` is specified
+        as a scalar value, that depth will be applied only to the non-channels
+        axes (a depth of 0 will be used along the channels axis). If the user
+        manually specified both `chunks` and a `depth` tuple, then this
+        argument will have no effect.
+
+        .. versionadded:: 0.18
+           ``multichannel`` was added in 0.18.
     compute : bool, optional
         If ``True``, compute eagerly returning a NumPy Array.
         If ``False``, compute lazily returning a Dask Array.
@@ -126,7 +149,10 @@
             ncpu = cpu_count()
         except NotImplementedError:
             ncpu = 4
-        chunks = _get_chunks(shape, ncpu)
+        if multichannel:
+            chunks = _get_chunks(shape[:-1], ncpu) + (shape[-1],)
+        else:
+            chunks = _get_chunks(shape, ncpu)
 
     if mode == 'wrap':
         mode = 'periodic'
@@ -135,12 +161,16 @@
     elif mode == 'edge':
         mode = 'nearest'
 
+    if multichannel and numpy.isscalar(depth):
+        # depth is only used along the non-channel axes
+        depth = (depth,) * (len(array.shape) - 1) + (0,)
+
     def wrapped_func(arr):
         return function(arr, *extra_arguments, **extra_keywords)
 
     darr = _ensure_dask_array(array, chunks=chunks)
 
-    res = darr.map_overlap(wrapped_func, depth, boundary=mode)
+    res = darr.map_overlap(wrapped_func, depth, boundary=mode, dtype=dtype)
     if compute:
         res = res.compute()

{"golden_diff": "diff --git a/skimage/util/apply_parallel.py b/skimage/util/apply_parallel.py\n--- a/skimage/util/apply_parallel.py\n+++ b/skimage/util/apply_parallel.py\n@@ -1,3 +1,6 @@\n+import numpy\n+\n+\n __all__ = ['apply_parallel']\n \n \n@@ -53,7 +56,8 @@\n \n \n def apply_parallel(function, array, chunks=None, depth=0, mode=None,\n-                   extra_arguments=(), extra_keywords={}, *, compute=None):\n+                   extra_arguments=(), extra_keywords={}, *, dtype=None,\n+                   multichannel=False, compute=None):\n     \"\"\"Map a function in parallel across an array.\n \n     Split an array into possibly overlapping chunks of a given depth and\n@@ -84,6 +88,25 @@\n         Tuple of arguments to be passed to the function.\n     extra_keywords : dictionary, optional\n         Dictionary of keyword arguments to be passed to the function.\n+    dtype : data-type or None, optional\n+        The data-type of the `function` output. If None, Dask will attempt to\n+        infer this by calling the function on data of shape ``(1,) * ndim``.\n+        For functions expecting RGB or multichannel data this may be\n+        problematic. In such cases, the user should manually specify this dtype\n+        argument instead.\n+\n+        .. versionadded:: 0.18\n+           ``dtype`` was added in 0.18.\n+    multichannel : bool, optional\n+        If `chunks` is None and `multichannel` is True, this function will keep\n+        only a single chunk along the channels axis. When `depth` is specified\n+        as a scalar value, that depth will be applied only to the non-channels\n+        axes (a depth of 0 will be used along the channels axis). If the user\n+        manually specified both `chunks` and a `depth` tuple, then this\n+        argument will have no effect.\n+\n+        .. versionadded:: 0.18\n+           ``multichannel`` was added in 0.18.\n     compute : bool, optional\n         If ``True``, compute eagerly returning a NumPy Array.\n         If ``False``, compute lazily returning a Dask Array.\n@@ -126,7 +149,10 @@\n             ncpu = cpu_count()\n         except NotImplementedError:\n             ncpu = 4\n-        chunks = _get_chunks(shape, ncpu)\n+        if multichannel:\n+            chunks = _get_chunks(shape[:-1], ncpu) + (shape[-1],)\n+        else:\n+            chunks = _get_chunks(shape, ncpu)\n \n     if mode == 'wrap':\n         mode = 'periodic'\n@@ -135,12 +161,16 @@\n     elif mode == 'edge':\n         mode = 'nearest'\n \n+    if multichannel and numpy.isscalar(depth):\n+        # depth is only used along the non-channel axes\n+        depth = (depth,) * (len(array.shape) - 1) + (0,)\n+\n     def wrapped_func(arr):\n         return function(arr, *extra_arguments, **extra_keywords)\n \n     darr = _ensure_dask_array(array, chunks=chunks)\n \n-    res = darr.map_overlap(wrapped_func, depth, boundary=mode)\n+    res = darr.map_overlap(wrapped_func, depth, boundary=mode, dtype=dtype)\n     if compute:\n         res = res.compute()\n", "issue": "IndexError with util.apply_parallel (RGB images)\nI don't really understand the origin of this error. I could only reproduce with RGB images, not with 2D arrays. Does `apply_parallel` only work for 2D images?\r\n\r\n## Way to reproduce\r\n```python\r\nfrom skimage import util, segmentation, data\r\nimg = data.astronaut()\r\n_ = util.apply_parallel(segmentation.slic, img, chunks=None)\r\n```\r\n\r\nOutput\r\n```\r\n---------------------------------------------------------------------------\r\nIndexError                                Traceback (most recent call last)\r\n<ipython-input-3-63f67e7b9ccd> in <module>\r\n----> 1 _ = util.apply_parallel(segmentation.slic, img, chunks=None)\r\n\r\n~/code/scikit-image/skimage/util/apply_parallel.py in apply_parallel(function, array, chunks, depth, mode, extra_arguments, extra_keywords, compute)\r\n    143     res = darr.map_overlap(wrapped_func, depth, boundary=mode)\r\n    144     if compute:\r\n--> 145         res = res.compute()\r\n    146 \r\n    147     return res\r\n\r\n~/.local/lib/python3.7/site-packages/dask/base.py in compute(self, **kwargs)\r\n    164         dask.base.compute\r\n    165         \"\"\"\r\n--> 166         (result,) = compute(self, traverse=False, **kwargs)\r\n    167         return result\r\n    168 \r\n\r\n~/.local/lib/python3.7/site-packages/dask/base.py in compute(*args, **kwargs)\r\n    443 \r\n    444     results = schedule(dsk, keys, **kwargs)\r\n--> 445     return repack([f(r, *a) for r, (f, a) in zip(results, postcomputes)])\r\n    446 \r\n    447 \r\n\r\n~/.local/lib/python3.7/site-packages/dask/base.py in <listcomp>(.0)\r\n    443 \r\n    444     results = schedule(dsk, keys, **kwargs)\r\n--> 445     return repack([f(r, *a) for r, (f, a) in zip(results, postcomputes)])\r\n    446 \r\n    447 \r\n\r\n~/.local/lib/python3.7/site-packages/dask/array/core.py in finalize(results)\r\n    984     while isinstance(results2, (tuple, list)):\r\n    985         if len(results2) > 1:\r\n--> 986             return concatenate3(results)\r\n    987         else:\r\n    988             results2 = results2[0]\r\n\r\n~/.local/lib/python3.7/site-packages/dask/array/core.py in concatenate3(arrays)\r\n   4383     if not ndim:\r\n   4384         return arrays\r\n-> 4385     chunks = chunks_from_arrays(arrays)\r\n   4386     shape = tuple(map(sum, chunks))\r\n   4387 \r\n\r\n~/.local/lib/python3.7/site-packages/dask/array/core.py in chunks_from_arrays(arrays)\r\n   4152 \r\n   4153     while isinstance(arrays, (list, tuple)):\r\n-> 4154         result.append(tuple([shape(deepfirst(a))[dim] for a in arrays]))\r\n   4155         arrays = arrays[0]\r\n   4156         dim += 1\r\n\r\n~/.local/lib/python3.7/site-packages/dask/array/core.py in <listcomp>(.0)\r\n   4152 \r\n   4153     while isinstance(arrays, (list, tuple)):\r\n-> 4154         result.append(tuple([shape(deepfirst(a))[dim] for a in arrays]))\r\n   4155         arrays = arrays[0]\r\n   4156         dim += 1\r\n\r\nIndexError: tuple index out of range\r\n```\r\n\r\n## Version information\r\n```python\r\n# Paste the output of the following python commands\r\nfrom __future__ import print_function\r\nimport sys; print(sys.version)\r\nimport platform; print(platform.platform())\r\nimport skimage; print(\"scikit-image version: {}\".format(skimage.__version__))\r\nimport numpy; print(\"numpy version: {}\".format(numpy.__version__))\r\n```\r\n\r\n```python\r\n# your output here\r\n3.7.3 (default, Oct  7 2019, 12:56:13) \r\n[GCC 8.3.0]\r\nLinux-5.0.0-38-generic-x86_64-with-Ubuntu-19.04-disco\r\nscikit-image version: 0.18.dev0\r\nnumpy version: 1.16.4\r\n```\r\n\n", "before_files": [{"content": "__all__ = ['apply_parallel']\n\n\ndef _get_chunks(shape, ncpu):\n    \"\"\"Split the array into equal sized chunks based on the number of\n    available processors. The last chunk in each dimension absorbs the\n    remainder array elements if the number of CPUs does not divide evenly into\n    the number of array elements.\n\n    Examples\n    --------\n    >>> _get_chunks((4, 4), 4)\n    ((2, 2), (2, 2))\n    >>> _get_chunks((4, 4), 2)\n    ((2, 2), (4,))\n    >>> _get_chunks((5, 5), 2)\n    ((2, 3), (5,))\n    >>> _get_chunks((2, 4), 2)\n    ((1, 1), (4,))\n    \"\"\"\n    # since apply_parallel is in the critical import path, we lazy import\n    # math just when we need it.\n    from math import ceil\n\n    chunks = []\n    nchunks_per_dim = int(ceil(ncpu ** (1./len(shape))))\n\n    used_chunks = 1\n    for i in shape:\n        if used_chunks < ncpu:\n            regular_chunk = i // nchunks_per_dim\n            remainder_chunk = regular_chunk + (i % nchunks_per_dim)\n\n            if regular_chunk == 0:\n                chunk_lens = (remainder_chunk,)\n            else:\n                chunk_lens = ((regular_chunk,) * (nchunks_per_dim - 1) +\n                              (remainder_chunk,))\n        else:\n            chunk_lens = (i,)\n\n        chunks.append(chunk_lens)\n        used_chunks *= nchunks_per_dim\n    return tuple(chunks)\n\n\ndef _ensure_dask_array(array, chunks=None):\n    import dask.array as da\n    if isinstance(array, da.Array):\n        return array\n\n    return da.from_array(array, chunks=chunks)\n\n\ndef apply_parallel(function, array, chunks=None, depth=0, mode=None,\n                   extra_arguments=(), extra_keywords={}, *, compute=None):\n    \"\"\"Map a function in parallel across an array.\n\n    Split an array into possibly overlapping chunks of a given depth and\n    boundary type, call the given function in parallel on the chunks, combine\n    the chunks and return the resulting array.\n\n    Parameters\n    ----------\n    function : function\n        Function to be mapped which takes an array as an argument.\n    array : numpy array or dask array\n        Array which the function will be applied to.\n    chunks : int, tuple, or tuple of tuples, optional\n        A single integer is interpreted as the length of one side of a square\n        chunk that should be tiled across the array.  One tuple of length\n        ``array.ndim`` represents the shape of a chunk, and it is tiled across\n        the array.  A list of tuples of length ``ndim``, where each sub-tuple\n        is a sequence of chunk sizes along the corresponding dimension. If\n        None, the array is broken up into chunks based on the number of\n        available cpus. More information about chunks is in the documentation\n        `here <https://dask.pydata.org/en/latest/array-design.html>`_.\n    depth : int, optional\n        Integer equal to the depth of the added boundary cells. Defaults to\n        zero.\n    mode : {'reflect', 'symmetric', 'periodic', 'wrap', 'nearest', 'edge'}, optional\n        type of external boundary padding.\n    extra_arguments : tuple, optional\n        Tuple of arguments to be passed to the function.\n    extra_keywords : dictionary, optional\n        Dictionary of keyword arguments to be passed to the function.\n    compute : bool, optional\n        If ``True``, compute eagerly returning a NumPy Array.\n        If ``False``, compute lazily returning a Dask Array.\n        If ``None`` (default), compute based on array type provided\n        (eagerly for NumPy Arrays and lazily for Dask Arrays).\n\n    Returns\n    -------\n    out : ndarray or dask Array\n        Returns the result of the applying the operation.\n        Type is dependent on the ``compute`` argument.\n\n    Notes\n    -----\n    Numpy edge modes 'symmetric', 'wrap', and 'edge' are converted to the\n    equivalent ``dask`` boundary modes 'reflect', 'periodic' and 'nearest',\n    respectively.\n    Setting ``compute=False`` can be useful for chaining later operations.\n    For example region selection to preview a result or storing large data\n    to disk instead of loading in memory.\n\n    \"\"\"\n    try:\n        # Importing dask takes time. since apply_parallel is on the\n        # minimum import path of skimage, we lazy attempt to import dask\n        import dask.array as da\n    except ImportError:\n        raise RuntimeError(\"Could not import 'dask'.  Please install \"\n                           \"using 'pip install dask'\")\n\n    if compute is None:\n        compute = not isinstance(array, da.Array)\n\n    if chunks is None:\n        shape = array.shape\n        try:\n            # since apply_parallel is in the critical import path, we lazy\n            # import multiprocessing just when we need it.\n            from multiprocessing import cpu_count\n            ncpu = cpu_count()\n        except NotImplementedError:\n            ncpu = 4\n        chunks = _get_chunks(shape, ncpu)\n\n    if mode == 'wrap':\n        mode = 'periodic'\n    elif mode == 'symmetric':\n        mode = 'reflect'\n    elif mode == 'edge':\n        mode = 'nearest'\n\n    def wrapped_func(arr):\n        return function(arr, *extra_arguments, **extra_keywords)\n\n    darr = _ensure_dask_array(array, chunks=chunks)\n\n    res = darr.map_overlap(wrapped_func, depth, boundary=mode)\n    if compute:\n        res = res.compute()\n\n    return res\n", "path": "skimage/util/apply_parallel.py"}], "after_files": [{"content": "import numpy\n\n\n__all__ = ['apply_parallel']\n\n\ndef _get_chunks(shape, ncpu):\n    \"\"\"Split the array into equal sized chunks based on the number of\n    available processors. The last chunk in each dimension absorbs the\n    remainder array elements if the number of CPUs does not divide evenly into\n    the number of array elements.\n\n    Examples\n    --------\n    >>> _get_chunks((4, 4), 4)\n    ((2, 2), (2, 2))\n    >>> _get_chunks((4, 4), 2)\n    ((2, 2), (4,))\n    >>> _get_chunks((5, 5), 2)\n    ((2, 3), (5,))\n    >>> _get_chunks((2, 4), 2)\n    ((1, 1), (4,))\n    \"\"\"\n    # since apply_parallel is in the critical import path, we lazy import\n    # math just when we need it.\n    from math import ceil\n\n    chunks = []\n    nchunks_per_dim = int(ceil(ncpu ** (1./len(shape))))\n\n    used_chunks = 1\n    for i in shape:\n        if used_chunks < ncpu:\n            regular_chunk = i // nchunks_per_dim\n            remainder_chunk = regular_chunk + (i % nchunks_per_dim)\n\n            if regular_chunk == 0:\n                chunk_lens = (remainder_chunk,)\n            else:\n                chunk_lens = ((regular_chunk,) * (nchunks_per_dim - 1) +\n                              (remainder_chunk,))\n        else:\n            chunk_lens = (i,)\n\n        chunks.append(chunk_lens)\n        used_chunks *= nchunks_per_dim\n    return tuple(chunks)\n\n\ndef _ensure_dask_array(array, chunks=None):\n    import dask.array as da\n    if isinstance(array, da.Array):\n        return array\n\n    return da.from_array(array, chunks=chunks)\n\n\ndef apply_parallel(function, array, chunks=None, depth=0, mode=None,\n                   extra_arguments=(), extra_keywords={}, *, dtype=None,\n                   multichannel=False, compute=None):\n    \"\"\"Map a function in parallel across an array.\n\n    Split an array into possibly overlapping chunks of a given depth and\n    boundary type, call the given function in parallel on the chunks, combine\n    the chunks and return the resulting array.\n\n    Parameters\n    ----------\n    function : function\n        Function to be mapped which takes an array as an argument.\n    array : numpy array or dask array\n        Array which the function will be applied to.\n    chunks : int, tuple, or tuple of tuples, optional\n        A single integer is interpreted as the length of one side of a square\n        chunk that should be tiled across the array.  One tuple of length\n        ``array.ndim`` represents the shape of a chunk, and it is tiled across\n        the array.  A list of tuples of length ``ndim``, where each sub-tuple\n        is a sequence of chunk sizes along the corresponding dimension. If\n        None, the array is broken up into chunks based on the number of\n        available cpus. More information about chunks is in the documentation\n        `here <https://dask.pydata.org/en/latest/array-design.html>`_.\n    depth : int, optional\n        Integer equal to the depth of the added boundary cells. Defaults to\n        zero.\n    mode : {'reflect', 'symmetric', 'periodic', 'wrap', 'nearest', 'edge'}, optional\n        type of external boundary padding.\n    extra_arguments : tuple, optional\n        Tuple of arguments to be passed to the function.\n    extra_keywords : dictionary, optional\n        Dictionary of keyword arguments to be passed to the function.\n    dtype : data-type or None, optional\n        The data-type of the `function` output. If None, Dask will attempt to\n        infer this by calling the function on data of shape ``(1,) * ndim``.\n        For functions expecting RGB or multichannel data this may be\n        problematic. In such cases, the user should manually specify this dtype\n        argument instead.\n\n        .. versionadded:: 0.18\n           ``dtype`` was added in 0.18.\n    multichannel : bool, optional\n        If `chunks` is None and `multichannel` is True, this function will keep\n        only a single chunk along the channels axis. When `depth` is specified\n        as a scalar value, that depth will be applied only to the non-channels\n        axes (a depth of 0 will be used along the channels axis). If the user\n        manually specified both `chunks` and a `depth` tuple, then this\n        argument will have no effect.\n\n        .. versionadded:: 0.18\n           ``multichannel`` was added in 0.18.\n    compute : bool, optional\n        If ``True``, compute eagerly returning a NumPy Array.\n        If ``False``, compute lazily returning a Dask Array.\n        If ``None`` (default), compute based on array type provided\n        (eagerly for NumPy Arrays and lazily for Dask Arrays).\n\n    Returns\n    -------\n    out : ndarray or dask Array\n        Returns the result of the applying the operation.\n        Type is dependent on the ``compute`` argument.\n\n    Notes\n    -----\n    Numpy edge modes 'symmetric', 'wrap', and 'edge' are converted to the\n    equivalent ``dask`` boundary modes 'reflect', 'periodic' and 'nearest',\n    respectively.\n    Setting ``compute=False`` can be useful for chaining later operations.\n    For example region selection to preview a result or storing large data\n    to disk instead of loading in memory.\n\n    \"\"\"\n    try:\n        # Importing dask takes time. since apply_parallel is on the\n        # minimum import path of skimage, we lazy attempt to import dask\n        import dask.array as da\n    except ImportError:\n        raise RuntimeError(\"Could not import 'dask'.  Please install \"\n                           \"using 'pip install dask'\")\n\n    if compute is None:\n        compute = not isinstance(array, da.Array)\n\n    if chunks is None:\n        shape = array.shape\n        try:\n            # since apply_parallel is in the critical import path, we lazy\n            # import multiprocessing just when we need it.\n            from multiprocessing import cpu_count\n            ncpu = cpu_count()\n        except NotImplementedError:\n            ncpu = 4\n        if multichannel:\n            chunks = _get_chunks(shape[:-1], ncpu) + (shape[-1],)\n        else:\n            chunks = _get_chunks(shape, ncpu)\n\n    if mode == 'wrap':\n        mode = 'periodic'\n    elif mode == 'symmetric':\n        mode = 'reflect'\n    elif mode == 'edge':\n        mode = 'nearest'\n\n    if multichannel and numpy.isscalar(depth):\n        # depth is only used along the non-channel axes\n        depth = (depth,) * (len(array.shape) - 1) + (0,)\n\n    def wrapped_func(arr):\n        return function(arr, *extra_arguments, **extra_keywords)\n\n    darr = _ensure_dask_array(array, chunks=chunks)\n\n    res = darr.map_overlap(wrapped_func, depth, boundary=mode, dtype=dtype)\n    if compute:\n        res = res.compute()\n\n    return res\n", "path": "skimage/util/apply_parallel.py"}]}