We are currently solving the following issue within our repository. Here is the issue text:
--- BEGIN ISSUE ---
dask.array regression involving arithmetic with xarray objects
This code worked with dask v0.7.6, but failed when I updated to v0.9.0:

```
In [1]: import numpy as np

In [2]: import xarray as xr

In [3]: import dask.array as da

In [4]: x = xr.core.indexing.LazilyIndexedArray(np.zeros((3, 3)))

In [5]: y = da.from_array(x, (2, 2))

In [6]: (y[:2, :2] * y[:2, :2]).compute()
---------------------------------------------------------------------------
TypeError                                 Traceback (most recent call last)
<ipython-input-6-82666a5a7d29> in <module>()
----> 1 (y[:2, :2] * y[:2, :2]).compute()

/Users/shoyer/conda/envs/xarray-dev/lib/python3.5/site-packages/dask/base.py in compute(self, **kwargs)
     35
     36     def compute(self, **kwargs):
---> 37         return compute(self, **kwargs)[0]
     38
     39     @classmethod

/Users/shoyer/conda/envs/xarray-dev/lib/python3.5/site-packages/dask/base.py in compute(*args, **kwargs)
    108                 for opt, val in groups.items()])
    109     keys = [var._keys() for var in variables]
--> 110     results = get(dsk, keys, **kwargs)
    111
    112     results_iter = iter(results)

/Users/shoyer/conda/envs/xarray-dev/lib/python3.5/site-packages/dask/threaded.py in get(dsk, result, cache, num_workers, **kwargs)
     55     results = get_async(pool.apply_async, len(pool._pool), dsk, result,
     56                         cache=cache, queue=queue, get_id=_thread_get_id,
---> 57                         **kwargs)
     58
     59     return results

/Users/shoyer/conda/envs/xarray-dev/lib/python3.5/site-packages/dask/async.py in get_async(apply_async, num_workers, dsk, result, cache, queue, get_id, raise_on_exception, rerun_exceptions_locally, callbacks, **kwargs)
    486                 _execute_task(task, data)  # Re-execute locally
    487             else:
--> 488                 raise(remote_exception(res, tb))
    489         state['cache'][key] = res
    490         finish_task(dsk, key, state, results, keyorder.get)

TypeError: unsupported operand type(s) for *: 'LazilyIndexedArray' and 'LazilyIndexedArray'

Traceback
---------
  File "/Users/shoyer/conda/envs/xarray-dev/lib/python3.5/site-packages/dask/async.py", line 267, in execute_task
    result = _execute_task(task, data)
  File "/Users/shoyer/conda/envs/xarray-dev/lib/python3.5/site-packages/dask/async.py", line 249, in _execute_task
    return func(*args2)
```

LazilyIndexedArray is an object that satisfies some of the usual duck-array API (e.g. shape, dtype, `__getitem__`) and that can be coerced into a NumPy array:
https://github.com/pydata/xarray/blob/v0.7.2/xarray/core/indexing.py#L272

I _think_ it should be valid into to `da.from_array` -- this certainly worked in old versions of dask.


--- END ISSUE ---

Below are some code segments, each from a relevant file. One or more of these files may contain bugs.
--- BEGIN FILES ---
Path: `dask/array/optimization.py`
Content:
```
1 from __future__ import absolute_import, division, print_function
2 
3 from operator import getitem
4 
5 import numpy as np
6 from toolz import valmap, partial
7 
8 from .core import getarray
9 from ..core import flatten
10 from ..optimize import cull, fuse, dealias, inline_functions
11 from ..rewrite import RuleSet, RewriteRule
12 
13 
14 def optimize(dsk, keys, **kwargs):
15     """ Optimize dask for array computation
16 
17     1.  Cull tasks not necessary to evaluate keys
18     2.  Remove full slicing, e.g. x[:]
19     3.  Inline fast functions like getitem and np.transpose
20     """
21     keys = list(flatten(keys))
22     fast_functions = kwargs.get('fast_functions',
23                              set([getarray, np.transpose]))
24     dsk2, dependencies = cull(dsk, keys)
25     dsk4, dependencies = fuse(dsk2, keys, dependencies)
26     dsk5 = optimize_slices(dsk4)
27     dsk6 = inline_functions(dsk5, keys, fast_functions=fast_functions,
28             dependencies=dependencies)
29     return dsk6
30 
31 
32 def optimize_slices(dsk):
33     """ Optimize slices
34 
35     1.  Fuse repeated slices, like x[5:][2:6] -> x[7:11]
36     2.  Remove full slices, like         x[:] -> x
37 
38     See also:
39         fuse_slice_dict
40     """
41     dsk = dsk.copy()
42     for k, v in dsk.items():
43         if type(v) is tuple:
44             if v[0] is getitem or v[0] is getarray:
45                 try:
46                     func, a, a_index = v
47                 except ValueError:  # has four elements, includes a lock
48                     continue
49                 while type(a) is tuple and (a[0] is getitem or a[0] is getarray):
50                     try:
51                         _, b, b_index = a
52                     except ValueError:  # has four elements, includes a lock
53                         break
54                     if (type(a_index) is tuple) != (type(b_index) is tuple):
55                         break
56                     if ((type(a_index) is tuple) and
57                         (len(a_index) != len(b_index)) and
58                         any(i is None for i in b_index + a_index)):
59                         break
60                     try:
61                         c_index = fuse_slice(b_index, a_index)
62                     except NotImplementedError:
63                         break
64                     (a, a_index) = (b, c_index)
65                 if (type(a_index) is slice and
66                     not a_index.start and
67                     a_index.stop is None and
68                     a_index.step is None):
69                     dsk[k] = a
70                 elif type(a_index) is tuple and all(type(s) is slice and
71                                                     not s.start and
72                                                     s.stop is None and
73                                                     s.step is None
74                                                     for s in a_index):
75                     dsk[k] = a
76                 else:
77                     dsk[k] = (func, a, a_index)
78     return dsk
79 
80 
81 def normalize_slice(s):
82     """ Replace Nones in slices with integers
83 
84     >>> normalize_slice(slice(None, None, None))
85     slice(0, None, 1)
86     """
87     start, stop, step = s.start, s.stop, s.step
88     if start is None:
89         start = 0
90     if step is None:
91         step = 1
92     if start < 0 or step < 0 or stop is not None and stop < 0:
93         raise NotImplementedError()
94     return slice(start, stop, step)
95 
96 
97 def fuse_slice(a, b):
98     """ Fuse stacked slices together
99 
100     Fuse a pair of repeated slices into a single slice:
101 
102     >>> fuse_slice(slice(1000, 2000), slice(10, 15))
103     slice(1010, 1015, None)
104 
105     This also works for tuples of slices
106 
107     >>> fuse_slice((slice(100, 200), slice(100, 200, 10)),
108     ...            (slice(10, 15), [5, 2]))
109     (slice(110, 115, None), [150, 120])
110 
111     And a variety of other interesting cases
112 
113     >>> fuse_slice(slice(1000, 2000), 10)  # integers
114     1010
115 
116     >>> fuse_slice(slice(1000, 2000, 5), slice(10, 20, 2))
117     slice(1050, 1100, 10)
118 
119     >>> fuse_slice(slice(1000, 2000, 5), [1, 2, 3])  # lists
120     [1005, 1010, 1015]
121 
122     >>> fuse_slice(None, slice(None, None))  # doctest: +SKIP
123     None
124     """
125     # None only works if the second side is a full slice
126     if a is None and b == slice(None, None):
127         return None
128 
129     # Replace None with 0 and one in start and step
130     if isinstance(a, slice):
131         a = normalize_slice(a)
132     if isinstance(b, slice):
133         b = normalize_slice(b)
134 
135     if isinstance(a, slice) and isinstance(b, int):
136         if b < 0:
137             raise NotImplementedError()
138         return a.start + b*a.step
139 
140     if isinstance(a, slice) and isinstance(b, slice):
141         start = a.start + a.step * b.start
142         if b.stop is not None:
143             stop = a.start + a.step * b.stop
144         else:
145             stop = None
146         if a.stop is not None:
147             if stop is not None:
148                 stop = min(a.stop, stop)
149             else:
150                 stop = a.stop
151             stop = stop
152         step = a.step * b.step
153         if step == 1:
154             step = None
155         return slice(start, stop, step)
156 
157     if isinstance(b, list):
158         return [fuse_slice(a, bb) for bb in b]
159     if isinstance(a, list) and isinstance(b, (int, slice)):
160         return a[b]
161 
162     if isinstance(a, tuple) and not isinstance(b, tuple):
163         b = (b,)
164 
165     # If given two tuples walk through both, being mindful of uneven sizes
166     # and newaxes
167     if isinstance(a, tuple) and isinstance(b, tuple):
168 
169         if (any(isinstance(item, list) for item in a) and
170             any(isinstance(item, list) for item in b)):
171             raise NotImplementedError("Can't handle multiple list indexing")
172 
173         j = 0
174         result = list()
175         for i in range(len(a)):
176             #  axis ceased to exist  or we're out of b
177             if isinstance(a[i], int) or j == len(b):
178                 result.append(a[i])
179                 continue
180             while b[j] is None:  # insert any Nones on the rhs
181                 result.append(None)
182                 j += 1
183             result.append(fuse_slice(a[i], b[j]))  # Common case
184             j += 1
185         while j < len(b):  # anything leftover on the right?
186             result.append(b[j])
187             j += 1
188         return tuple(result)
189     raise NotImplementedError()
190
```
--- 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/dask/array/optimization.py b/dask/array/optimization.py
--- a/dask/array/optimization.py
+++ b/dask/array/optimization.py
@@ -3,12 +3,10 @@
 from operator import getitem
 
 import numpy as np
-from toolz import valmap, partial
 
 from .core import getarray
 from ..core import flatten
-from ..optimize import cull, fuse, dealias, inline_functions
-from ..rewrite import RuleSet, RewriteRule
+from ..optimize import cull, fuse, inline_functions
 
 
 def optimize(dsk, keys, **kwargs):
@@ -20,12 +18,12 @@
     """
     keys = list(flatten(keys))
     fast_functions = kwargs.get('fast_functions',
-                             set([getarray, np.transpose]))
+                                set([getarray, np.transpose]))
     dsk2, dependencies = cull(dsk, keys)
     dsk4, dependencies = fuse(dsk2, keys, dependencies)
     dsk5 = optimize_slices(dsk4)
     dsk6 = inline_functions(dsk5, keys, fast_functions=fast_functions,
-            dependencies=dependencies)
+                            dependencies=dependencies)
     return dsk6
 
 
@@ -38,34 +36,39 @@
     See also:
         fuse_slice_dict
     """
+    getters = (getarray, getitem)
     dsk = dsk.copy()
     for k, v in dsk.items():
         if type(v) is tuple:
-            if v[0] is getitem or v[0] is getarray:
+            if v[0] in getters:
                 try:
                     func, a, a_index = v
+                    use_getarray = func is getarray
                 except ValueError:  # has four elements, includes a lock
                     continue
-                while type(a) is tuple and (a[0] is getitem or a[0] is getarray):
+                while type(a) is tuple and a[0] in getters:
                     try:
-                        _, b, b_index = a
+                        f2, b, b_index = a
+                        use_getarray |= f2 is getarray
                     except ValueError:  # has four elements, includes a lock
                         break
                     if (type(a_index) is tuple) != (type(b_index) is tuple):
                         break
                     if ((type(a_index) is tuple) and
-                        (len(a_index) != len(b_index)) and
-                        any(i is None for i in b_index + a_index)):
+                            (len(a_index) != len(b_index)) and
+                            any(i is None for i in b_index + a_index)):
                         break
                     try:
                         c_index = fuse_slice(b_index, a_index)
                     except NotImplementedError:
                         break
                     (a, a_index) = (b, c_index)
-                if (type(a_index) is slice and
-                    not a_index.start and
-                    a_index.stop is None and
-                    a_index.step is None):
+                if use_getarray:
+                    dsk[k] = (getarray, a, a_index)
+                elif (type(a_index) is slice and
+                      not a_index.start and
+                      a_index.stop is None and
+                      a_index.step is None):
                     dsk[k] = a
                 elif type(a_index) is tuple and all(type(s) is slice and
                                                     not s.start and
@@ -74,7 +77,7 @@
                                                     for s in a_index):
                     dsk[k] = a
                 else:
-                    dsk[k] = (func, a, a_index)
+                    dsk[k] = (getitem, a, a_index)
     return dsk
 
 
@@ -167,7 +170,7 @@
     if isinstance(a, tuple) and isinstance(b, tuple):
 
         if (any(isinstance(item, list) for item in a) and
-            any(isinstance(item, list) for item in b)):
+                any(isinstance(item, list) for item in b)):
             raise NotImplementedError("Can't handle multiple list indexing")
 
         j = 0

{"golden_diff": "diff --git a/dask/array/optimization.py b/dask/array/optimization.py\n--- a/dask/array/optimization.py\n+++ b/dask/array/optimization.py\n@@ -3,12 +3,10 @@\n from operator import getitem\n \n import numpy as np\n-from toolz import valmap, partial\n \n from .core import getarray\n from ..core import flatten\n-from ..optimize import cull, fuse, dealias, inline_functions\n-from ..rewrite import RuleSet, RewriteRule\n+from ..optimize import cull, fuse, inline_functions\n \n \n def optimize(dsk, keys, **kwargs):\n@@ -20,12 +18,12 @@\n     \"\"\"\n     keys = list(flatten(keys))\n     fast_functions = kwargs.get('fast_functions',\n-                             set([getarray, np.transpose]))\n+                                set([getarray, np.transpose]))\n     dsk2, dependencies = cull(dsk, keys)\n     dsk4, dependencies = fuse(dsk2, keys, dependencies)\n     dsk5 = optimize_slices(dsk4)\n     dsk6 = inline_functions(dsk5, keys, fast_functions=fast_functions,\n-            dependencies=dependencies)\n+                            dependencies=dependencies)\n     return dsk6\n \n \n@@ -38,34 +36,39 @@\n     See also:\n         fuse_slice_dict\n     \"\"\"\n+    getters = (getarray, getitem)\n     dsk = dsk.copy()\n     for k, v in dsk.items():\n         if type(v) is tuple:\n-            if v[0] is getitem or v[0] is getarray:\n+            if v[0] in getters:\n                 try:\n                     func, a, a_index = v\n+                    use_getarray = func is getarray\n                 except ValueError:  # has four elements, includes a lock\n                     continue\n-                while type(a) is tuple and (a[0] is getitem or a[0] is getarray):\n+                while type(a) is tuple and a[0] in getters:\n                     try:\n-                        _, b, b_index = a\n+                        f2, b, b_index = a\n+                        use_getarray |= f2 is getarray\n                     except ValueError:  # has four elements, includes a lock\n                         break\n                     if (type(a_index) is tuple) != (type(b_index) is tuple):\n                         break\n                     if ((type(a_index) is tuple) and\n-                        (len(a_index) != len(b_index)) and\n-                        any(i is None for i in b_index + a_index)):\n+                            (len(a_index) != len(b_index)) and\n+                            any(i is None for i in b_index + a_index)):\n                         break\n                     try:\n                         c_index = fuse_slice(b_index, a_index)\n                     except NotImplementedError:\n                         break\n                     (a, a_index) = (b, c_index)\n-                if (type(a_index) is slice and\n-                    not a_index.start and\n-                    a_index.stop is None and\n-                    a_index.step is None):\n+                if use_getarray:\n+                    dsk[k] = (getarray, a, a_index)\n+                elif (type(a_index) is slice and\n+                      not a_index.start and\n+                      a_index.stop is None and\n+                      a_index.step is None):\n                     dsk[k] = a\n                 elif type(a_index) is tuple and all(type(s) is slice and\n                                                     not s.start and\n@@ -74,7 +77,7 @@\n                                                     for s in a_index):\n                     dsk[k] = a\n                 else:\n-                    dsk[k] = (func, a, a_index)\n+                    dsk[k] = (getitem, a, a_index)\n     return dsk\n \n \n@@ -167,7 +170,7 @@\n     if isinstance(a, tuple) and isinstance(b, tuple):\n \n         if (any(isinstance(item, list) for item in a) and\n-            any(isinstance(item, list) for item in b)):\n+                any(isinstance(item, list) for item in b)):\n             raise NotImplementedError(\"Can't handle multiple list indexing\")\n \n         j = 0\n", "issue": "dask.array regression involving arithmetic with xarray objects\nThis code worked with dask v0.7.6, but failed when I updated to v0.9.0:\n\n```\nIn [1]: import numpy as np\n\nIn [2]: import xarray as xr\n\nIn [3]: import dask.array as da\n\nIn [4]: x = xr.core.indexing.LazilyIndexedArray(np.zeros((3, 3)))\n\nIn [5]: y = da.from_array(x, (2, 2))\n\nIn [6]: (y[:2, :2] * y[:2, :2]).compute()\n---------------------------------------------------------------------------\nTypeError                                 Traceback (most recent call last)\n<ipython-input-6-82666a5a7d29> in <module>()\n----> 1 (y[:2, :2] * y[:2, :2]).compute()\n\n/Users/shoyer/conda/envs/xarray-dev/lib/python3.5/site-packages/dask/base.py in compute(self, **kwargs)\n     35\n     36     def compute(self, **kwargs):\n---> 37         return compute(self, **kwargs)[0]\n     38\n     39     @classmethod\n\n/Users/shoyer/conda/envs/xarray-dev/lib/python3.5/site-packages/dask/base.py in compute(*args, **kwargs)\n    108                 for opt, val in groups.items()])\n    109     keys = [var._keys() for var in variables]\n--> 110     results = get(dsk, keys, **kwargs)\n    111\n    112     results_iter = iter(results)\n\n/Users/shoyer/conda/envs/xarray-dev/lib/python3.5/site-packages/dask/threaded.py in get(dsk, result, cache, num_workers, **kwargs)\n     55     results = get_async(pool.apply_async, len(pool._pool), dsk, result,\n     56                         cache=cache, queue=queue, get_id=_thread_get_id,\n---> 57                         **kwargs)\n     58\n     59     return results\n\n/Users/shoyer/conda/envs/xarray-dev/lib/python3.5/site-packages/dask/async.py in get_async(apply_async, num_workers, dsk, result, cache, queue, get_id, raise_on_exception, rerun_exceptions_locally, callbacks, **kwargs)\n    486                 _execute_task(task, data)  # Re-execute locally\n    487             else:\n--> 488                 raise(remote_exception(res, tb))\n    489         state['cache'][key] = res\n    490         finish_task(dsk, key, state, results, keyorder.get)\n\nTypeError: unsupported operand type(s) for *: 'LazilyIndexedArray' and 'LazilyIndexedArray'\n\nTraceback\n---------\n  File \"/Users/shoyer/conda/envs/xarray-dev/lib/python3.5/site-packages/dask/async.py\", line 267, in execute_task\n    result = _execute_task(task, data)\n  File \"/Users/shoyer/conda/envs/xarray-dev/lib/python3.5/site-packages/dask/async.py\", line 249, in _execute_task\n    return func(*args2)\n```\n\nLazilyIndexedArray is an object that satisfies some of the usual duck-array API (e.g. shape, dtype, `__getitem__`) and that can be coerced into a NumPy array:\nhttps://github.com/pydata/xarray/blob/v0.7.2/xarray/core/indexing.py#L272\n\nI _think_ it should be valid into to `da.from_array` -- this certainly worked in old versions of dask.\n\n", "before_files": [{"content": "from __future__ import absolute_import, division, print_function\n\nfrom operator import getitem\n\nimport numpy as np\nfrom toolz import valmap, partial\n\nfrom .core import getarray\nfrom ..core import flatten\nfrom ..optimize import cull, fuse, dealias, inline_functions\nfrom ..rewrite import RuleSet, RewriteRule\n\n\ndef optimize(dsk, keys, **kwargs):\n    \"\"\" Optimize dask for array computation\n\n    1.  Cull tasks not necessary to evaluate keys\n    2.  Remove full slicing, e.g. x[:]\n    3.  Inline fast functions like getitem and np.transpose\n    \"\"\"\n    keys = list(flatten(keys))\n    fast_functions = kwargs.get('fast_functions',\n                             set([getarray, np.transpose]))\n    dsk2, dependencies = cull(dsk, keys)\n    dsk4, dependencies = fuse(dsk2, keys, dependencies)\n    dsk5 = optimize_slices(dsk4)\n    dsk6 = inline_functions(dsk5, keys, fast_functions=fast_functions,\n            dependencies=dependencies)\n    return dsk6\n\n\ndef optimize_slices(dsk):\n    \"\"\" Optimize slices\n\n    1.  Fuse repeated slices, like x[5:][2:6] -> x[7:11]\n    2.  Remove full slices, like         x[:] -> x\n\n    See also:\n        fuse_slice_dict\n    \"\"\"\n    dsk = dsk.copy()\n    for k, v in dsk.items():\n        if type(v) is tuple:\n            if v[0] is getitem or v[0] is getarray:\n                try:\n                    func, a, a_index = v\n                except ValueError:  # has four elements, includes a lock\n                    continue\n                while type(a) is tuple and (a[0] is getitem or a[0] is getarray):\n                    try:\n                        _, b, b_index = a\n                    except ValueError:  # has four elements, includes a lock\n                        break\n                    if (type(a_index) is tuple) != (type(b_index) is tuple):\n                        break\n                    if ((type(a_index) is tuple) and\n                        (len(a_index) != len(b_index)) and\n                        any(i is None for i in b_index + a_index)):\n                        break\n                    try:\n                        c_index = fuse_slice(b_index, a_index)\n                    except NotImplementedError:\n                        break\n                    (a, a_index) = (b, c_index)\n                if (type(a_index) is slice and\n                    not a_index.start and\n                    a_index.stop is None and\n                    a_index.step is None):\n                    dsk[k] = a\n                elif type(a_index) is tuple and all(type(s) is slice and\n                                                    not s.start and\n                                                    s.stop is None and\n                                                    s.step is None\n                                                    for s in a_index):\n                    dsk[k] = a\n                else:\n                    dsk[k] = (func, a, a_index)\n    return dsk\n\n\ndef normalize_slice(s):\n    \"\"\" Replace Nones in slices with integers\n\n    >>> normalize_slice(slice(None, None, None))\n    slice(0, None, 1)\n    \"\"\"\n    start, stop, step = s.start, s.stop, s.step\n    if start is None:\n        start = 0\n    if step is None:\n        step = 1\n    if start < 0 or step < 0 or stop is not None and stop < 0:\n        raise NotImplementedError()\n    return slice(start, stop, step)\n\n\ndef fuse_slice(a, b):\n    \"\"\" Fuse stacked slices together\n\n    Fuse a pair of repeated slices into a single slice:\n\n    >>> fuse_slice(slice(1000, 2000), slice(10, 15))\n    slice(1010, 1015, None)\n\n    This also works for tuples of slices\n\n    >>> fuse_slice((slice(100, 200), slice(100, 200, 10)),\n    ...            (slice(10, 15), [5, 2]))\n    (slice(110, 115, None), [150, 120])\n\n    And a variety of other interesting cases\n\n    >>> fuse_slice(slice(1000, 2000), 10)  # integers\n    1010\n\n    >>> fuse_slice(slice(1000, 2000, 5), slice(10, 20, 2))\n    slice(1050, 1100, 10)\n\n    >>> fuse_slice(slice(1000, 2000, 5), [1, 2, 3])  # lists\n    [1005, 1010, 1015]\n\n    >>> fuse_slice(None, slice(None, None))  # doctest: +SKIP\n    None\n    \"\"\"\n    # None only works if the second side is a full slice\n    if a is None and b == slice(None, None):\n        return None\n\n    # Replace None with 0 and one in start and step\n    if isinstance(a, slice):\n        a = normalize_slice(a)\n    if isinstance(b, slice):\n        b = normalize_slice(b)\n\n    if isinstance(a, slice) and isinstance(b, int):\n        if b < 0:\n            raise NotImplementedError()\n        return a.start + b*a.step\n\n    if isinstance(a, slice) and isinstance(b, slice):\n        start = a.start + a.step * b.start\n        if b.stop is not None:\n            stop = a.start + a.step * b.stop\n        else:\n            stop = None\n        if a.stop is not None:\n            if stop is not None:\n                stop = min(a.stop, stop)\n            else:\n                stop = a.stop\n            stop = stop\n        step = a.step * b.step\n        if step == 1:\n            step = None\n        return slice(start, stop, step)\n\n    if isinstance(b, list):\n        return [fuse_slice(a, bb) for bb in b]\n    if isinstance(a, list) and isinstance(b, (int, slice)):\n        return a[b]\n\n    if isinstance(a, tuple) and not isinstance(b, tuple):\n        b = (b,)\n\n    # If given two tuples walk through both, being mindful of uneven sizes\n    # and newaxes\n    if isinstance(a, tuple) and isinstance(b, tuple):\n\n        if (any(isinstance(item, list) for item in a) and\n            any(isinstance(item, list) for item in b)):\n            raise NotImplementedError(\"Can't handle multiple list indexing\")\n\n        j = 0\n        result = list()\n        for i in range(len(a)):\n            #  axis ceased to exist  or we're out of b\n            if isinstance(a[i], int) or j == len(b):\n                result.append(a[i])\n                continue\n            while b[j] is None:  # insert any Nones on the rhs\n                result.append(None)\n                j += 1\n            result.append(fuse_slice(a[i], b[j]))  # Common case\n            j += 1\n        while j < len(b):  # anything leftover on the right?\n            result.append(b[j])\n            j += 1\n        return tuple(result)\n    raise NotImplementedError()\n", "path": "dask/array/optimization.py"}], "after_files": [{"content": "from __future__ import absolute_import, division, print_function\n\nfrom operator import getitem\n\nimport numpy as np\n\nfrom .core import getarray\nfrom ..core import flatten\nfrom ..optimize import cull, fuse, inline_functions\n\n\ndef optimize(dsk, keys, **kwargs):\n    \"\"\" Optimize dask for array computation\n\n    1.  Cull tasks not necessary to evaluate keys\n    2.  Remove full slicing, e.g. x[:]\n    3.  Inline fast functions like getitem and np.transpose\n    \"\"\"\n    keys = list(flatten(keys))\n    fast_functions = kwargs.get('fast_functions',\n                                set([getarray, np.transpose]))\n    dsk2, dependencies = cull(dsk, keys)\n    dsk4, dependencies = fuse(dsk2, keys, dependencies)\n    dsk5 = optimize_slices(dsk4)\n    dsk6 = inline_functions(dsk5, keys, fast_functions=fast_functions,\n                            dependencies=dependencies)\n    return dsk6\n\n\ndef optimize_slices(dsk):\n    \"\"\" Optimize slices\n\n    1.  Fuse repeated slices, like x[5:][2:6] -> x[7:11]\n    2.  Remove full slices, like         x[:] -> x\n\n    See also:\n        fuse_slice_dict\n    \"\"\"\n    getters = (getarray, getitem)\n    dsk = dsk.copy()\n    for k, v in dsk.items():\n        if type(v) is tuple:\n            if v[0] in getters:\n                try:\n                    func, a, a_index = v\n                    use_getarray = func is getarray\n                except ValueError:  # has four elements, includes a lock\n                    continue\n                while type(a) is tuple and a[0] in getters:\n                    try:\n                        f2, b, b_index = a\n                        use_getarray |= f2 is getarray\n                    except ValueError:  # has four elements, includes a lock\n                        break\n                    if (type(a_index) is tuple) != (type(b_index) is tuple):\n                        break\n                    if ((type(a_index) is tuple) and\n                            (len(a_index) != len(b_index)) and\n                            any(i is None for i in b_index + a_index)):\n                        break\n                    try:\n                        c_index = fuse_slice(b_index, a_index)\n                    except NotImplementedError:\n                        break\n                    (a, a_index) = (b, c_index)\n                if use_getarray:\n                    dsk[k] = (getarray, a, a_index)\n                elif (type(a_index) is slice and\n                      not a_index.start and\n                      a_index.stop is None and\n                      a_index.step is None):\n                    dsk[k] = a\n                elif type(a_index) is tuple and all(type(s) is slice and\n                                                    not s.start and\n                                                    s.stop is None and\n                                                    s.step is None\n                                                    for s in a_index):\n                    dsk[k] = a\n                else:\n                    dsk[k] = (getitem, a, a_index)\n    return dsk\n\n\ndef normalize_slice(s):\n    \"\"\" Replace Nones in slices with integers\n\n    >>> normalize_slice(slice(None, None, None))\n    slice(0, None, 1)\n    \"\"\"\n    start, stop, step = s.start, s.stop, s.step\n    if start is None:\n        start = 0\n    if step is None:\n        step = 1\n    if start < 0 or step < 0 or stop is not None and stop < 0:\n        raise NotImplementedError()\n    return slice(start, stop, step)\n\n\ndef fuse_slice(a, b):\n    \"\"\" Fuse stacked slices together\n\n    Fuse a pair of repeated slices into a single slice:\n\n    >>> fuse_slice(slice(1000, 2000), slice(10, 15))\n    slice(1010, 1015, None)\n\n    This also works for tuples of slices\n\n    >>> fuse_slice((slice(100, 200), slice(100, 200, 10)),\n    ...            (slice(10, 15), [5, 2]))\n    (slice(110, 115, None), [150, 120])\n\n    And a variety of other interesting cases\n\n    >>> fuse_slice(slice(1000, 2000), 10)  # integers\n    1010\n\n    >>> fuse_slice(slice(1000, 2000, 5), slice(10, 20, 2))\n    slice(1050, 1100, 10)\n\n    >>> fuse_slice(slice(1000, 2000, 5), [1, 2, 3])  # lists\n    [1005, 1010, 1015]\n\n    >>> fuse_slice(None, slice(None, None))  # doctest: +SKIP\n    None\n    \"\"\"\n    # None only works if the second side is a full slice\n    if a is None and b == slice(None, None):\n        return None\n\n    # Replace None with 0 and one in start and step\n    if isinstance(a, slice):\n        a = normalize_slice(a)\n    if isinstance(b, slice):\n        b = normalize_slice(b)\n\n    if isinstance(a, slice) and isinstance(b, int):\n        if b < 0:\n            raise NotImplementedError()\n        return a.start + b*a.step\n\n    if isinstance(a, slice) and isinstance(b, slice):\n        start = a.start + a.step * b.start\n        if b.stop is not None:\n            stop = a.start + a.step * b.stop\n        else:\n            stop = None\n        if a.stop is not None:\n            if stop is not None:\n                stop = min(a.stop, stop)\n            else:\n                stop = a.stop\n            stop = stop\n        step = a.step * b.step\n        if step == 1:\n            step = None\n        return slice(start, stop, step)\n\n    if isinstance(b, list):\n        return [fuse_slice(a, bb) for bb in b]\n    if isinstance(a, list) and isinstance(b, (int, slice)):\n        return a[b]\n\n    if isinstance(a, tuple) and not isinstance(b, tuple):\n        b = (b,)\n\n    # If given two tuples walk through both, being mindful of uneven sizes\n    # and newaxes\n    if isinstance(a, tuple) and isinstance(b, tuple):\n\n        if (any(isinstance(item, list) for item in a) and\n                any(isinstance(item, list) for item in b)):\n            raise NotImplementedError(\"Can't handle multiple list indexing\")\n\n        j = 0\n        result = list()\n        for i in range(len(a)):\n            #  axis ceased to exist  or we're out of b\n            if isinstance(a[i], int) or j == len(b):\n                result.append(a[i])\n                continue\n            while b[j] is None:  # insert any Nones on the rhs\n                result.append(None)\n                j += 1\n            result.append(fuse_slice(a[i], b[j]))  # Common case\n            j += 1\n        while j < len(b):  # anything leftover on the right?\n            result.append(b[j])\n            j += 1\n        return tuple(result)\n    raise NotImplementedError()\n", "path": "dask/array/optimization.py"}]}